From 91e2a87be5d05d7e14c6495dbe6072e7d695d6d5 Mon Sep 17 00:00:00 2001 From: m4 Date: Wed, 12 Aug 2026 16:34:16 +0800 Subject: [PATCH] feat(update): rollback version list with BREAKING-DB truncation --- EvoScientist/update_check.py | 65 ++++++++++++++++++++++++++++++ tests/test_update_check.py | 78 ++++++++++++++++++++++++++++++++++++ 2 files changed, 143 insertions(+) diff --git a/EvoScientist/update_check.py b/EvoScientist/update_check.py index c4c3c33..b65777a 100644 --- a/EvoScientist/update_check.py +++ b/EvoScientist/update_check.py @@ -126,6 +126,7 @@ class UpdateInfo(TypedDict): published_at: str | None cached: bool warning: str | None + breaking_db: bool def _semver_key(v: str) -> tuple[int, int, int]: @@ -199,6 +200,12 @@ def _fetch_latest() -> tuple[str, dict | None]: return tag.lstrip("vV"), release +def _is_breaking_release(release: dict | None) -> bool: + if not release: + return False + return bool(release.get("prerelease")) and "BREAKING-DB" in (release.get("body") or "") + + def get_update_info(*, force: bool = False) -> UpdateInfo: """Resolve current vs latest published version, with a 20-minute cache.""" current = _current_version() @@ -213,6 +220,7 @@ def get_update_info(*, force: bool = False) -> UpdateInfo: published_at=None, cached=cached, warning=warning, + breaking_db=False, ) if os.environ.get("EVOSCIENTIST_UPDATE_CHECK_DISABLED") == "1": @@ -251,12 +259,69 @@ def get_update_info(*, force: bool = False) -> UpdateInfo: published_at=release.get("published_at") if release else None, cached=False, warning=None, + breaking_db=_is_breaking_release(release), ) _UPDATE_CACHE["info"] = info _UPDATE_CACHE["fetched_at"] = now return info +# --------------------------------------------------------------------------- +# Rollback version list +# --------------------------------------------------------------------------- + + +class RollbackVersion(TypedDict): + version: str + published_at: str | None + release_url: str | None + + +def list_rollback_versions(limit: int = 3) -> list[RollbackVersion]: + """Non-draft, non-prerelease releases strictly older than the current version. + + Stops collecting at a BREAKING-DB marker release: rollback past a breaking + schema change is not offered (and the breaking release itself, being a + prerelease, is never listed). + """ + base, repo = _gitea_base(), _gitea_repo() + releases = _http_get_json(f"{base}/api/v1/repos/{repo}/releases?limit=20") or [] + current = _current_version() + out: list[RollbackVersion] = [] + seen: set[str] = set() + ordered = sorted( + (r for r in releases if r.get("tag_name")), + key=lambda r: _semver_key(r["tag_name"]), + reverse=True, + ) + for r in ordered: + if r.get("draft"): + continue + if r.get("prerelease"): + if "BREAKING-DB" in (r.get("body") or ""): + break # breaking point: nothing older is rollback-safe + continue + v = r["tag_name"].lstrip("vV") + if v in seen or _semver_key(v) >= _semver_key(current): + continue + seen.add(v) + out.append( + RollbackVersion( + version=v, + published_at=r.get("published_at"), + release_url=r.get("html_url"), + ) + ) + if len(out) >= limit: + break + return out + + +def is_allowed_rollback(version: str) -> bool: + target = version.strip().lstrip("vV") + return any(v["version"] == target for v in list_rollback_versions()) + + # --------------------------------------------------------------------------- # Update download (powers POST /internal/system/version/download) # --------------------------------------------------------------------------- diff --git a/tests/test_update_check.py b/tests/test_update_check.py index b7245fe..7d46fe4 100644 --- a/tests/test_update_check.py +++ b/tests/test_update_check.py @@ -496,3 +496,81 @@ class TestStagingDir: result = update_check.download_update("9.9.9") assert result["file"] == wheel assert result["path"] == str(staged / wheel) + + +def _rel(tag, body="", prerelease=False, draft=False): + return _release( + tag, + body=body, + prerelease=prerelease, + draft=draft, + ) + + +class TestRollbackVersions: + def test_filters_and_orders(self, monkeypatch): + releases = [ + _rel("v0.3.0"), + _rel("v0.2.9", prerelease=True), # excluded + _rel("v0.2.8"), + _rel("v0.2.7", draft=True), # excluded + _rel("v0.2.6"), + _rel("v0.2.5"), + ] + monkeypatch.setattr(update_check, "_http_get_json", lambda url, **k: releases) + monkeypatch.setattr(update_check, "_current_version", lambda: "0.3.1") + versions = [v["version"] for v in update_check.list_rollback_versions()] + assert versions == ["0.3.0", "0.2.8", "0.2.6"] + + def test_excludes_current_and_newer(self, monkeypatch): + releases = [_rel("v0.3.1"), _rel("v0.3.2"), _rel("v0.3.0")] + monkeypatch.setattr(update_check, "_http_get_json", lambda url, **k: releases) + monkeypatch.setattr(update_check, "_current_version", lambda: "0.3.1") + assert [v["version"] for v in update_check.list_rollback_versions()] == ["0.3.0"] + + def test_truncates_at_breaking_db(self, monkeypatch): + releases = [_rel("v0.3.2"), _rel("v0.3.0", body="BREAKING-DB", prerelease=True), _rel("v0.2.9")] + monkeypatch.setattr(update_check, "_http_get_json", lambda url, **k: releases) + monkeypatch.setattr(update_check, "_current_version", lambda: "0.3.3") + assert [v["version"] for v in update_check.list_rollback_versions()] == ["0.3.2"] + + def test_is_allowed_rollback(self, monkeypatch): + monkeypatch.setattr(update_check, "_http_get_json", lambda url, **k: [_rel("v0.3.0")]) + monkeypatch.setattr(update_check, "_current_version", lambda: "0.3.1") + assert update_check.is_allowed_rollback("0.3.0") + assert update_check.is_allowed_rollback("v0.3.0") + assert not update_check.is_allowed_rollback("0.2.0") + + +class TestBreakingDbFlag: + def test_update_info_breaking_db_flag(self, monkeypatch): + monkeypatch.setattr( + update_check, + "_fetch_latest", + lambda: ( + "0.9.9", + { + "body": "x BREAKING-DB y", + "prerelease": True, + "html_url": "u", + "published_at": "p", + }, + ), + ) + monkeypatch.setenv("EVOSCIENTIST_UPDATE_CHECK_DISABLED", "0") + update_check._UPDATE_CACHE["info"] = None + info = update_check.get_update_info(force=True) + assert info["breaking_db"] is True + + def test_update_info_not_breaking_for_regular_release(self, monkeypatch): + monkeypatch.setattr( + update_check, + "_fetch_latest", + lambda: ( + "0.9.9", + {"body": "normal", "prerelease": False, "html_url": "u", "published_at": "p"}, + ), + ) + update_check._UPDATE_CACHE["info"] = None + info = update_check.get_update_info(force=True) + assert info["breaking_db"] is False