Zhengshou Lai
5d34b247ef
- upgrade: 查版本(install.sh)→确认→rich 进度条下载→落 runtime 切 current(留两版)→重装三件套→自动 sync
- launcher: 'Xiaohe Agent.app' / 'Xiaohe Agent.desktop'，自动清理旧 XiaoheAgent.{command,app,desktop}
- 修复 _current_version 对缺失 current 软链误判为 'current'
- 全部提示/log 英文化；93 tests passed
135 lines
4.1 KiB
Python
135 lines
4.1 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
|
|
|
|
|
|
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
|