Zhengshou Lai
7b6da50c93
uv run injects VIRTUAL_ENV, which silently selects the uv pip branch and broke test_uses_break_system_packages_on_externally_managed. Both tests now delete VIRTUAL_ENV to pin the plain-pip branch; new test pins the uv branch via mocks.
290 lines
9.9 KiB
Python
290 lines
9.9 KiB
Python
"""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
|
|
|
|
# Force the plain-pip branch regardless of how pytest was launched
|
|
# (e.g. `uv run` sets VIRTUAL_ENV, which would select the uv branch).
|
|
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
|
|
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.delenv("VIRTUAL_ENV", raising=False)
|
|
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]
|
|
|
|
def test_prefers_uv_pip_inside_virtualenv(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setenv("VIRTUAL_ENV", str(tmp_path / "venv"))
|
|
monkeypatch.setattr(up_mod.shutil, "which", lambda name: f"/usr/bin/{name}")
|
|
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 seen[0][:3] == ["uv", "pip", "install"]
|
|
assert "--break-system-packages" not in seen[0]
|