ouyangbo/EvoScientist
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 3 Wiki Activity

feat(update): stage artifacts under config dir and reuse verified local files

Browse Source
This commit is contained in:
m4
2026-08-12 16:21:45 +08:00
parent 194402fc88
commit 174f03b92d
2 changed files with 696 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
EvoScientist/update_check.py
+355
View File
@@ -3,13 +3,28 @@
Compares the installed version against PyPI and caches the result
(see ``CACHE_TTL``). All errors are silently swallowed so startup
is never blocked or degraded.
Also exposes a Gitea-based checker (``get_update_info``) used by the
``/internal/system/version`` HTTP route. It resolves "latest" as the
max semver across the instance's releases and tags — Gitea orders
``releases/latest`` by tag creation date, not semver, so a single
endpoint cannot be trusted.
``download_update`` stages a release artifact under the update staging
dir (``~/.evoscientist/updates/`` by default) and returns the suggested
install command; it never applies the update itself.
"""
from __future__ import annotations
import hashlib
import json
import logging
import os
import time
from pathlib import Path
from typing import Any, TypedDict
from urllib.parse import urlparse
from .config.settings import get_config_dir
@@ -91,3 +106,343 @@ def is_update_available() -> tuple[bool, str | None]:
logger.debug("Failed to compare versions", exc_info=True)
return False, None
# ---------------------------------------------------------------------------
# Gitea-based checker (powers /internal/system/version)
# ---------------------------------------------------------------------------
GITEA_CACHE_TTL = 1_200 # 20 minutes
_UPDATE_CACHE: dict[str, Any] = {"info": None, "fetched_at": 0.0}
class UpdateInfo(TypedDict):
current_version: str
latest_version: str
has_update: bool
release_url: str | None
release_notes: str | None
published_at: str | None
cached: bool
warning: str | None
def _semver_key(v: str) -> tuple[int, int, int]:
"""Map a version string to a comparable 3-tuple; junk segments count as 0."""
parts = v.strip().lstrip("vV").split(".")[:3]
out = []
for p in parts:
try:
out.append(int(p))
except ValueError:
out.append(0)
while len(out) < 3:
out.append(0)
return tuple(out) # type: ignore[return-value]
def _gitea_base() -> str:
return os.environ.get("EVOSCIENTIST_UPDATE_BASE_URL", "https://git.foksai.com").rstrip("/")
def _gitea_repo() -> str:
return os.environ.get("EVOSCIENTIST_UPDATE_REPO", "ouyangbo/EvoScientist")
def _http_get_json(url: str, *, timeout: float = 10.0) -> Any:
import httpx
headers = {"User-Agent": "EvoScientist update-check"}
token = os.environ.get("EVOSCIENTIST_UPDATE_TOKEN")
if token:
headers["Authorization"] = f"token {token}"
resp = httpx.get(url, headers=headers, timeout=timeout, follow_redirects=True)
if resp.status_code == 404:
return None
resp.raise_for_status()
return resp.json()
def _current_version() -> str:
try:
return _installed_version()
except Exception:
return "0.0.0-dev"
def _fetch_latest() -> tuple[str, dict | None]:
"""Return (latest_version, release_dict_or_None) as max semver over releases+tags."""
base, repo = _gitea_base(), _gitea_repo()
releases = _http_get_json(f"{base}/api/v1/repos/{repo}/releases?limit=20") or []
tags = _http_get_json(f"{base}/api/v1/repos/{repo}/tags?limit=20") or []
candidates: list[tuple[tuple[int, int, int], str, dict | None]] = []
for r in releases:
tag = r.get("tag_name", "")
if tag:
candidates.append((_semver_key(tag), tag, r))
for t in tags:
name = t.get("name", "")
if name:
candidates.append((_semver_key(name), name, None))
if not candidates:
return _current_version(), None
candidates.sort(key=lambda c: c[0])
_key, tag, release = candidates[-1]
if release is None:
# the newest candidate may still have a release lower in the list;
# prefer its metadata only when it matches the winning tag
release = next((r for k, tg, r in candidates if r and tg == tag), None)
return tag.lstrip("vV"), release
def get_update_info(*, force: bool = False) -> UpdateInfo:
"""Resolve current vs latest published version, with a 20-minute cache."""
current = _current_version()
def stale(latest: str | None = None, warning: str | None = None, cached: bool = False) -> UpdateInfo:
return UpdateInfo(
current_version=current,
latest_version=latest or current,
has_update=False,
release_url=None,
release_notes=None,
published_at=None,
cached=cached,
warning=warning,
)
if os.environ.get("EVOSCIENTIST_UPDATE_CHECK_DISABLED") == "1":
return stale()
now = time.time()
cached_info: UpdateInfo | None = _UPDATE_CACHE["info"]
if cached_info is not None and not force:
age = now - _UPDATE_CACHE["fetched_at"]
if age < GITEA_CACHE_TTL:
return UpdateInfo(
**{**cached_info, "cached": True, "current_version": current}
)
try:
latest, release = _fetch_latest()
except Exception as exc: # network/HTTP failure: serve stale cache if any
logger.debug("Gitea update check failed", exc_info=True)
if cached_info is not None:
return UpdateInfo(
**{
**cached_info,
"cached": True,
"current_version": current,
"warning": f"update check failed: {exc}",
}
)
return stale(warning=f"update check failed: {exc}")
info = UpdateInfo(
current_version=current,
latest_version=latest,
has_update=_semver_key(latest) > _semver_key(current),
release_url=release.get("html_url") if release else None,
release_notes=release.get("body") if release else None,
published_at=release.get("published_at") if release else None,
cached=False,
warning=None,
)
_UPDATE_CACHE["info"] = info
_UPDATE_CACHE["fetched_at"] = now
return info
# ---------------------------------------------------------------------------
# Update download (powers POST /internal/system/version/download)
# ---------------------------------------------------------------------------
MAX_DOWNLOAD_BYTES = 200 * 1024 * 1024
class UpdateDownloadError(RuntimeError):
"""A user-safe download failure (message may reach the browser)."""
class DownloadResult(TypedDict):
version: str
file: str
path: str
suggested_command: str
def _staging_dir() -> Path:
override = os.environ.get("EVOSCIENTIST_UPDATE_STAGING_DIR", "").strip()
if override:
return Path(override)
return get_config_dir() / "updates"
def _updates_dir(version: str) -> Path:
return _staging_dir() / f"v{version}"
def _http_download(
url: str, dest: Path, *, max_bytes: int, timeout: float = 120.0
) -> Path:
import httpx
headers = {"User-Agent": "EvoScientist update-check"}
token = os.environ.get("EVOSCIENTIST_UPDATE_TOKEN")
if token:
headers["Authorization"] = f"token {token}"
with httpx.stream(
"GET", url, headers=headers, timeout=timeout, follow_redirects=True
) as resp:
resp.raise_for_status()
total = 0
with open(dest, "wb") as fh:
for chunk in resp.iter_bytes():
total += len(chunk)
if total > max_bytes:
raise UpdateDownloadError(
f"download exceeds the {max_bytes}-byte cap"
)
fh.write(chunk)
return dest
def _check_asset_url(url: str) -> None:
"""SSRF guard: only download from the configured Gitea host (any port)."""
if urlparse(url).hostname != urlparse(_gitea_base()).hostname:
raise UpdateDownloadError(f"download URL host is not allowed: {url!r}")
def _pick_asset(release: dict) -> tuple[str, str]:
"""Choose (filename, url): wheel > sdist > repo tarball."""
assets = release.get("assets") or []
wheels = [a for a in assets if str(a.get("name", "")).endswith(".whl")]
sdists = [a for a in assets if str(a.get("name", "")).endswith(".tar.gz")]
for asset in (*wheels, *sdists):
return asset["name"], asset["browser_download_url"]
tarball = release.get("tarball_url")
if tarball:
return f"{release.get('tag_name', 'release')}.tar.gz", tarball
raise UpdateDownloadError("release has no downloadable assets")
def _verify_local_checksum(sums_path: Path, filename: str, path: Path) -> None:
"""Verify path against the checksums.txt entry for filename (if listed)."""
expected = None
for line in sums_path.read_text(encoding="utf-8").splitlines():
parts = line.split()
if len(parts) == 2 and parts[1] == filename:
expected = parts[0]
break
if expected is None:
return # artifact not listed — nothing to verify against
actual = hashlib.sha256(path.read_bytes()).hexdigest()
if actual != expected:
raise UpdateDownloadError(f"checksum mismatch for {filename}")
def _verify_checksum(release: dict, updates: Path, filename: str, path: Path) -> None:
"""If the release ships checksums.txt with an entry for filename, verify it."""
sums = next(
(a for a in (release.get("assets") or []) if a.get("name") == "checksums.txt"),
None,
)
if sums is None:
return
_check_asset_url(sums["browser_download_url"])
sums_path = updates / "checksums.txt"
_http_download(sums["browser_download_url"], sums_path, max_bytes=MAX_DOWNLOAD_BYTES)
_verify_local_checksum(sums_path, filename, path)
def _find_verified_local_artifact(updates: Path) -> tuple[str, Path] | None:
"""Return (filename, path) of a staged artifact that passes checksums.txt."""
sums_path = updates / "checksums.txt"
if not sums_path.exists():
return None
for line in sums_path.read_text(encoding="utf-8").splitlines():
parts = line.split()
if len(parts) != 2:
continue
filename = parts[1]
if not (filename.endswith(".whl") or filename.endswith(".tar.gz")):
continue
dest = updates / filename
if not dest.exists():
continue
try:
_verify_local_checksum(sums_path, filename, dest)
except UpdateDownloadError:
dest.unlink(missing_ok=True) # corrupt staging — fall through to download
continue
return filename, dest
return None
def _resolve_release(version: str | None) -> tuple[str, dict]:
base, repo = _gitea_base(), _gitea_repo()
if version is not None:
tag = version if version.startswith(("v", "V")) else f"v{version}"
release = _http_get_json(f"{base}/api/v1/repos/{repo}/releases/tags/{tag}")
if not release:
raise UpdateDownloadError(f"no release found for version {version}")
return tag.lstrip("vV"), release
latest, release = _fetch_latest()
if release is None:
# newest candidate is a bare tag — fall back to its source archive
release = {
"tag_name": f"v{latest}",
"assets": [],
"tarball_url": f"{base}/api/v1/repos/{repo}/archive/v{latest}.tar.gz",
}
return latest, release
def download_update(version: str | None = None) -> DownloadResult:
"""Stage a release artifact locally; never installs it."""
try:
resolved_version, release = _resolve_release(version)
except UpdateDownloadError:
raise
except Exception as exc:
raise UpdateDownloadError(f"could not resolve release: {exc}") from exc
updates = _updates_dir(resolved_version)
updates.mkdir(parents=True, exist_ok=True)
reused = _find_verified_local_artifact(updates)
if reused is not None:
filename, dest = reused
return DownloadResult(
version=resolved_version,
file=filename,
path=str(dest),
suggested_command=f"uv pip install {dest} # then restart the backend",
)
try:
filename, url = _pick_asset(release)
_check_asset_url(url)
except UpdateDownloadError:
raise
except Exception as exc:
raise UpdateDownloadError(f"could not resolve release: {exc}") from exc
dest = updates / filename
try:
_http_download(url, dest, max_bytes=MAX_DOWNLOAD_BYTES)
_verify_checksum(release, updates, filename, dest)
except Exception:
dest.unlink(missing_ok=True)
raise
return DownloadResult(
version=resolved_version,
file=filename,
path=str(dest),
suggested_command=f"uv pip install {dest} # then restart the backend",
)
tests/test_update_check.py
+341
View File
@@ -4,6 +4,9 @@ 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,
@@ -155,3 +158,341 @@ class TestIsUpdateAvailable:
):
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)