"""Comprehensive tests for ``muse read-snapshot``. Coverage tiers -------------- - Unit: schema keys constant - Integration: JSON/text, --no-manifest, --path-prefix, manifest presence - Security: ANSI in snapshot IDs rejected, no traceback on bad input - Stress: 1000-path manifest, 200 sequential reads """ from __future__ import annotations import datetime import json import pathlib from muse.core.errors import ExitCode from muse.core.paths import muse_dir from muse.core.ids import hash_snapshot from muse.core.store import SnapshotRecord, write_snapshot from tests.cli_test_helper import CliRunner, InvokeResult from muse.core.types import NULL_COMMIT_ID, fake_id, long_id, short_id runner = CliRunner() _CREATED_AT: datetime.datetime = datetime.datetime( 2026, 3, 18, 12, 0, tzinfo=datetime.timezone.utc ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: repo = tmp_path / "repo" dot_muse = muse_dir(repo) for sub in ("objects", "commits", "snapshots", "refs/heads"): (dot_muse / sub).mkdir(parents=True) (dot_muse / "HEAD").write_text("ref: refs/heads/main") (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"})) return repo def _snap( repo: pathlib.Path, manifest: Manifest | None = None, ) -> str: """Write a snapshot with a real content-addressed ID; return the snapshot_id.""" m: Manifest = manifest or {} snap_id = hash_snapshot(m) rec = SnapshotRecord( snapshot_id=snap_id, manifest=m, created_at=_CREATED_AT, ) write_snapshot(repo, rec) return snap_id def _rs(repo: pathlib.Path, *args: str) -> InvokeResult: from muse.cli.app import main as cli return runner.invoke( cli, ["read-snapshot", *args], env={"MUSE_REPO_ROOT": str(repo)}, ) def _rsj(repo: pathlib.Path, *args: str) -> InvokeResult: """Like _rs but always passes --json.""" return _rs(repo, "--json", *args) def _fake_oid(n: int) -> str: return format(n, "064x") # --------------------------------------------------------------------------- # Integration — JSON format # --------------------------------------------------------------------------- class TestJsonFormat: def test_full_output_empty_manifest(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) result = _rsj(repo, sid) assert result.exit_code == 0 data = json.loads(result.output) assert data["snapshot_id"] == sid assert data["file_count"] == 0 assert data["manifest"] == {} def test_manifest_paths_present(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = NULL_COMMIT_ID sid = _snap(repo, {"src/main.py": oid, "tests/test_main.py": oid}) data = json.loads(_rsj(repo, sid).output) assert "src/main.py" in data["manifest"] assert "tests/test_main.py" in data["manifest"] assert data["file_count"] == 2 def test_json_flag_shorthand(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo, {"a.py": _fake_oid(1)}) result = _rs(repo, "--json", sid) assert result.exit_code == 0 assert "snapshot_id" in json.loads(result.output) def test_created_at_iso8601(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo, {"b.py": _fake_oid(2)}) data = json.loads(_rsj(repo, sid).output) datetime.datetime.fromisoformat(data["created_at"]) def test_file_count_reflects_manifest(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) manifest = {f"file{i}.py": _fake_oid(i) for i in range(5)} sid = _snap(repo, manifest) data = json.loads(_rsj(repo, sid).output) assert data["file_count"] == 5 # --------------------------------------------------------------------------- # Integration — text format # --------------------------------------------------------------------------- class TestTextFormat: def test_text_contains_prefix(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo, {"c.py": _fake_oid(3)}) result = _rs(repo, sid) assert result.exit_code == 0 assert short_id(sid) in result.output def test_text_contains_file_count(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo, {"a.py": _fake_oid(1), "b.py": _fake_oid(2)}) result = _rs(repo, sid) assert "2 files" in result.output def test_text_single_line(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) result = _rs(repo, sid) lines = [l for l in result.output.splitlines() if l.strip()] assert len(lines) == 1 # --------------------------------------------------------------------------- # Integration — --no-manifest # --------------------------------------------------------------------------- class TestNoManifest: def test_manifest_absent(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo, {"a.py": _fake_oid(1)}) data = json.loads(_rsj(repo, "--no-manifest", sid).output) assert "manifest" not in data assert data["file_count"] == 1 def test_snapshot_id_and_created_at_still_present(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_rsj(repo, "--no-manifest", sid).output) assert data["snapshot_id"] == sid assert "created_at" in data def test_no_manifest_with_text_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) result = _rs(repo, "--no-manifest", sid) assert result.exit_code == ExitCode.USER_ERROR # --------------------------------------------------------------------------- # Integration — --path-prefix # --------------------------------------------------------------------------- class TestPathPrefix: def test_prefix_filters_manifest(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo, { "src/a.py": _fake_oid(1), "src/b.py": _fake_oid(2), "tests/c.py": _fake_oid(3), }) data = json.loads(_rsj(repo, "--path-prefix", "src/", sid).output) assert set(data["manifest"].keys()) == {"src/a.py", "src/b.py"} assert data["file_count"] == 2 def test_prefix_no_match_returns_empty(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo, {"src/a.py": _fake_oid(1)}) data = json.loads(_rsj(repo, "--path-prefix", "docs/", sid).output) assert data["manifest"] == {} assert data["file_count"] == 0 def test_prefix_with_text_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) result = _rs(repo, "--path-prefix", "src/", sid) assert result.exit_code == ExitCode.USER_ERROR # --------------------------------------------------------------------------- # Error cases # --------------------------------------------------------------------------- class TestErrors: def test_missing_snapshot_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) # Valid sha256: format but not present in the store — must get "not found". result = _rs(repo, long_id(f"dead{'beef' * 15}")) assert result.exit_code == ExitCode.USER_ERROR def test_invalid_snapshot_id_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _rs(repo, "not-valid") assert result.exit_code == ExitCode.USER_ERROR def test_unknown_format_errors_argparse_rejects(self, tmp_path: pathlib.Path) -> None: """--format flag no longer exists; argparse exits 2.""" repo = _make_repo(tmp_path) sid = _snap(repo) result = _rs(repo, "--format", "msgpack", sid) assert result.exit_code != 0 # argparse rejects unknown flag with exit 2 # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: def test_ansi_in_snapshot_id_rejected(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _rs(repo, f"\x1b[31m{'a' * 64}") assert result.exit_code == ExitCode.USER_ERROR def test_no_traceback_on_bad_id(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _rs(repo, "bad-id") assert "Traceback" not in result.output # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestStress: def test_1000_file_manifest(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) manifest = {f"src/module{i:04d}.py": _fake_oid(i) for i in range(1000)} sid = _snap(repo, manifest) result = _rsj(repo, sid) assert result.exit_code == 0 data = json.loads(result.output) assert data["file_count"] == 1000 assert len(data["manifest"]) == 1000 def test_1000_file_manifest_no_manifest(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) manifest = {f"src/module{i:04d}.py": _fake_oid(i) for i in range(1000)} sid = _snap(repo, manifest) result = _rsj(repo, "--no-manifest", sid) assert result.exit_code == 0 data = json.loads(result.output) assert data["file_count"] == 1000 assert "manifest" not in data def test_200_sequential_reads(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo, {"a.py": _fake_oid(0)}) for i in range(200): result = _rsj(repo, sid) assert result.exit_code == 0, f"failed at iteration {i}" data = json.loads(result.output) assert data["file_count"] == 1 class TestRegisterFlags: def _parse(self, *args: str) -> "argparse.Namespace": import argparse from muse.cli.commands.read_snapshot import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) return p.parse_args(["read-snapshot", fake_id("a"), *args]) def test_json_short_flag(self) -> None: args = self._parse("-j") assert args.json_out is True def test_json_long_flag(self) -> None: args = self._parse("--json") assert args.json_out is True def test_default_no_json(self) -> None: args = self._parse() assert args.json_out is False