gabriel / muse public
test_cmd_read_snapshot.py python
305 lines 10.9 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 140 days ago
1 """Comprehensive tests for ``muse read-snapshot``.
2
3 Coverage tiers
4 --------------
5 - Unit: schema keys constant
6 - Integration: JSON/text, --no-manifest, --path-prefix, manifest presence
7 - Security: ANSI in snapshot IDs rejected, no traceback on bad input
8 - Stress: 1000-path manifest, 200 sequential reads
9 """
10 from __future__ import annotations
11
12 import datetime
13 import json
14 import pathlib
15
16 from muse.core.errors import ExitCode
17 from muse.core.snapshot import compute_snapshot_id
18 from muse.core.store import SnapshotRecord, write_snapshot
19 from tests.cli_test_helper import CliRunner, InvokeResult
20 from muse.core._types import fake_id, long_id, short_id
21
22 runner = CliRunner()
23
24 _CREATED_AT: datetime.datetime = datetime.datetime(
25 2026, 3, 18, 12, 0, tzinfo=datetime.timezone.utc
26 )
27
28
29 # ---------------------------------------------------------------------------
30 # Helpers
31 # ---------------------------------------------------------------------------
32
33 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
34 repo = tmp_path / "repo"
35 muse = repo / ".muse"
36 for sub in ("objects", "commits", "snapshots", "refs/heads"):
37 (muse / sub).mkdir(parents=True)
38 (muse / "HEAD").write_text("ref: refs/heads/main")
39 (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"}))
40 return repo
41
42
43 def _snap(
44 repo: pathlib.Path,
45 manifest: Manifest | None = None,
46 ) -> str:
47 """Write a snapshot with a real content-addressed ID; return the snapshot_id."""
48 m: Manifest = manifest or {}
49 snap_id = compute_snapshot_id(m)
50 rec = SnapshotRecord(
51 snapshot_id=snap_id,
52 manifest=m,
53 created_at=_CREATED_AT,
54 )
55 write_snapshot(repo, rec)
56 return snap_id
57
58
59 def _rs(repo: pathlib.Path, *args: str) -> InvokeResult:
60 from muse.cli.app import main as cli
61 return runner.invoke(
62 cli,
63 ["read-snapshot", *args],
64 env={"MUSE_REPO_ROOT": str(repo)},
65 )
66
67
68 def _rsj(repo: pathlib.Path, *args: str) -> InvokeResult:
69 """Like _rs but always passes --json."""
70 return _rs(repo, "--json", *args)
71
72
73 def _fake_oid(n: int) -> str:
74 return format(n, "064x")
75
76
77 # ---------------------------------------------------------------------------
78 # Integration — JSON format
79 # ---------------------------------------------------------------------------
80
81
82 class TestJsonFormat:
83 def test_full_output_empty_manifest(self, tmp_path: pathlib.Path) -> None:
84 repo = _make_repo(tmp_path)
85 sid = _snap(repo)
86 result = _rsj(repo, sid)
87 assert result.exit_code == 0
88 data = json.loads(result.output)
89 assert data["snapshot_id"] == sid
90 assert data["file_count"] == 0
91 assert data["manifest"] == {}
92
93 def test_manifest_paths_present(self, tmp_path: pathlib.Path) -> None:
94 repo = _make_repo(tmp_path)
95 oid = "0" * 64
96 sid = _snap(repo, {"src/main.py": oid, "tests/test_main.py": oid})
97 data = json.loads(_rsj(repo, sid).output)
98 assert "src/main.py" in data["manifest"]
99 assert "tests/test_main.py" in data["manifest"]
100 assert data["file_count"] == 2
101
102 def test_json_flag_shorthand(self, tmp_path: pathlib.Path) -> None:
103 repo = _make_repo(tmp_path)
104 sid = _snap(repo, {"a.py": _fake_oid(1)})
105 result = _rs(repo, "--json", sid)
106 assert result.exit_code == 0
107 assert "snapshot_id" in json.loads(result.output)
108
109 def test_created_at_iso8601(self, tmp_path: pathlib.Path) -> None:
110 repo = _make_repo(tmp_path)
111 sid = _snap(repo, {"b.py": _fake_oid(2)})
112 data = json.loads(_rsj(repo, sid).output)
113 datetime.datetime.fromisoformat(data["created_at"])
114
115 def test_file_count_reflects_manifest(self, tmp_path: pathlib.Path) -> None:
116 repo = _make_repo(tmp_path)
117 manifest = {f"file{i}.py": _fake_oid(i) for i in range(5)}
118 sid = _snap(repo, manifest)
119 data = json.loads(_rsj(repo, sid).output)
120 assert data["file_count"] == 5
121
122
123 # ---------------------------------------------------------------------------
124 # Integration — text format
125 # ---------------------------------------------------------------------------
126
127
128 class TestTextFormat:
129 def test_text_contains_prefix(self, tmp_path: pathlib.Path) -> None:
130 repo = _make_repo(tmp_path)
131 sid = _snap(repo, {"c.py": _fake_oid(3)})
132 result = _rs(repo, sid)
133 assert result.exit_code == 0
134 assert short_id(sid) in result.output
135
136 def test_text_contains_file_count(self, tmp_path: pathlib.Path) -> None:
137 repo = _make_repo(tmp_path)
138 sid = _snap(repo, {"a.py": _fake_oid(1), "b.py": _fake_oid(2)})
139 result = _rs(repo, sid)
140 assert "2 files" in result.output
141
142 def test_text_single_line(self, tmp_path: pathlib.Path) -> None:
143 repo = _make_repo(tmp_path)
144 sid = _snap(repo)
145 result = _rs(repo, sid)
146 lines = [l for l in result.output.splitlines() if l.strip()]
147 assert len(lines) == 1
148
149
150 # ---------------------------------------------------------------------------
151 # Integration — --no-manifest
152 # ---------------------------------------------------------------------------
153
154
155 class TestNoManifest:
156 def test_manifest_absent(self, tmp_path: pathlib.Path) -> None:
157 repo = _make_repo(tmp_path)
158 sid = _snap(repo, {"a.py": _fake_oid(1)})
159 data = json.loads(_rsj(repo, "--no-manifest", sid).output)
160 assert "manifest" not in data
161 assert data["file_count"] == 1
162
163 def test_snapshot_id_and_created_at_still_present(self, tmp_path: pathlib.Path) -> None:
164 repo = _make_repo(tmp_path)
165 sid = _snap(repo)
166 data = json.loads(_rsj(repo, "--no-manifest", sid).output)
167 assert data["snapshot_id"] == sid
168 assert "created_at" in data
169
170 def test_no_manifest_with_text_errors(self, tmp_path: pathlib.Path) -> None:
171 repo = _make_repo(tmp_path)
172 sid = _snap(repo)
173 result = _rs(repo, "--no-manifest", sid)
174 assert result.exit_code == ExitCode.USER_ERROR
175
176
177 # ---------------------------------------------------------------------------
178 # Integration — --path-prefix
179 # ---------------------------------------------------------------------------
180
181
182 class TestPathPrefix:
183 def test_prefix_filters_manifest(self, tmp_path: pathlib.Path) -> None:
184 repo = _make_repo(tmp_path)
185 sid = _snap(repo, {
186 "src/a.py": _fake_oid(1),
187 "src/b.py": _fake_oid(2),
188 "tests/c.py": _fake_oid(3),
189 })
190 data = json.loads(_rsj(repo, "--path-prefix", "src/", sid).output)
191 assert set(data["manifest"].keys()) == {"src/a.py", "src/b.py"}
192 assert data["file_count"] == 2
193
194 def test_prefix_no_match_returns_empty(self, tmp_path: pathlib.Path) -> None:
195 repo = _make_repo(tmp_path)
196 sid = _snap(repo, {"src/a.py": _fake_oid(1)})
197 data = json.loads(_rsj(repo, "--path-prefix", "docs/", sid).output)
198 assert data["manifest"] == {}
199 assert data["file_count"] == 0
200
201 def test_prefix_with_text_errors(self, tmp_path: pathlib.Path) -> None:
202 repo = _make_repo(tmp_path)
203 sid = _snap(repo)
204 result = _rs(repo, "--path-prefix", "src/", sid)
205 assert result.exit_code == ExitCode.USER_ERROR
206
207
208 # ---------------------------------------------------------------------------
209 # Error cases
210 # ---------------------------------------------------------------------------
211
212
213 class TestErrors:
214 def test_missing_snapshot_errors(self, tmp_path: pathlib.Path) -> None:
215 repo = _make_repo(tmp_path)
216 # Valid sha256: format but not present in the store — must get "not found".
217 result = _rs(repo, long_id("dead" + "beef" * 15))
218 assert result.exit_code == ExitCode.USER_ERROR
219
220 def test_invalid_snapshot_id_errors(self, tmp_path: pathlib.Path) -> None:
221 repo = _make_repo(tmp_path)
222 result = _rs(repo, "not-valid")
223 assert result.exit_code == ExitCode.USER_ERROR
224
225 def test_unknown_format_errors_argparse_rejects(self, tmp_path: pathlib.Path) -> None:
226 """--format flag no longer exists; argparse exits 2."""
227 repo = _make_repo(tmp_path)
228 sid = _snap(repo)
229 result = _rs(repo, "--format", "msgpack", sid)
230 assert result.exit_code != 0 # argparse rejects unknown flag with exit 2
231
232
233 # ---------------------------------------------------------------------------
234 # Security
235 # ---------------------------------------------------------------------------
236
237
238 class TestSecurity:
239 def test_ansi_in_snapshot_id_rejected(self, tmp_path: pathlib.Path) -> None:
240 repo = _make_repo(tmp_path)
241 result = _rs(repo, "\x1b[31m" + "a" * 64)
242 assert result.exit_code == ExitCode.USER_ERROR
243
244 def test_no_traceback_on_bad_id(self, tmp_path: pathlib.Path) -> None:
245 repo = _make_repo(tmp_path)
246 result = _rs(repo, "bad-id")
247 assert "Traceback" not in result.output
248
249
250 # ---------------------------------------------------------------------------
251 # Stress
252 # ---------------------------------------------------------------------------
253
254
255 class TestStress:
256 def test_1000_file_manifest(self, tmp_path: pathlib.Path) -> None:
257 repo = _make_repo(tmp_path)
258 manifest = {f"src/module{i:04d}.py": _fake_oid(i) for i in range(1000)}
259 sid = _snap(repo, manifest)
260 result = _rsj(repo, sid)
261 assert result.exit_code == 0
262 data = json.loads(result.output)
263 assert data["file_count"] == 1000
264 assert len(data["manifest"]) == 1000
265
266 def test_1000_file_manifest_no_manifest(self, tmp_path: pathlib.Path) -> None:
267 repo = _make_repo(tmp_path)
268 manifest = {f"src/module{i:04d}.py": _fake_oid(i) for i in range(1000)}
269 sid = _snap(repo, manifest)
270 result = _rsj(repo, "--no-manifest", sid)
271 assert result.exit_code == 0
272 data = json.loads(result.output)
273 assert data["file_count"] == 1000
274 assert "manifest" not in data
275
276 def test_200_sequential_reads(self, tmp_path: pathlib.Path) -> None:
277 repo = _make_repo(tmp_path)
278 sid = _snap(repo, {"a.py": _fake_oid(0)})
279 for i in range(200):
280 result = _rsj(repo, sid)
281 assert result.exit_code == 0, f"failed at iteration {i}"
282 data = json.loads(result.output)
283 assert data["file_count"] == 1
284
285
286 class TestRegisterFlags:
287 def _parse(self, *args: str) -> "argparse.Namespace":
288 import argparse
289 from muse.cli.commands.read_snapshot import register
290 p = argparse.ArgumentParser()
291 subs = p.add_subparsers()
292 register(subs)
293 return p.parse_args(["read-snapshot", fake_id("a"), *args])
294
295 def test_json_short_flag(self):
296 args = self._parse("-j")
297 assert args.json_out is True
298
299 def test_json_long_flag(self):
300 args = self._parse("--json")
301 assert args.json_out is True
302
303 def test_default_no_json(self):
304 args = self._parse()
305 assert args.json_out is False
File History 3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 140 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 146 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 149 days ago