"""Tests for myagents.commands.upgrade.""" import io from pathlib import Path import pytest from click.testing import CliRunner from myagents.commands import upgrade as up_mod @pytest.fixture() def fake_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: home = tmp_path / "home" home.mkdir() monkeypatch.setattr(Path, "home", staticmethod(lambda: home)) return home def _make_runtime(home: Path, version: str) -> Path: target = home / ".xiaohe" / "runtime" / version target.mkdir(parents=True) current = target.parent / "current" current.unlink(missing_ok=True) current.symlink_to(target) return target class TestParseVersion: def test_parses_quoted_version(self) -> None: assert up_mod._parse_version('BASE_URL="x"\nVERSION="v1.2.3"\n') == "v1.2.3" def test_returns_none_on_garbage(self) -> None: assert up_mod._parse_version("no version here") is None class TestPrune: def test_keeps_two_newest(self, fake_home: Path) -> None: root = fake_home / ".xiaohe" / "runtime" for name in ("v1", "v2", "v3", "v4"): d = root / name d.mkdir(parents=True) up_mod._prune() remaining = sorted(p.name for p in root.iterdir()) assert remaining == ["v3", "v4"] class TestUpgradeCommand: def test_already_latest_shortcircuits( self, fake_home: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _make_runtime(fake_home, "v1.0.0") monkeypatch.setattr( up_mod, "_fetch_text", lambda *a: 'VERSION="v1.0.0"\n' ) result = CliRunner().invoke( up_mod.upgrade_cmd, ["--user", "u", "--password", "p"] ) assert result.exit_code == 0 assert "Already up to date" in result.output def test_missing_runtime_errors(self, fake_home: Path) -> None: result = CliRunner().invoke( up_mod.upgrade_cmd, ["--user", "u", "--password", "p"] ) assert result.exit_code != 0 assert "install.sh" in result.output def test_cancelled_by_user( self, fake_home: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _make_runtime(fake_home, "v1.0.0") monkeypatch.setattr( up_mod, "_fetch_text", lambda *a: 'VERSION="v2.0.0"\n' ) result = CliRunner().invoke( up_mod.upgrade_cmd, ["--user", "u", "--password", "p"], input="n\n" ) assert result.exit_code == 0 assert "Cancelled" in result.output def _make_tarball_bytes() -> bytes: import tarfile buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tf: data = b"#!/bin/sh\n" info = tarfile.TarInfo("workspace/setup.sh") info.size = len(data) tf.addfile(info, io.BytesIO(data)) return buf.getvalue() def _wire_upgrade( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, payload: bytes, sha: str, ) -> None: import hashlib def fake_fetch(url: str, user: str, password: str) -> str: if url.endswith("install.sh"): return 'VERSION="v2.0.0"\n' if url.endswith(".sha256"): return f"{sha} xiaohe-agent-latest.tar.gz\n" raise AssertionError(f"unexpected url {url}") monkeypatch.setattr(up_mod, "_fetch_text", fake_fetch) monkeypatch.setattr( up_mod, "_download", lambda url, u, p, dest: dest.write_bytes(payload) ) monkeypatch.setattr(up_mod, "_install_tools", lambda runtime: []) workspace = tmp_path / "ws" workspace.mkdir() monkeypatch.setattr(up_mod, "get_workspace_root", lambda create=False: workspace) monkeypatch.setattr( up_mod, "sync_workspace", lambda ws: {"added": [], "updated": [], "skipped": [], "removed": []}, ) class TestUpgradeFlow: def test_downloads_verifies_and_switches( self, fake_home: Path, monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: import hashlib _make_runtime(fake_home, "v1.0.0") payload = _make_tarball_bytes() sha = hashlib.sha256(payload).hexdigest() _wire_upgrade(monkeypatch, tmp_path, payload, sha) result = CliRunner().invoke( up_mod.upgrade_cmd, ["--user", "u", "--password", "p"], input="y\n" ) assert result.exit_code == 0, result.output assert "Upgraded: v1.0.0 -> v2.0.0" in result.output current = fake_home / ".xiaohe" / "runtime" / "current" assert current.resolve().name == "v2.0.0" assert (current / "setup.sh").is_file() def test_checksum_mismatch_aborts_before_install( self, fake_home: Path, monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: _make_runtime(fake_home, "v1.0.0") payload = _make_tarball_bytes() _wire_upgrade(monkeypatch, tmp_path, payload, "0" * 64) result = CliRunner().invoke( up_mod.upgrade_cmd, ["--user", "u", "--password", "p"], input="y\n" ) assert result.exit_code != 0 assert "Checksum mismatch" in result.output # Old runtime untouched, no new version dir. current = fake_home / ".xiaohe" / "runtime" / "current" assert current.resolve().name == "v1.0.0" assert not (fake_home / ".xiaohe" / "runtime" / "v2.0.0").exists() class TestVerifyChecksum: def test_accepts_matching(self, tmp_path: Path) -> None: import hashlib blob = tmp_path / "f.tar.gz" blob.write_bytes(b"data") sha = hashlib.sha256(b"data").hexdigest() up_mod._verify_checksum(blob, f"{sha} f.tar.gz\n", "http://x/") def test_rejects_mismatch_and_garbage(self, tmp_path: Path) -> None: import click blob = tmp_path / "f.tar.gz" blob.write_bytes(b"data") with pytest.raises(click.ClickException, match="Checksum mismatch"): up_mod._verify_checksum(blob, f"{'0' * 64} f.tar.gz\n", "http://x/") with pytest.raises(click.ClickException, match="looks wrong"): up_mod._verify_checksum(blob, "not-a-checksum\n", "http://x/") class _FakeResponse: def __init__(self, data: bytes) -> None: self.headers = {"Content-Length": str(len(data))} self._buf = io.BytesIO(data) def read(self, n: int = -1) -> bytes: return self._buf.read(n) def __enter__(self) -> "_FakeResponse": return self def __exit__(self, *args: object) -> None: return None class TestFetch: def test_fetch_text_sends_basic_auth( self, monkeypatch: pytest.MonkeyPatch ) -> None: seen: dict[str, str] = {} def fake_urlopen(req, timeout=0): seen["auth"] = req.headers["Authorization"] return _FakeResponse(b"hello") monkeypatch.setattr("urllib.request.urlopen", fake_urlopen) assert up_mod._fetch_text("http://x/", "u", "p") == "hello" assert seen["auth"].startswith("Basic ") def test_401_maps_to_friendly_error( self, monkeypatch: pytest.MonkeyPatch ) -> None: import urllib.error def fake_urlopen(req, timeout=0): raise urllib.error.HTTPError("http://x/", 401, "Unauthorized", {}, None) monkeypatch.setattr("urllib.request.urlopen", fake_urlopen) with pytest.raises(Exception, match="401"): up_mod._fetch_text("http://x/", "u", "p") def test_download_writes_file( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: payload = b"x" * 100_000 monkeypatch.setattr( "urllib.request.urlopen", lambda req, timeout=0: _FakeResponse(payload), ) dest = tmp_path / "out.tar.gz" up_mod._download("http://x/f", "u", "p", dest) assert dest.read_bytes() == payload class TestPipInstallFlags: def test_uses_break_system_packages_on_externally_managed( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: import sysconfig em = tmp_path / "EXTERNALLY-MANAGED" em.write_text("[externally-managed]\n") monkeypatch.setattr( sysconfig, "get_path", lambda name: str(tmp_path) if name == "stdlib" else "" ) seen: list[list[str]] = [] monkeypatch.setattr( "subprocess.run", lambda cmd, **kw: seen.append(cmd) or type("R", (), {"returncode": 0, "stderr": ""})(), ) up_mod._pip_install(tmp_path / "pkg") assert "--break-system-packages" in seen[0] def test_skips_flag_without_externally_managed( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: import sysconfig monkeypatch.setattr( sysconfig, "get_path", lambda name: str(tmp_path) if name == "stdlib" else "" ) seen: list[list[str]] = [] monkeypatch.setattr( "subprocess.run", lambda cmd, **kw: seen.append(cmd) or type("R", (), {"returncode": 0, "stderr": ""})(), ) up_mod._pip_install(tmp_path / "pkg") assert "--break-system-packages" not in seen[0]