gabriel / muse public
test_read_snapshot_supercharge.py python
422 lines 16.6 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 122 days ago
1 """Supercharge tests for ``muse read-snapshot``.
2
3 Coverage tiers
4 --------------
5 - Unit: _short_id helper — prefix preservation, hex length
6 - Integration: duration_ms + exit_code in JSON; text short-ID format
7 - Flag interaction: --no-manifest + --path-prefix together
8 - Data integrity: sha256: on snapshot_id; valid JSON; unicode paths
9 - Path prefix edge cases: empty prefix, no trailing slash, exact filename
10 - Performance: single read and 1000-file manifest under threshold
11 """
12 from __future__ import annotations
13 from collections.abc import Mapping
14
15 import datetime
16 import json
17 import pathlib
18 import re
19 import time
20
21 from muse.core.errors import ExitCode
22 from muse.core.snapshot import compute_snapshot_id
23 from muse.core.store import SnapshotRecord, write_snapshot
24 from tests.cli_test_helper import CliRunner, InvokeResult
25 from muse.core.types import NULL_LONG_ID, fake_id, long_id, split_id
26 from muse.core.paths import muse_dir
27
28 runner = CliRunner()
29
30 _CREATED_AT = datetime.datetime(2026, 3, 18, 12, 0, tzinfo=datetime.timezone.utc)
31
32 _SHA256_FULL = re.compile(r"^sha256:[0-9a-f]{64}$")
33 _SHA256_SHORT_19 = re.compile(r"^sha256:[0-9a-f]{12}$")
34
35
36 # ---------------------------------------------------------------------------
37 # Helpers
38 # ---------------------------------------------------------------------------
39
40
41 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
42 repo = tmp_path / "repo"
43 dot_muse = muse_dir(repo)
44 for sub in ("objects", "commits", "snapshots", "refs/heads"):
45 (dot_muse / sub).mkdir(parents=True)
46 (dot_muse / "HEAD").write_text("ref: refs/heads/main")
47 (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"}))
48 return repo
49
50
51 def _snap(repo: pathlib.Path, manifest: Mapping[str, object] | None = None) -> str:
52 m = manifest or {}
53 sid = compute_snapshot_id(m)
54 write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest=m, created_at=_CREATED_AT))
55 return sid
56
57
58 def _rs(repo: pathlib.Path, *args: str) -> InvokeResult:
59 from muse.cli.app import main as cli
60 return runner.invoke(cli, ["read-snapshot", *args], env={"MUSE_REPO_ROOT": str(repo)})
61
62
63 def _rsj(repo: pathlib.Path, *args: str) -> InvokeResult:
64 """Like _rs but always passes --json for JSON-output tests."""
65 return _rs(repo, "--json", *args)
66
67
68 def _oid(n: int) -> str:
69 """Canonical sha256:-prefixed object ID for test manifests."""
70 return long_id(format(n, "064x"))
71
72
73 # ---------------------------------------------------------------------------
74 # Unit — _short_id
75 # ---------------------------------------------------------------------------
76
77
78 class TestSnapshotIdFormat:
79 """Snapshot IDs are full sha256:<64-hex> (71 chars)."""
80
81 def test_snapshot_id_keeps_sha256_prefix(self) -> None:
82 sid = long_id("a" * 64)
83 assert sid.startswith("sha256:")
84
85 def test_snapshot_id_total_length_is_71(self) -> None:
86 sid = NULL_LONG_ID
87 assert len(sid) == 71
88
89 def test_snapshot_id_matches_full_regex(self) -> None:
90 sid = long_id("abcdef01" * 8)
91 assert _SHA256_FULL.match(sid)
92
93
94 # ---------------------------------------------------------------------------
95 # Integration — duration_ms and exit_code
96 # ---------------------------------------------------------------------------
97
98
99 class TestDurationAndExitCode:
100 def test_duration_ms_present_on_success(self, tmp_path: pathlib.Path) -> None:
101 repo = _make_repo(tmp_path)
102 sid = _snap(repo)
103 data = json.loads(_rsj(repo, sid).output)
104 assert "duration_ms" in data, "duration_ms must be present in JSON output"
105
106 def test_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None:
107 repo = _make_repo(tmp_path)
108 sid = _snap(repo)
109 data = json.loads(_rsj(repo, sid).output)
110 assert data["exit_code"] == 0
111
112 def test_duration_ms_is_float(self, tmp_path: pathlib.Path) -> None:
113 repo = _make_repo(tmp_path)
114 sid = _snap(repo)
115 data = json.loads(_rsj(repo, sid).output)
116 assert isinstance(data["duration_ms"], float)
117
118 def test_duration_ms_non_negative(self, tmp_path: pathlib.Path) -> None:
119 repo = _make_repo(tmp_path)
120 sid = _snap(repo)
121 data = json.loads(_rsj(repo, sid).output)
122 assert data["duration_ms"] >= 0.0
123
124 def test_duration_ms_3dp_precision(self, tmp_path: pathlib.Path) -> None:
125 repo = _make_repo(tmp_path)
126 sid = _snap(repo)
127 ms = json.loads(_rsj(repo, sid).output)["duration_ms"]
128 assert round(ms, 3) == ms
129
130 def test_duration_ms_present_with_no_manifest(self, tmp_path: pathlib.Path) -> None:
131 repo = _make_repo(tmp_path)
132 sid = _snap(repo, {"a.py": _oid(1)})
133 data = json.loads(_rsj(repo, "--no-manifest", sid).output)
134 assert "duration_ms" in data
135 assert "exit_code" in data
136
137 def test_duration_ms_present_with_path_prefix(self, tmp_path: pathlib.Path) -> None:
138 repo = _make_repo(tmp_path)
139 sid = _snap(repo, {"src/a.py": _oid(1), "tests/b.py": _oid(2)})
140 data = json.loads(_rsj(repo, "--path-prefix", "src/", sid).output)
141 assert "duration_ms" in data
142 assert data["exit_code"] == 0
143
144
145 # ---------------------------------------------------------------------------
146 # Integration — text format short ID
147 # ---------------------------------------------------------------------------
148
149
150 class TestTextFormatFullId:
151 """Text format must emit the full sha256:<64-hex> (71 chars) snapshot ID."""
152
153 def _full_token(self, line: str) -> str | None:
154 for tok in line.split():
155 if _SHA256_FULL.match(tok):
156 return tok
157 return None
158
159 def test_text_full_id_has_sha256_prefix(self, tmp_path: pathlib.Path) -> None:
160 repo = _make_repo(tmp_path)
161 sid = _snap(repo, {"f.py": _oid(1)})
162 result = _rs(repo, sid)
163 assert result.exit_code == 0
164 tok = self._full_token(result.output.strip())
165 assert tok is not None, f"no sha256:<64-hex> token in: {result.output!r}"
166 assert tok.startswith("sha256:")
167
168 def test_text_full_id_has_64_hex_chars(self, tmp_path: pathlib.Path) -> None:
169 repo = _make_repo(tmp_path)
170 sid = _snap(repo)
171 result = _rs(repo, sid)
172 tok = self._full_token(result.output.strip())
173 assert tok is not None
174 assert tok.startswith("sha256:")
175 assert len(tok[len("sha256:"):]) == 64
176
177 def test_text_full_id_total_length_is_71(self, tmp_path: pathlib.Path) -> None:
178 repo = _make_repo(tmp_path)
179 sid = _snap(repo)
180 result = _rs(repo, sid)
181 tok = self._full_token(result.output.strip())
182 assert tok is not None
183 assert len(tok) == 71
184
185 def test_text_full_id_matches_snapshot_id(self, tmp_path: pathlib.Path) -> None:
186 repo = _make_repo(tmp_path)
187 sid = _snap(repo, {"x.py": _oid(9)})
188 result = _rs(repo, sid)
189 tok = self._full_token(result.output.strip())
190 assert tok is not None
191 assert tok == sid, f"text output ID {tok!r} does not match snapshot_id {sid!r}"
192
193
194 # ---------------------------------------------------------------------------
195 # Flag interaction — --no-manifest + --path-prefix together
196 # ---------------------------------------------------------------------------
197
198
199 class TestFlagInteraction:
200 """--no-manifest and --path-prefix may be combined.
201
202 Use case: "how many files are under src/ without downloading any OIDs?"
203 The file_count reflects the filtered count; manifest is omitted.
204 """
205
206 def test_no_manifest_plus_path_prefix_succeeds(self, tmp_path: pathlib.Path) -> None:
207 repo = _make_repo(tmp_path)
208 sid = _snap(repo, {
209 "src/a.py": _oid(1),
210 "src/b.py": _oid(2),
211 "tests/c.py": _oid(3),
212 })
213 result = _rsj(repo, "--no-manifest", "--path-prefix", "src/", sid)
214 assert result.exit_code == 0, result.output
215
216 def test_no_manifest_plus_path_prefix_file_count_is_filtered(self, tmp_path: pathlib.Path) -> None:
217 repo = _make_repo(tmp_path)
218 sid = _snap(repo, {
219 "src/a.py": _oid(1),
220 "src/b.py": _oid(2),
221 "tests/c.py": _oid(3),
222 })
223 data = json.loads(_rsj(repo, "--no-manifest", "--path-prefix", "src/", sid).output)
224 assert data["file_count"] == 2, "file_count must reflect the prefix-filtered count"
225
226 def test_no_manifest_plus_path_prefix_manifest_absent(self, tmp_path: pathlib.Path) -> None:
227 repo = _make_repo(tmp_path)
228 sid = _snap(repo, {"src/a.py": _oid(1)})
229 data = json.loads(_rsj(repo, "--no-manifest", "--path-prefix", "src/", sid).output)
230 assert "manifest" not in data
231
232 def test_no_manifest_plus_path_prefix_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
233 repo = _make_repo(tmp_path)
234 sid = _snap(repo, {"src/a.py": _oid(1)})
235 data = json.loads(_rsj(repo, "--no-manifest", "--path-prefix", "src/", sid).output)
236 assert "duration_ms" in data
237 assert data["exit_code"] == 0
238
239
240 # ---------------------------------------------------------------------------
241 # Data integrity
242 # ---------------------------------------------------------------------------
243
244
245 class TestDataIntegrity:
246 def test_snapshot_id_has_sha256_prefix(self, tmp_path: pathlib.Path) -> None:
247 repo = _make_repo(tmp_path)
248 sid = _snap(repo)
249 data = json.loads(_rsj(repo, sid).output)
250 assert _SHA256_FULL.match(data["snapshot_id"]), \
251 f"snapshot_id must be sha256:<64hex>, got {data['snapshot_id']!r}"
252
253 def test_json_output_is_valid_json(self, tmp_path: pathlib.Path) -> None:
254 repo = _make_repo(tmp_path)
255 sid = _snap(repo, {"a.py": _oid(1)})
256 result = _rsj(repo, sid)
257 assert result.exit_code == 0
258 data = json.loads(result.output)
259 assert isinstance(data, dict)
260
261 def test_manifest_values_are_strings(self, tmp_path: pathlib.Path) -> None:
262 """Manifest object IDs are strings — no type coercion."""
263 repo = _make_repo(tmp_path)
264 sid = _snap(repo, {"a.py": _oid(1), "b.py": _oid(2)})
265 data = json.loads(_rsj(repo, sid).output)
266 for path, oid in data["manifest"].items():
267 assert isinstance(oid, str), f"manifest[{path!r}] must be a string, got {type(oid)}"
268
269 def test_unicode_paths_in_manifest(self, tmp_path: pathlib.Path) -> None:
270 """Unicode file paths round-trip through JSON without corruption."""
271 repo = _make_repo(tmp_path)
272 paths = {
273 "src/音楽.py": _oid(1),
274 "tracks/café/main.mid": _oid(2),
275 "docs/naïve_approach.md": _oid(3),
276 }
277 sid = _snap(repo, paths)
278 data = json.loads(_rsj(repo, sid).output)
279 assert data["file_count"] == 3
280 for p in paths:
281 assert p in data["manifest"], f"unicode path {p!r} missing from manifest"
282
283 def test_created_at_iso8601_with_timezone(self, tmp_path: pathlib.Path) -> None:
284 repo = _make_repo(tmp_path)
285 sid = _snap(repo)
286 data = json.loads(_rsj(repo, sid).output)
287 dt = datetime.datetime.fromisoformat(data["created_at"])
288 assert dt.tzinfo is not None, "created_at must include timezone"
289
290 def test_file_count_matches_manifest_length(self, tmp_path: pathlib.Path) -> None:
291 """file_count must equal len(manifest) in the response."""
292 repo = _make_repo(tmp_path)
293 n = 17
294 sid = _snap(repo, {f"f{i}.py": _oid(i) for i in range(n)})
295 data = json.loads(_rsj(repo, sid).output)
296 assert data["file_count"] == n
297 assert len(data["manifest"]) == n
298
299
300 # ---------------------------------------------------------------------------
301 # Path prefix edge cases
302 # ---------------------------------------------------------------------------
303
304
305 class TestPathPrefixEdgeCases:
306 def test_empty_prefix_matches_all(self, tmp_path: pathlib.Path) -> None:
307 """Empty --path-prefix matches every path (prefix of every string)."""
308 repo = _make_repo(tmp_path)
309 sid = _snap(repo, {"src/a.py": _oid(1), "tests/b.py": _oid(2)})
310 data = json.loads(_rsj(repo, "--path-prefix", "", sid).output)
311 assert data["file_count"] == 2
312
313 def test_prefix_without_trailing_slash(self, tmp_path: pathlib.Path) -> None:
314 """Prefix 'src' (no slash) matches 'src/a.py' and also 'src_util.py'."""
315 repo = _make_repo(tmp_path)
316 sid = _snap(repo, {
317 "src/a.py": _oid(1),
318 "src_util.py": _oid(2),
319 "tests/b.py": _oid(3),
320 })
321 data = json.loads(_rsj(repo, "--path-prefix", "src", sid).output)
322 assert "src/a.py" in data["manifest"]
323 assert "src_util.py" in data["manifest"]
324 assert "tests/b.py" not in data["manifest"]
325
326 def test_prefix_exact_filename_match(self, tmp_path: pathlib.Path) -> None:
327 """A prefix equal to an exact filename matches only that file."""
328 repo = _make_repo(tmp_path)
329 sid = _snap(repo, {"README.md": _oid(1), "README.md.bak": _oid(2)})
330 data = json.loads(_rsj(repo, "--path-prefix", "README.md", sid).output)
331 assert "README.md" in data["manifest"]
332 assert "README.md.bak" in data["manifest"] # startswith matches both
333
334 def test_prefix_no_match_empty_manifest_with_duration(self, tmp_path: pathlib.Path) -> None:
335 """No-match prefix returns empty manifest with duration_ms."""
336 repo = _make_repo(tmp_path)
337 sid = _snap(repo, {"src/a.py": _oid(1)})
338 data = json.loads(_rsj(repo, "--path-prefix", "nonexistent/", sid).output)
339 assert data["file_count"] == 0
340 assert data["manifest"] == {}
341 assert "duration_ms" in data
342
343
344 # ---------------------------------------------------------------------------
345 # Security
346 # ---------------------------------------------------------------------------
347
348
349 class TestSecuritySupercharge:
350 def test_path_prefix_with_traversal_attempt(self, tmp_path: pathlib.Path) -> None:
351 """Path prefix with '../' traversal must not escape manifest keys."""
352 repo = _make_repo(tmp_path)
353 sid = _snap(repo, {"src/a.py": _oid(1), "../etc/passwd": _oid(2)})
354 # The manifest key itself is literally '../etc/passwd' — filter should match it
355 # only if the prefix is '../', not silently escape the repo root
356 data = json.loads(_rsj(repo, "--path-prefix", "src/", sid).output)
357 # Only src/a.py should match src/ prefix
358 assert "src/a.py" in data["manifest"]
359 assert "../etc/passwd" not in data["manifest"]
360
361 def test_no_traceback_on_sha256_prefixed_missing_id(self, tmp_path: pathlib.Path) -> None:
362 """Valid sha256: format but non-existent ID — no traceback, clean error."""
363 repo = _make_repo(tmp_path)
364 missing = long_id("dead" * 16)
365 result = _rs(repo, missing)
366 assert result.exit_code == ExitCode.USER_ERROR
367 assert "Traceback" not in result.output
368
369
370 # ---------------------------------------------------------------------------
371 # Performance
372 # ---------------------------------------------------------------------------
373
374
375 class TestPerformanceSupercharge:
376 def test_single_read_under_500ms(self, tmp_path: pathlib.Path) -> None:
377 repo = _make_repo(tmp_path)
378 sid = _snap(repo, {"a.py": _oid(0)})
379 t0 = time.monotonic()
380 result = _rs(repo, sid)
381 duration_ms = (time.monotonic() - t0) * 1000
382 assert result.exit_code == 0
383 assert duration_ms < 500
384
385 def test_1000_file_manifest_under_1000ms(self, tmp_path: pathlib.Path) -> None:
386 repo = _make_repo(tmp_path)
387 manifest = {f"src/module{i:04d}.py": _oid(i) for i in range(1000)}
388 sid = _snap(repo, manifest)
389 t0 = time.monotonic()
390 result = _rs(repo, sid)
391 duration_ms = (time.monotonic() - t0) * 1000
392 assert result.exit_code == 0
393 assert duration_ms < 1000
394
395 def test_duration_ms_plausible(self, tmp_path: pathlib.Path) -> None:
396 """duration_ms from the output itself must be < 500ms for a warm read."""
397 repo = _make_repo(tmp_path)
398 sid = _snap(repo, {"a.py": _oid(0)})
399 data = json.loads(_rsj(repo, sid).output)
400 assert data["duration_ms"] < 500
401
402
403 class TestRegisterFlags:
404 def _parse(self, *args: str) -> "argparse.Namespace":
405 import argparse
406 from muse.cli.commands.read_snapshot import register
407 p = argparse.ArgumentParser()
408 subs = p.add_subparsers()
409 register(subs)
410 return p.parse_args(["read-snapshot", fake_id("a"), *args])
411
412 def test_json_short_flag(self) -> None:
413 args = self._parse("-j")
414 assert args.json_out is True
415
416 def test_json_long_flag(self) -> None:
417 args = self._parse("--json")
418 assert args.json_out is True
419
420 def test_default_no_json(self) -> None:
421 args = self._parse()
422 assert args.json_out is False
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 122 days ago