feat: xiaohe upgrade + 启动器更名 Xiaohe Agent + CLI 输出全英文
- 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
This commit is contained in:
@@ -0,0 +1,134 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user