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