From 84ad0fa4cf499be83eb154be77db9f3b54edbe07 Mon Sep 17 00:00:00 2001 From: Zhengshou Lai Date: Thu, 16 Jul 2026 19:36:18 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=A1=E6=9F=A5=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E4=B8=80=E8=BD=AE=E2=80=94=E2=80=94sync=20=E8=A7=A3=E5=BC=95?= =?UTF-8?q?=E7=94=A8=E8=BD=AF=E9=93=BE=E6=8A=80=E8=83=BD/=E9=98=B2=20CLAUD?= =?UTF-8?q?E.md=20=E5=B4=A9=E6=BA=83/init=20=E5=BB=BA=20agent-tasks?= =?UTF-8?q?=EF=BC=9Bupgrade=20SHA256=20=E6=A0=A1=E9=AA=8C+=E6=94=B6?= =?UTF-8?q?=E5=B0=BE=E5=AE=B9=E9=94=99=EF=BC=9B=E6=96=B0=E5=A2=9E=20xiaohe?= =?UTF-8?q?=20switch=20=E5=9B=9E=E6=BB=9A=EF=BC=9Buninstall=20--all=20?= =?UTF-8?q?=E6=89=A9=E5=88=B0=20~/.metabot=20~/.mytoolkit=EF=BC=9B?= =?UTF-8?q?=E4=B8=8B=E8=BD=BD=E8=B7=AF=E5=BE=84=E6=94=B9=20xiaohe-agent?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- myagents/commands/switch.py | 84 +++++++++++++++++++++++++ myagents/commands/sync_workspace.py | 40 ++++++++++-- myagents/commands/uninstall.py | 16 +++-- myagents/commands/upgrade.py | 42 +++++++++++-- myagents/entrypoints.py | 4 +- tests/test_switch.py | 80 +++++++++++++++++++++++ tests/test_sync_workspace.py | 75 ++++++++++++++++++++++ tests/test_uninstall.py | 10 ++- tests/test_upgrade.py | 98 +++++++++++++++++++++++++++++ 9 files changed, 432 insertions(+), 17 deletions(-) create mode 100644 myagents/commands/switch.py create mode 100644 tests/test_switch.py diff --git a/myagents/commands/switch.py b/myagents/commands/switch.py new file mode 100644 index 0000000..5e2b55e --- /dev/null +++ b/myagents/commands/switch.py @@ -0,0 +1,84 @@ +"""``xiaohe switch`` — move ``current`` to another installed runtime version. + +Rollback/roll-forward counterpart to ``xiaohe upgrade``: repoints the +``current`` symlink, reinstalls the CLI tools from that runtime, and re-syncs +the workspace. Run without an argument to list installed versions and pick +one interactively. +""" + +import click +from rich.console import Console + +from myagents.commands.sync_workspace import _print_report, sync_workspace +from myagents.commands.upgrade import _current_version, _install_tools, _runtime_root +from myagents.project_root import get_workspace_root + +stderr_console = Console(stderr=True) +console = Console() + + +def _installed_versions() -> list[str]: + root = _runtime_root() + if not root.is_dir(): + return [] + return sorted( + d.name for d in root.iterdir() if d.is_dir() and not d.is_symlink() + ) + + +@click.command("switch") +@click.argument("version", required=False) +@click.option("--yes", is_flag=True, help="Skip the confirmation prompt") +def switch_cmd(version: str | None, yes: bool) -> None: + """Switch the active runtime to another installed version (rollback).""" + versions = _installed_versions() + current = _current_version() + if not versions: + raise click.ClickException( + "No installed runtime versions found (~/.xiaohe/runtime)." + ) + + if not version: + console.print("[bold]Installed versions:[/bold]") + for name in versions: + marker = " [green](current)[/green]" if name == current else "" + console.print(f" {name}{marker}") + version = str(click.prompt("Switch to", err=True)) + + if version not in versions: + raise click.ClickException( + f"Version {version!r} is not installed. Installed: {', '.join(versions)}" + ) + if version == current: + console.print(f"Already on {version}.") + return + if not yes and not click.confirm( + f"Switch runtime {current} -> {version} and reinstall tools?", + default=True, + err=True, + ): + console.print("Cancelled.") + return + + target = _runtime_root() / version + current_link = _runtime_root() / "current" + current_link.unlink(missing_ok=True) + current_link.symlink_to(target) + + warnings = _install_tools(target) + + workspace = get_workspace_root(create=False) + if workspace.is_dir(): + try: + report = sync_workspace(workspace) + except click.ClickException as exc: + stderr_console.print(f"[yellow]workspace sync skipped: {exc.message}[/yellow]") + else: + _print_report(workspace, report) + + for warning in warnings: + stderr_console.print(f"[yellow]warning: {warning}[/yellow]") + console.print( + f"\n[bold green]Switched: {current} -> {version}[/bold green] " + "(open a new terminal to pick up the change)" + ) diff --git a/myagents/commands/sync_workspace.py b/myagents/commands/sync_workspace.py index cd11796..6e8e11d 100644 --- a/myagents/commands/sync_workspace.py +++ b/myagents/commands/sync_workspace.py @@ -86,15 +86,35 @@ def _hash_file(path: Path) -> str: def _enumerate_managed(runtime: Path) -> dict[str, str]: - """rel-path -> sha256 for every managed file in the runtime tree.""" + """rel-path -> sha256 for every managed file in the runtime tree. + + Symlinked dirs (e.g. .agents/skills/mytoolkit -> contrib/...) are followed + and their content is hashed through the link, so the workspace receives + real self-contained files. Directory inodes are tracked to break cycles. + """ files: dict[str, str] = {} for rel in MANAGED_PATHS: src = runtime / rel if src.is_file(): files[rel] = _hash_file(src) - elif src.is_dir(): - for path in sorted(src.rglob("*")): - if path.is_file() and not path.is_symlink(): + continue + if not src.is_dir(): + continue + seen: set[tuple[int, int]] = set() + for dirpath, dirnames, filenames in os.walk(src, followlinks=True): + try: + real = Path(dirpath).resolve().stat() + except OSError: + dirnames[:] = [] + continue + key = (real.st_dev, real.st_ino) + if key in seen: + dirnames[:] = [] + continue + seen.add(key) + for name in sorted(filenames): + path = Path(dirpath) / name + if path.is_file(): files[str(path.relative_to(runtime))] = _hash_file(path) return files @@ -175,7 +195,10 @@ def sync_workspace(workspace: Path, force: bool = False) -> dict: for link, target_rel in MANAGED_SYMLINKS.items(): link_path = workspace / link - if not link_path.is_symlink() and (workspace / target_rel).exists(): + # Never clobber: an existing entry (real file, dir, or link) stays put. + if link_path.is_symlink() or link_path.exists(): + continue + if (workspace / target_rel).exists(): link_path.symlink_to(target_rel) _write_stamp(workspace, _runtime_version(runtime), new_stamp) @@ -328,6 +351,13 @@ def init_cmd(path: str | None, force: bool) -> None: "skills/docs will sync after reinstall.[/yellow]" ) + # Worklog task board: instantiate the live file from the synced template. + template = workspace / "assistant" / "agent-tasks.template.md" + live = workspace / "assistant" / "agent-tasks.md" + if template.is_file() and not live.exists(): + shutil.copy2(template, live) + console.print(" [green]created:[/green] assistant/agent-tasks.md (from template)") + _create_launcher() console.print("\n[bold green]Done.[/bold green] Daily use: [cyan]xiaohe[/cyan]") diff --git a/myagents/commands/uninstall.py b/myagents/commands/uninstall.py index a424ad8..4bef9bd 100644 --- a/myagents/commands/uninstall.py +++ b/myagents/commands/uninstall.py @@ -2,8 +2,9 @@ Default: removes the entry points (xiaohe/myclaude/.../mytoolkit), metabot CLI, shell completions, and the desktop launcher. Keeps ~/.xiaohe (runtime, -settings, keys) and the workspace. ``--all`` also deletes ~/.xiaohe; the -workspace is never touched. +settings, keys), ~/.metabot, ~/.mytoolkit and the workspace. ``--all`` also +deletes ~/.xiaohe, ~/.metabot and ~/.mytoolkit (every saved credential); +the workspace is never touched. """ import os @@ -89,9 +90,12 @@ def uninstall_cmd(remove_all: bool, yes: bool) -> None: console.print(" - pip packages: mytoolkit, myagents") if remove_all: console.print(" - [red]~/.xiaohe (runtime, settings.json, config.json keys)[/red]") + console.print(" - [red]~/.metabot (Feishu bot config)[/red]") + console.print(" - [red]~/.mytoolkit (mytoolkit config, incl. keys)[/red]") console.print("[bold]Will keep:[/bold]") if not remove_all: console.print(" - ~/.xiaohe (runtime, settings, keys)") + console.print(" - ~/.metabot and ~/.mytoolkit (bot/tool configs)") console.print(f" - workspace: {get_workspace_root(create=False)}") console.print(" - rc-file edits (PATH line, ANTHROPIC_* exports) and the claude CLI") @@ -99,11 +103,11 @@ def uninstall_cmd(remove_all: bool, yes: bool) -> None: console.print("Cancelled.") return if remove_all: - # Deleting ~/.xiaohe destroys settings and every saved API key — + # Deleting these dirs destroys settings and every saved credential — # always require the full confirmation chain, even with --yes. if not click.confirm( - "This also deletes ~/.xiaohe — runtime, settings and ALL saved API keys. " - "Continue?", + "This also deletes ~/.xiaohe, ~/.metabot and ~/.mytoolkit — runtime, " + "bot config and ALL saved API keys. Continue?", default=False, err=True, ): @@ -138,6 +142,8 @@ def uninstall_cmd(remove_all: bool, yes: bool) -> None: if remove_all: _remove_path(home / ".xiaohe", removed) + _remove_path(home / ".metabot", removed) + _remove_path(home / ".mytoolkit", removed) for warning in warnings: stderr_console.print(f"[yellow]warning: {warning}[/yellow]") diff --git a/myagents/commands/upgrade.py b/myagents/commands/upgrade.py index 042d265..87cebcc 100644 --- a/myagents/commands/upgrade.py +++ b/myagents/commands/upgrade.py @@ -8,6 +8,7 @@ workspace content. Credentials come from XIAOHE_USER/XIAOHE_PASS, --user/ """ import base64 +import hashlib import re import shutil import subprocess @@ -40,7 +41,7 @@ from myagents.settings import get_setting stderr_console = Console(stderr=True) console = Console() -DEFAULT_BASE_URL = "http://1.14.226.205:8088/xiaohe" +DEFAULT_BASE_URL = "http://1.14.226.205:8088/xiaohe-agent" KEEP_VERSIONS = 2 @@ -113,6 +114,29 @@ def _parse_version(install_sh: str) -> str | None: return match.group(1) if match else None +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as fh: + for chunk in iter(lambda: fh.read(1 << 16), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _verify_checksum( + tarball: Path, sha256_text: str, url: str +) -> None: + """Verify tarball against the published " " line.""" + expected = sha256_text.split()[0].strip().lower() if sha256_text.split() else "" + if not expected or not all(c in "0123456789abcdef" for c in expected): + raise click.ClickException(f"Checksum file looks wrong — {url}") + actual = _sha256(tarball) + if actual != expected: + raise click.ClickException( + "Checksum mismatch — the download is corrupted or tampered with. " + "Aborting; nothing was installed." + ) + + def _extract(tarball: Path, dest: Path) -> None: with tarfile.open(tarball) as tf: try: @@ -200,8 +224,11 @@ def upgrade_cmd(user: str | None, password: str | None, force: bool) -> None: tmpdir = Path(tempfile.mkdtemp(prefix="xiaohe-upgrade-")) try: - tarball = tmpdir / "xiaohe-agent.tar.gz" + tarball = tmpdir / "xiaohe-agent-latest.tar.gz" _download(f"{base}/xiaohe-agent-latest.tar.gz", user, password, tarball) + with console.status("Verifying checksum ..."): + sha_url = f"{base}/xiaohe-agent-latest.tar.gz.sha256" + _verify_checksum(tarball, _fetch_text(sha_url, user, password), sha_url) target = _runtime_root() / latest with console.status(f"Installing to {target} ..."): @@ -225,8 +252,13 @@ def upgrade_cmd(user: str | None, password: str | None, force: bool) -> None: workspace = get_workspace_root(create=False) if workspace.is_dir(): - report = sync_workspace(workspace) - _print_report(workspace, report) + try: + report = sync_workspace(workspace) + except click.ClickException as exc: + # e.g. dev machine: workspace is a git checkout — not a failure. + stderr_console.print(f"[yellow]workspace sync skipped: {exc.message}[/yellow]") + else: + _print_report(workspace, report) else: console.print(f"[yellow]workspace {workspace} missing — run 'xiaohe init' first.[/yellow]") @@ -234,5 +266,5 @@ def upgrade_cmd(user: str | None, password: str | None, force: bool) -> None: stderr_console.print(f"[yellow]warning: {warning}[/yellow]") console.print( f"\n[bold green]Upgraded: {current} -> {latest}[/bold green] " - "(open a new terminal; previous runtime kept for rollback)" + "(open a new terminal; roll back anytime with 'xiaohe switch')" ) diff --git a/myagents/entrypoints.py b/myagents/entrypoints.py index 449a2b9..7a43679 100644 --- a/myagents/entrypoints.py +++ b/myagents/entrypoints.py @@ -26,7 +26,8 @@ def default_agent() -> str: def build_xiaohe_cli(): - """The ``xiaohe`` command group: agent forwarding + init/sync/upgrade/uninstall/version.""" + """The ``xiaohe`` command group: agent forwarding + init/sync/upgrade/switch/uninstall/version.""" + from myagents.commands.switch import switch_cmd from myagents.commands.sync_workspace import init_cmd, sync_cmd from myagents.commands.uninstall import uninstall_cmd from myagents.commands.upgrade import upgrade_cmd @@ -36,6 +37,7 @@ def build_xiaohe_cli(): xiaohe_cli.add_command(init_cmd) xiaohe_cli.add_command(sync_cmd) xiaohe_cli.add_command(upgrade_cmd) + xiaohe_cli.add_command(switch_cmd) xiaohe_cli.add_command(uninstall_cmd) xiaohe_cli.add_command(version_cmd) return xiaohe_cli diff --git a/tests/test_switch.py b/tests/test_switch.py new file mode 100644 index 0000000..9f6802b --- /dev/null +++ b/tests/test_switch.py @@ -0,0 +1,80 @@ +"""Tests for myagents.commands.switch.""" + +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from myagents.commands import switch as sw_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)) + monkeypatch.setattr(sw_mod, "_install_tools", lambda runtime: []) + workspace = tmp_path / "ws" + workspace.mkdir() + monkeypatch.setattr(sw_mod, "get_workspace_root", lambda create=False: workspace) + monkeypatch.setattr( + sw_mod, + "sync_workspace", + lambda ws: {"added": [], "updated": [], "skipped": [], "removed": []}, + ) + return home + + +def _make_versions(home: Path, versions: tuple[str, ...], current: str) -> None: + root = home / ".xiaohe" / "runtime" + for name in versions: + (root / name).mkdir(parents=True) + (root / "current").symlink_to(root / current) + + +class TestSwitch: + def test_switches_current_and_reinstalls( + self, fake_home: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + _make_versions(fake_home, ("v1", "v2"), "v2") + installed: list[str] = [] + monkeypatch.setattr( + sw_mod, "_install_tools", lambda rt: installed.append(rt.name) or [] + ) + result = CliRunner().invoke(sw_mod.switch_cmd, ["v1", "--yes"]) + assert result.exit_code == 0, result.output + assert "Switched: v2 -> v1" in result.output + current = fake_home / ".xiaohe" / "runtime" / "current" + assert current.resolve().name == "v1" + assert installed == ["v1"] + + def test_unknown_version_errors(self, fake_home: Path) -> None: + _make_versions(fake_home, ("v1", "v2"), "v2") + result = CliRunner().invoke(sw_mod.switch_cmd, ["v9", "--yes"]) + assert result.exit_code != 0 + assert "not installed" in result.output + + def test_already_current_is_noop(self, fake_home: Path) -> None: + _make_versions(fake_home, ("v1", "v2"), "v2") + result = CliRunner().invoke(sw_mod.switch_cmd, ["v2", "--yes"]) + assert result.exit_code == 0 + assert "Already on v2" in result.output + + def test_interactive_list_and_prompt(self, fake_home: Path) -> None: + _make_versions(fake_home, ("v1", "v2"), "v2") + result = CliRunner().invoke(sw_mod.switch_cmd, [], input="v1\ny\n") + assert result.exit_code == 0, result.output + assert "v2" in result.output and "(current)" in result.output + assert (fake_home / ".xiaohe" / "runtime" / "current").resolve().name == "v1" + + def test_cancelled_keeps_current(self, fake_home: Path) -> None: + _make_versions(fake_home, ("v1", "v2"), "v2") + result = CliRunner().invoke(sw_mod.switch_cmd, ["v1"], input="n\n") + assert result.exit_code == 0 + assert "Cancelled" in result.output + assert (fake_home / ".xiaohe" / "runtime" / "current").resolve().name == "v2" + + def test_no_versions_errors(self, fake_home: Path) -> None: + result = CliRunner().invoke(sw_mod.switch_cmd, ["v1", "--yes"]) + assert result.exit_code != 0 + assert "No installed runtime" in result.output diff --git a/tests/test_sync_workspace.py b/tests/test_sync_workspace.py index 97de343..597bf4e 100644 --- a/tests/test_sync_workspace.py +++ b/tests/test_sync_workspace.py @@ -86,6 +86,41 @@ class TestSyncWorkspace: report = sync_mod.sync_workspace(workspace, force=True) assert report["added"] + def test_symlinked_skill_dir_is_dereferenced( + self, runtime: Path, workspace: Path + ) -> None: + # Skills living in contrib/ are linked into .agents/skills (relative + # symlink); the workspace must receive real self-contained files. + real = runtime / "contrib" / "mytoolkit" / ".claude" / "skills" / "mytoolkit" + real.mkdir(parents=True) + (real / "SKILL.md").write_text("toolkit skill v1") + import os + + rel_target = os.path.relpath(real, runtime / ".agents" / "skills") + (runtime / ".agents" / "skills" / "mytoolkit").symlink_to( + rel_target, target_is_directory=True + ) + report = sync_mod.sync_workspace(workspace) + copied = workspace / ".agents" / "skills" / "mytoolkit" / "SKILL.md" + assert copied.is_file() and not copied.is_symlink() + assert copied.read_text() == "toolkit skill v1" + assert ".agents/skills/mytoolkit/SKILL.md" in report["added"] + # Second run: content identical -> no spurious updates. + assert sync_mod.sync_workspace(workspace) == { + "added": [], "updated": [], "skipped": [], "removed": [] + } + + def test_existing_real_managed_symlink_target_is_kept( + self, runtime: Path, workspace: Path + ) -> None: + # A pre-existing real CLAUDE.md must not be replaced by a symlink. + (workspace / "CLAUDE.md").write_text("my own claude rules") + sync_mod.sync_workspace(workspace) + claude_md = workspace / "CLAUDE.md" + assert not claude_md.is_symlink() + assert claude_md.read_text() == "my own claude rules" + assert (workspace / ".claude").is_symlink() # other links still made + class TestCreateLauncher: @pytest.fixture() @@ -133,3 +168,43 @@ class TestCreateLauncher: text = entry.read_text() assert "Name=Xiaohe Agent" in text assert "Icon=xiaohe-agent" in text + + +class TestInitCommand: + @pytest.fixture() + def fake_home(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + home = tmp_path / "home" + home.mkdir() + monkeypatch.setattr(Path, "home", staticmethod(lambda: home)) + monkeypatch.setattr(sync_mod, "get_setting", lambda *a, **kw: "") + monkeypatch.setattr(sync_mod, "set_setting", lambda *a, **kw: None) + return home + + def test_agent_tasks_created_from_template( + self, runtime: Path, workspace: Path, fake_home: Path + ) -> None: + from click.testing import CliRunner + + template = runtime / "assistant" / "agent-tasks.template.md" + template.parent.mkdir(parents=True) + template.write_text("# task board template") + result = CliRunner().invoke(sync_mod.init_cmd, [str(workspace)]) + assert result.exit_code == 0, result.output + live = workspace / "assistant" / "agent-tasks.md" + assert live.read_text() == "# task board template" + assert (workspace / "tmp").is_dir() # skeleton dirs created + + def test_existing_agent_tasks_untouched( + self, runtime: Path, workspace: Path, fake_home: Path + ) -> None: + from click.testing import CliRunner + + template = runtime / "assistant" / "agent-tasks.template.md" + template.parent.mkdir(parents=True) + template.write_text("# task board template") + live = workspace / "assistant" / "agent-tasks.md" + live.parent.mkdir(parents=True) + live.write_text("# my live board") + result = CliRunner().invoke(sync_mod.init_cmd, [str(workspace)]) + assert result.exit_code == 0, result.output + assert live.read_text() == "# my live board" diff --git a/tests/test_uninstall.py b/tests/test_uninstall.py index 826b30e..3cefb83 100644 --- a/tests/test_uninstall.py +++ b/tests/test_uninstall.py @@ -30,6 +30,10 @@ def _populate(home: Path) -> None: (local_bin / "completions" / "_metabot").write_text("comp") (home / ".xiaohe" / "agent").mkdir(parents=True) (home / ".xiaohe" / "agent" / "config.json").write_text("{}") + (home / ".metabot").mkdir() + (home / ".metabot" / "bots.json").write_text("{}") + (home / ".mytoolkit").mkdir() + (home / ".mytoolkit" / "config.json").write_text("{}") (home / "workspace").mkdir() app = home / "Desktop" / "Xiaohe Agent.app" / "Contents" app.mkdir(parents=True) @@ -49,15 +53,19 @@ class TestUninstall: assert not (local_bin / "completions").exists() # empty dir cleaned assert not (fake_home / "Desktop" / "Xiaohe Agent.app").exists() assert (fake_home / ".xiaohe" / "agent" / "config.json").is_file() + assert (fake_home / ".metabot" / "bots.json").is_file() + assert (fake_home / ".mytoolkit" / "config.json").is_file() assert (fake_home / "workspace").is_dir() - def test_all_removes_xiaohe_dir(self, fake_home: Path) -> None: + def test_all_removes_config_dirs(self, fake_home: Path) -> None: _populate(fake_home) result = CliRunner().invoke( un_mod.uninstall_cmd, ["--all", "--yes"], input="y\ndelete-all\n" ) assert result.exit_code == 0 assert not (fake_home / ".xiaohe").exists() + assert not (fake_home / ".metabot").exists() + assert not (fake_home / ".mytoolkit").exists() assert (fake_home / "workspace").is_dir() # workspace never touched def test_all_requires_full_chain_even_with_yes(self, fake_home: Path) -> None: diff --git a/tests/test_upgrade.py b/tests/test_upgrade.py index 45810be..813b369 100644 --- a/tests/test_upgrade.py +++ b/tests/test_upgrade.py @@ -80,6 +80,104 @@ class TestUpgradeCommand: 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))}