Files
EvoScientist/tests/test_update_check.py
T

577 lines
22 KiB
Python

"""Tests for EvoScientist.update_check module."""
import json
import time
from unittest.mock import MagicMock, patch
import pytest
from EvoScientist import update_check
from EvoScientist.update_check import (
CACHE_TTL,
_parse_version,
get_latest_version,
is_update_available,
)
class TestParseVersion:
"""Tests for _parse_version."""
def test_basic(self):
assert _parse_version("1.2.3") == (1, 2, 3)
def test_single_digit(self):
assert _parse_version("5") == (5,)
def test_whitespace(self):
assert _parse_version(" 0.0.2 ") == (0, 0, 2)
def test_comparison(self):
assert _parse_version("0.0.3") > _parse_version("0.0.2")
assert _parse_version("0.1.0") > _parse_version("0.0.9")
assert _parse_version("1.0.0") > _parse_version("0.9.9")
def test_equal(self):
assert _parse_version("0.0.2") == _parse_version("0.0.2")
class TestGetLatestVersion:
"""Tests for get_latest_version."""
def test_fresh_cache_hit(self, tmp_path):
cache_file = tmp_path / "latest_version.json"
cache_file.write_text(
json.dumps({"version": "1.0.0", "checked_at": time.time()})
)
with patch("EvoScientist.update_check.CACHE_FILE", cache_file):
assert get_latest_version() == "1.0.0"
def test_stale_cache_fetches_pypi(self, tmp_path):
cache_file = tmp_path / "latest_version.json"
cache_file.write_text(
json.dumps(
{
"version": "0.0.1",
"checked_at": time.time() - CACHE_TTL - 1,
}
)
)
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps(
{"info": {"version": "2.0.0"}}
).encode()
mock_resp.__enter__ = lambda s: s
mock_resp.__exit__ = MagicMock(return_value=False)
with (
patch("EvoScientist.update_check.CACHE_FILE", cache_file),
patch("EvoScientist.update_check.CACHE_DIR", tmp_path),
patch("urllib.request.urlopen", return_value=mock_resp),
):
assert get_latest_version() == "2.0.0"
def test_no_cache_fetches_pypi(self, tmp_path):
cache_file = tmp_path / "latest_version.json"
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps(
{"info": {"version": "3.0.0"}}
).encode()
mock_resp.__enter__ = lambda s: s
mock_resp.__exit__ = MagicMock(return_value=False)
with (
patch("EvoScientist.update_check.CACHE_FILE", cache_file),
patch("EvoScientist.update_check.CACHE_DIR", tmp_path),
patch("urllib.request.urlopen", return_value=mock_resp),
):
assert get_latest_version() == "3.0.0"
# Cache should have been written
assert cache_file.exists()
data = json.loads(cache_file.read_text())
assert data["version"] == "3.0.0"
def test_network_error_returns_none(self, tmp_path):
cache_file = tmp_path / "latest_version.json"
with (
patch("EvoScientist.update_check.CACHE_FILE", cache_file),
patch("urllib.request.urlopen", side_effect=OSError("network down")),
):
assert get_latest_version() is None
def test_corrupt_cache_recovers(self, tmp_path):
cache_file = tmp_path / "latest_version.json"
cache_file.write_text("not valid json!!!")
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps(
{"info": {"version": "1.5.0"}}
).encode()
mock_resp.__enter__ = lambda s: s
mock_resp.__exit__ = MagicMock(return_value=False)
with (
patch("EvoScientist.update_check.CACHE_FILE", cache_file),
patch("EvoScientist.update_check.CACHE_DIR", tmp_path),
patch("urllib.request.urlopen", return_value=mock_resp),
):
assert get_latest_version() == "1.5.0"
class TestIsUpdateAvailable:
"""Tests for is_update_available."""
def test_newer_version_available(self):
with (
patch("EvoScientist.update_check.get_latest_version", return_value="9.9.9"),
patch("EvoScientist.update_check._installed_version", return_value="0.0.2"),
):
available, latest = is_update_available()
assert available is True
assert latest == "9.9.9"
def test_same_version(self):
with (
patch("EvoScientist.update_check.get_latest_version", return_value="0.0.2"),
patch("EvoScientist.update_check._installed_version", return_value="0.0.2"),
):
available, _latest = is_update_available()
assert available is False
def test_older_pypi_version(self):
with (
patch("EvoScientist.update_check.get_latest_version", return_value="0.0.1"),
patch("EvoScientist.update_check._installed_version", return_value="0.0.2"),
):
available, _latest = is_update_available()
assert available is False
def test_pypi_unreachable(self):
with patch("EvoScientist.update_check.get_latest_version", return_value=None):
available, latest = is_update_available()
assert available is False
assert latest is None
def test_invalid_version_string(self):
with (
patch("EvoScientist.update_check.get_latest_version", return_value="abc"),
patch("EvoScientist.update_check._installed_version", return_value="0.0.2"),
):
available, _latest = is_update_available()
assert available is False
@pytest.fixture(autouse=True)
def _reset_gitea_cache(monkeypatch):
monkeypatch.setattr(update_check, "_UPDATE_CACHE", {"info": None, "fetched_at": 0.0})
for var in (
"EVOSCIENTIST_UPDATE_BASE_URL",
"EVOSCIENTIST_UPDATE_REPO",
"EVOSCIENTIST_UPDATE_TOKEN",
"EVOSCIENTIST_UPDATE_CHECK_DISABLED",
):
monkeypatch.delenv(var, raising=False)
yield
def _release(tag, **kw):
r = {
"tag_name": tag,
"html_url": f"https://git.foksai.com/ouyangbo/EvoScientist/releases/tag/{tag}",
"body": f"notes for {tag}",
"published_at": "2026-08-01T00:00:00Z",
"assets": [],
"tarball_url": f"https://git.foksai.com/api/v1/repos/ouyangbo/EvoScientist/archive/{tag}.tar.gz",
}
r.update(kw)
return r
class TestSemverKey:
def test_ordering(self):
assert update_check._semver_key("v0.1.19") < update_check._semver_key("0.2.2")
assert update_check._semver_key("1.0.0") > update_check._semver_key("0.9.9")
def test_equal(self):
assert update_check._semver_key("v0.2.2") == update_check._semver_key("0.2.2")
def test_malformed_segments_are_zero(self):
assert update_check._semver_key("0.2.x") == update_check._semver_key("0.2.0")
assert update_check._semver_key("garbage") == (0, 0, 0)
def test_short_versions_pad(self):
assert update_check._semver_key("1.2") == (1, 2, 0)
class TestGetUpdateInfo:
def _patch_fetch(self, monkeypatch, releases, tags):
def fake_get(url, **kw):
if "/releases" in url:
return releases
if "/tags" in url:
return [{"name": t} for t in tags]
raise AssertionError(f"unexpected url {url}")
monkeypatch.setattr(update_check, "_http_get_json", fake_get)
def test_latest_from_release(self, monkeypatch):
self._patch_fetch(monkeypatch, [_release("v9.9.9")], ["v0.1.0"])
with patch("EvoScientist.update_check._installed_version", return_value="0.2.2"):
info = update_check.get_update_info()
assert info["latest_version"] == "9.9.9"
assert info["has_update"] is True
assert info["release_url"].endswith("/v9.9.9")
assert info["release_notes"] == "notes for v9.9.9"
assert info["warning"] is None
assert info["cached"] is False
def test_tag_newer_than_release_wins(self, monkeypatch):
# Gitea orders releases/latest by tag creation date, not semver;
# the checker must take the max across releases AND tags.
self._patch_fetch(monkeypatch, [_release("v0.1.19")], ["v0.1.20", "v0.2.2"])
with patch("EvoScientist.update_check._installed_version", return_value="0.2.1"):
info = update_check.get_update_info()
assert info["latest_version"] == "0.2.2"
assert info["has_update"] is True
# tag-only version has no release metadata
assert info["release_url"] is None
def test_no_releases_no_tags(self, monkeypatch):
self._patch_fetch(monkeypatch, [], [])
with patch("EvoScientist.update_check._installed_version", return_value="0.2.2"):
info = update_check.get_update_info()
assert info["latest_version"] == "0.2.2"
assert info["has_update"] is False
def test_up_to_date(self, monkeypatch):
self._patch_fetch(monkeypatch, [_release("v0.2.2")], [])
with patch("EvoScientist.update_check._installed_version", return_value="0.2.2"):
info = update_check.get_update_info()
assert info["has_update"] is False
def test_cache_within_ttl(self, monkeypatch):
calls = []
def fake_get(url, **kw):
calls.append(url)
return [_release("v9.9.9")] if "/releases" in url else []
monkeypatch.setattr(update_check, "_http_get_json", fake_get)
with patch("EvoScientist.update_check._installed_version", return_value="0.2.2"):
first = update_check.get_update_info()
second = update_check.get_update_info()
assert len(calls) == 2 # releases + tags, once
assert first["cached"] is False
assert second["cached"] is True
def test_force_bypasses_cache(self, monkeypatch):
calls = []
def fake_get(url, **kw):
calls.append(url)
return [_release("v9.9.9")] if "/releases" in url else []
monkeypatch.setattr(update_check, "_http_get_json", fake_get)
with patch("EvoScientist.update_check._installed_version", return_value="0.2.2"):
update_check.get_update_info()
forced = update_check.get_update_info(force=True)
assert len(calls) == 4
assert forced["cached"] is False
def test_network_error_without_cache(self, monkeypatch):
def boom(url, **kw):
raise OSError("network down")
monkeypatch.setattr(update_check, "_http_get_json", boom)
with patch("EvoScientist.update_check._installed_version", return_value="0.2.2"):
info = update_check.get_update_info()
assert info["has_update"] is False
assert info["latest_version"] == "0.2.2"
assert info["warning"]
def test_network_error_serves_stale_cache(self, monkeypatch):
self._patch_fetch(monkeypatch, [_release("v9.9.9")], [])
with patch("EvoScientist.update_check._installed_version", return_value="0.2.2"):
update_check.get_update_info()
def boom(url, **kw):
raise OSError("network down")
monkeypatch.setattr(update_check, "_http_get_json", boom)
monkeypatch.setattr(
update_check,
"_UPDATE_CACHE",
{**update_check._UPDATE_CACHE, "fetched_at": time.time() - 10_000},
)
with patch("EvoScientist.update_check._installed_version", return_value="0.2.2"):
info = update_check.get_update_info()
assert info["latest_version"] == "9.9.9"
assert info["cached"] is True
assert info["warning"]
def test_disabled_env_short_circuits(self, monkeypatch):
monkeypatch.setenv("EVOSCIENTIST_UPDATE_CHECK_DISABLED", "1")
def boom(url, **kw):
raise AssertionError("network must not be touched")
monkeypatch.setattr(update_check, "_http_get_json", boom)
with patch("EvoScientist.update_check._installed_version", return_value="0.2.2"):
info = update_check.get_update_info()
assert info["latest_version"] == "0.2.2"
assert info["has_update"] is False
def _asset(name, url=None):
return {
"name": name,
"browser_download_url": url
or f"https://git.foksai.com:8443/attachments/{name}-uuid",
}
class TestDownloadUpdate:
def _patch(self, monkeypatch, tmp_path, releases, files=None):
monkeypatch.setenv("EVOSCIENTIST_UPDATE_STAGING_DIR", str(tmp_path))
monkeypatch.setattr(
update_check, "_http_get_json",
lambda url, **kw: releases if "/releases" in url else [],
)
written = {}
def fake_download(url, dest, *, max_bytes, timeout=120.0):
data = (files or {}).get(url, b"payload-" + url.encode()[:8])
if len(data) > max_bytes:
raise update_check.UpdateDownloadError("too large")
dest.write_bytes(data)
return dest
monkeypatch.setattr(update_check, "_http_download", fake_download)
return written
def test_prefers_wheel_over_sdist(self, monkeypatch, tmp_path):
rel = _release(
"v9.9.9",
assets=[
_asset("EvoScientist-9.9.9.tar.gz"),
_asset("evoscientist-9.9.9-py3-none-any.whl"),
],
)
self._patch(monkeypatch, tmp_path, [rel])
result = update_check.download_update()
assert result["version"] == "9.9.9"
assert result["file"].endswith(".whl")
assert "v9.9.9" in result["path"]
assert "uv pip install" in result["suggested_command"]
def test_sdist_when_no_wheel(self, monkeypatch, tmp_path):
rel = _release("v9.9.9", assets=[_asset("EvoScientist-9.9.9.tar.gz")])
self._patch(monkeypatch, tmp_path, [rel])
result = update_check.download_update()
assert result["file"].endswith(".tar.gz")
def test_tarball_fallback_when_no_assets(self, monkeypatch, tmp_path):
rel = _release("v9.9.9", assets=[])
self._patch(monkeypatch, tmp_path, [rel])
result = update_check.download_update()
assert result["file"].endswith(".tar.gz")
def test_specific_version_uses_tag_endpoint(self, monkeypatch, tmp_path):
monkeypatch.setenv("EVOSCIENTIST_UPDATE_STAGING_DIR", str(tmp_path))
seen = []
def fake_get(url, **kw):
seen.append(url)
if url.endswith("/releases/tags/v0.2.2"):
return _release("v0.2.2", assets=[_asset("evoscientist-0.2.2-py3-none-any.whl")])
return []
monkeypatch.setattr(update_check, "_http_get_json", fake_get)
monkeypatch.setattr(
update_check, "_http_download",
lambda url, dest, **kw: dest.write_bytes(b"x") or dest,
)
result = update_check.download_update("0.2.2")
assert result["version"] == "0.2.2"
assert any("/releases/tags/v0.2.2" in u for u in seen)
def test_rejects_foreign_host(self, monkeypatch, tmp_path):
rel = _release(
"v9.9.9",
assets=[_asset("evil.whl", url="https://evil.com/payload.whl")],
)
self._patch(monkeypatch, tmp_path, [rel])
with pytest.raises(update_check.UpdateDownloadError, match="host"):
update_check.download_update()
def test_size_cap_aborts_and_removes_file(self, monkeypatch, tmp_path):
rel = _release("v9.9.9", assets=[_asset("big-py3-none-any.whl")])
self._patch(monkeypatch, tmp_path, [rel], files={
"https://git.foksai.com:8443/attachments/big-py3-none-any.whl-uuid": b"x" * 10,
})
monkeypatch.setattr(update_check, "MAX_DOWNLOAD_BYTES", 4)
with pytest.raises(update_check.UpdateDownloadError):
update_check.download_update()
target = tmp_path / "v9.9.9"
assert not any(target.glob("*.whl"))
def test_checksum_mismatch_deletes_file(self, monkeypatch, tmp_path):
import hashlib
wheel_url = "https://git.foksai.com:8443/attachments/evoscientist-9.9.9-py3-none-any.whl-uuid"
sums_url = "https://git.foksai.com:8443/attachments/checksums.txt-uuid"
rel = _release(
"v9.9.9",
assets=[
_asset("evoscientist-9.9.9-py3-none-any.whl"),
_asset("checksums.txt"),
],
)
bad = hashlib.sha256(b"different").hexdigest()
self._patch(monkeypatch, tmp_path, [rel], files={
wheel_url: b"wheel-bytes",
sums_url: f"{bad} evoscientist-9.9.9-py3-none-any.whl\n".encode(),
})
with pytest.raises(update_check.UpdateDownloadError, match="checksum"):
update_check.download_update()
target = tmp_path / "v9.9.9"
assert not any(target.glob("*.whl"))
def test_checksum_ok_when_matching(self, monkeypatch, tmp_path):
import hashlib
wheel_url = "https://git.foksai.com:8443/attachments/evoscientist-9.9.9-py3-none-any.whl-uuid"
sums_url = "https://git.foksai.com:8443/attachments/checksums.txt-uuid"
rel = _release(
"v9.9.9",
assets=[
_asset("evoscientist-9.9.9-py3-none-any.whl"),
_asset("checksums.txt"),
],
)
good = hashlib.sha256(b"wheel-bytes").hexdigest()
self._patch(monkeypatch, tmp_path, [rel], files={
wheel_url: b"wheel-bytes",
sums_url: f"{good} evoscientist-9.9.9-py3-none-any.whl\n".encode(),
})
result = update_check.download_update()
assert (tmp_path / "v9.9.9" / result["file"]).exists()
def test_unknown_version_raises(self, monkeypatch, tmp_path):
monkeypatch.setenv("EVOSCIENTIST_UPDATE_STAGING_DIR", str(tmp_path))
monkeypatch.setattr(update_check, "_http_get_json", lambda url, **kw: None)
with pytest.raises(update_check.UpdateDownloadError):
update_check.download_update("1.2.3")
class TestStagingDir:
def test_updates_dir_defaults_to_config_dir(self, monkeypatch, tmp_path):
monkeypatch.delenv("EVOSCIENTIST_UPDATE_STAGING_DIR", raising=False)
monkeypatch.delenv("EVOSCIENTIST_WORKSPACE_DIR", raising=False)
monkeypatch.setattr(update_check, "get_config_dir", lambda: tmp_path)
assert update_check._updates_dir("1.2.3") == tmp_path / "updates" / "v1.2.3"
def test_updates_dir_env_override(self, monkeypatch, tmp_path):
monkeypatch.setenv("EVOSCIENTIST_UPDATE_STAGING_DIR", str(tmp_path / "staging"))
assert update_check._updates_dir("1.2.3") == tmp_path / "staging" / "v1.2.3"
def test_download_update_reuses_verified_local_artifact(self, monkeypatch, tmp_path):
monkeypatch.setenv("EVOSCIENTIST_UPDATE_STAGING_DIR", str(tmp_path))
version, wheel = "9.9.9", "EvoScientist-9.9.9-py3-none-any.whl"
staged = tmp_path / "v9.9.9"
staged.mkdir(parents=True)
payload = b"fake wheel bytes"
(staged / wheel).write_bytes(payload)
import hashlib
digest = hashlib.sha256(payload).hexdigest()
(staged / "checksums.txt").write_text(f"{digest} {wheel}\n", encoding="utf-8")
def _boom(*a, **k): # any network call fails the test
raise AssertionError("network must not be touched")
monkeypatch.setattr(update_check, "_http_download", _boom)
monkeypatch.setattr(
update_check, "_resolve_release", lambda v: (v or "9.9.9", {"assets": []})
)
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