gabriel / muse public
test_mpack_cmd_verify.py python
650 lines 25.3 KB
Raw
sha256:a154bc65916614c833d5a40a10d81ba3eae0d0495b0afddd34dc34f18d5e91b8 fix: test suite alignment and typing audit — zero violations Sonnet 4.6 minor ⚠ breaking 22 days ago
1 """Comprehensive tests for muse verify-pack.
2
3 Coverage:
4 - Unit: TypedDict schemas, _Failure, _VerifyPackResult, _StatResult
5 - Integration: JSON/text formats, --stat, --quiet, --no-local, --file, --json shorthand
6 - Verification: object integrity, snapshot consistency, commit consistency
7 - Security: ANSI sanitization, format error → stderr, no tracebacks, invalid object IDs
8 - Stress: 500-object mpack, 200 sequential verifications, large object hashing
9 """
10
11 from __future__ import annotations
12 from collections.abc import Mapping
13
14 import datetime
15 import io
16 import json
17 import pathlib
18
19 import msgpack
20 import pytest
21 from tests.cli_test_helper import CliRunner, InvokeResult
22
23 from muse.cli.commands.verify_pack import (
24 _Failure,
25 _StatResultJson as _StatResult,
26 _VerifyPackJson as _VerifyPackResult,
27 )
28 from muse.core.commits import (
29 CommitRecord,
30 write_commit,
31 )
32 from muse.core.snapshots import (
33 SnapshotRecord,
34 write_snapshot,
35 )
36 from muse.core.types import Manifest, blob_id, long_id, fake_id
37 from muse.core.paths import muse_dir
38 from typing import TypedDict
39
40 runner = CliRunner()
41
42
43 # ---------------------------------------------------------------------------
44 # MPack TypedDicts (mirror the output contract shape)
45 # ---------------------------------------------------------------------------
46
47 class _ObjectEntry(TypedDict):
48 object_id: str
49 content: bytes
50
51
52 class _SnapshotEntry(TypedDict, total=False):
53 snapshot_id: str
54 manifest: Manifest
55
56
57 class _CommitEntry(TypedDict):
58 commit_id: str
59 snapshot_id: str
60
61
62 type _EnvMap = dict[str, str]
63 type _CommitEntryDict = dict[str, str]
64 type _BadObjectEntry = dict[str, str | bytes]
65 type _BundleDict = dict[str, list[_ObjectEntry] | list[_SnapshotEntry] | list[_CommitEntry]]
66
67
68 # ---------------------------------------------------------------------------
69 # Helpers
70 # ---------------------------------------------------------------------------
71
72 def _init_repo(path: pathlib.Path) -> pathlib.Path:
73 dot_muse = muse_dir(path)
74 (dot_muse / "commits").mkdir(parents=True, exist_ok=True)
75 (dot_muse / "snapshots").mkdir(parents=True, exist_ok=True)
76 (dot_muse / "objects" / "ab").mkdir(parents=True, exist_ok=True)
77 (dot_muse / "refs" / "heads").mkdir(parents=True, exist_ok=True)
78 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
79 (dot_muse / "repo.json").write_text(
80 json.dumps({"repo_id": "test-repo", "domain": "generic"}), encoding="utf-8"
81 )
82 return path
83
84
85 def _env(repo: pathlib.Path) -> _EnvMap:
86 return {"MUSE_REPO_ROOT": str(repo)}
87
88
89
90
91 def _good_obj(data: bytes = b"hello world") -> _ObjectEntry:
92 return _ObjectEntry(object_id=blob_id(data), content=data)
93
94
95 def _bad_hash_obj(data: bytes = b"hello world") -> _ObjectEntry:
96 return _ObjectEntry(object_id=fake_id("bad-hash"), content=data)
97
98
99 _FULL_META = {
100 "mode": "full",
101 "base_commits": [],
102 "created_at": "2025-01-01T00:00:00Z",
103 }
104
105
106 def _make_bundle(
107 objects: list[_ObjectEntry] | None = None,
108 snapshots: list[_SnapshotEntry] | None = None,
109 commits: list[_CommitEntry] | None = None,
110 meta: Mapping[str, object] | None = None,
111 ) -> bytes:
112 mpack: _BundleDict = {
113 "meta": meta if meta is not None else _FULL_META,
114 "blobs": objects or [],
115 "snapshots": snapshots or [],
116 "commits": commits or [],
117 }
118 return msgpack.packb(mpack, use_bin_type=True)
119
120
121 def _vp(
122 tmp_path: pathlib.Path,
123 args: list[str],
124 stdin: bytes | None = None,
125 ) -> InvokeResult:
126 """Invoke verify-pack against a freshly initialised repo in tmp_path."""
127 from muse.cli.app import main as cli
128 _init_repo(tmp_path)
129 return runner.invoke(
130 cli,
131 ["verify-pack"] + args,
132 env=_env(tmp_path),
133 input=stdin,
134 )
135
136
137 # ---------------------------------------------------------------------------
138 # Unit: schemas
139 # ---------------------------------------------------------------------------
140
141 class TestSchemas:
142 def test_failure_fields(self) -> None:
143 f: _Failure = {"kind": "object", "id": "abc", "error": "hash mismatch"}
144 assert f["kind"] == "object"
145 assert f["id"] == "abc"
146 assert f["error"] == "hash mismatch"
147
148 def test_verify_pack_result_fields(self) -> None:
149 r: _VerifyPackResult = {
150 "blobs_checked": 3,
151 "snapshots_checked": 2,
152 "commits_checked": 1,
153 "all_ok": True,
154 "failures": [],
155 }
156 assert r["all_ok"] is True
157 assert isinstance(r["failures"], list)
158
159 def test_stat_result_fields(self) -> None:
160 s: _StatResult = {"blobs": 10, "snapshots": 5, "commits": 3}
161 assert s["blobs"] == 10
162
163
164 # ---------------------------------------------------------------------------
165 # Integration: JSON output — clean bundles
166 # ---------------------------------------------------------------------------
167
168 class TestJsonOutputClean:
169 def test_empty_bundle(self, tmp_path: pathlib.Path) -> None:
170 bf = tmp_path / "b.muse"
171 bf.write_bytes(_make_bundle())
172 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
173 assert r.exit_code == 0
174 d = json.loads(r.output)
175 assert d["all_ok"] is True
176 assert d["failures"] == []
177 assert d["blobs_checked"] == 0
178 assert d["snapshots_checked"] == 0
179 assert d["commits_checked"] == 0
180
181 def test_single_good_object(self, tmp_path: pathlib.Path) -> None:
182 bf = tmp_path / "b.muse"
183 bf.write_bytes(_make_bundle([_good_obj()]))
184 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
185 assert r.exit_code == 0
186 d = json.loads(r.output)
187 assert d["all_ok"] is True
188 assert d["blobs_checked"] == 1
189
190 def test_json_shorthand(self, tmp_path: pathlib.Path) -> None:
191 bf = tmp_path / "b.muse"
192 bf.write_bytes(_make_bundle())
193 r = _vp(tmp_path, ["--file", str(bf), "--json", "--no-local"])
194 assert r.exit_code == 0
195 json.loads(r.output) # must be valid JSON
196
197 def test_multiple_good_objects(self, tmp_path: pathlib.Path) -> None:
198 objs = [_good_obj(f"content-{i}".encode()) for i in range(10)]
199 bf = tmp_path / "b.muse"
200 bf.write_bytes(_make_bundle(objs))
201 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
202 assert r.exit_code == 0
203 d = json.loads(r.output)
204 assert d["blobs_checked"] == 10
205 assert d["all_ok"] is True
206
207
208 # ---------------------------------------------------------------------------
209 # Integration: JSON output — hash integrity failures
210 # ---------------------------------------------------------------------------
211
212 class TestObjectIntegrity:
213 def test_bad_hash_detected(self, tmp_path: pathlib.Path) -> None:
214 bf = tmp_path / "b.muse"
215 bf.write_bytes(_make_bundle([_bad_hash_obj()]))
216 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
217 assert r.exit_code != 0
218 d = json.loads(r.output)
219 assert d["all_ok"] is False
220 assert d["failures"][0]["kind"] == "object"
221 assert "hash mismatch" in d["failures"][0]["error"]
222
223 def test_mix_good_and_bad(self, tmp_path: pathlib.Path) -> None:
224 objs = [_good_obj(b"good"), _bad_hash_obj(b"bad")]
225 bf = tmp_path / "b.muse"
226 bf.write_bytes(_make_bundle(objs))
227 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
228 assert r.exit_code != 0
229 d = json.loads(r.output)
230 assert d["blobs_checked"] == 2
231 assert len(d["failures"]) == 1
232
233 def test_invalid_object_id_format(self, tmp_path: pathlib.Path) -> None:
234 """MPack with non-hex object ID should report a specific format error."""
235 bad: _BadObjectEntry = {
236 "object_id": f"\x1b[31mmalicious\x1b[0m{'a' * 55}",
237 "content": b"data",
238 }
239 bf = tmp_path / "b.muse"
240 bf.write_bytes(_make_bundle([bad]))
241 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
242 assert r.exit_code != 0
243 d = json.loads(r.output)
244 assert d["all_ok"] is False
245 assert "not a valid" in d["failures"][0]["error"].lower() or "sha-256" in d["failures"][0]["error"].lower()
246
247 def test_entry_not_dict(self, tmp_path: pathlib.Path) -> None:
248 mpack = {
249 "meta": _FULL_META,
250 "blobs": ["not-a-dict"],
251 "snapshots": [],
252 "commits": [],
253 }
254 bf = tmp_path / "b.muse"
255 bf.write_bytes(msgpack.packb(mpack, use_bin_type=True))
256 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
257 d = json.loads(r.output)
258 assert d["all_ok"] is False
259 assert "not a dict" in d["failures"][0]["error"]
260
261 def test_missing_content_field(self, tmp_path: pathlib.Path) -> None:
262 entry: _CommitEntryDict = {"object_id": "a" * 64} # no content
263 bf = tmp_path / "b.muse"
264 bf.write_bytes(_make_bundle([entry]))
265 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
266 d = json.loads(r.output)
267 assert d["all_ok"] is False
268
269 def test_empty_object_passes(self, tmp_path: pathlib.Path) -> None:
270 """An empty bytes object is valid — its SHA-256 is known."""
271 obj = _good_obj(b"")
272 bf = tmp_path / "b.muse"
273 bf.write_bytes(_make_bundle([obj]))
274 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
275 d = json.loads(r.output)
276 assert d["all_ok"] is True
277
278 def test_binary_object_passes(self, tmp_path: pathlib.Path) -> None:
279 obj = _good_obj(bytes(range(256)))
280 bf = tmp_path / "b.muse"
281 bf.write_bytes(_make_bundle([obj]))
282 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
283 d = json.loads(r.output)
284 assert d["all_ok"] is True
285
286
287 # ---------------------------------------------------------------------------
288 # Integration: snapshot consistency
289 # ---------------------------------------------------------------------------
290
291 class TestSnapshotConsistency:
292 def _snap_entry(self, snap_id: str, manifest: Manifest) -> _SnapshotEntry:
293 return _SnapshotEntry(snapshot_id=snap_id, manifest=manifest)
294
295 def test_snapshot_with_all_objects_present(self, tmp_path: pathlib.Path) -> None:
296 obj = _good_obj(b"snap-obj")
297 snap = self._snap_entry(blob_id(b"snap"), {"file.txt": obj["object_id"]})
298 bf = tmp_path / "b.muse"
299 bf.write_bytes(_make_bundle(objects=[obj], snapshots=[snap]))
300 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
301 d = json.loads(r.output)
302 assert d["all_ok"] is True
303 assert d["snapshots_checked"] == 1
304
305 def test_snapshot_missing_object_fails(self, tmp_path: pathlib.Path) -> None:
306 missing_oid = "b" * 64
307 snap = self._snap_entry(blob_id(b"snap"), {"file.txt": missing_oid})
308 bf = tmp_path / "b.muse"
309 bf.write_bytes(_make_bundle(snapshots=[snap]))
310 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
311 d = json.loads(r.output)
312 assert d["all_ok"] is False
313 assert d["failures"][0]["kind"] == "snapshot"
314 assert "missing object" in d["failures"][0]["error"]
315
316 def test_snapshot_entry_not_dict(self, tmp_path: pathlib.Path) -> None:
317 mpack = {
318 "meta": _FULL_META,
319 "blobs": [],
320 "snapshots": ["bad"],
321 "commits": [],
322 }
323 bf = tmp_path / "b.muse"
324 bf.write_bytes(msgpack.packb(mpack, use_bin_type=True))
325 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
326 d = json.loads(r.output)
327 assert d["all_ok"] is False
328
329 def test_skip_local_check_allows_missing(self, tmp_path: pathlib.Path) -> None:
330 """--no-local should NOT check the local store; missing objects pass."""
331 # With --no-local, snapshot references a missing object but no local check.
332 missing_oid = "c" * 64
333 snap = self._snap_entry(blob_id(b"snap"), {"f.txt": missing_oid})
334 bf = tmp_path / "b.muse"
335 bf.write_bytes(_make_bundle(snapshots=[snap]))
336 # Without --no-local, this fails (missing locally).
337 r_fail = _vp(tmp_path, ["--file", str(bf)])
338 assert r_fail.exit_code != 0
339 # With --no-local, this fails too (not in mpack either).
340 r_nol = _vp(tmp_path, ["--file", str(bf), "--no-local"])
341 assert r_nol.exit_code != 0
342
343
344 # ---------------------------------------------------------------------------
345 # Integration: commit consistency
346 # ---------------------------------------------------------------------------
347
348 class TestCommitConsistency:
349 def _commit_entry(self, commit_id: str, snapshot_id: str) -> _CommitEntryDict:
350 return {"commit_id": commit_id, "snapshot_id": snapshot_id}
351
352 def test_commit_with_bundled_snapshot(self, tmp_path: pathlib.Path) -> None:
353 snap_id = blob_id(b"snap-data")
354 snap = {"snapshot_id": snap_id, "manifest": {}}
355 commit = self._commit_entry(blob_id(b"c1"), snap_id)
356 bf = tmp_path / "b.muse"
357 bf.write_bytes(_make_bundle(snapshots=[snap], commits=[commit]))
358 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
359 d = json.loads(r.output)
360 assert d["all_ok"] is True
361 assert d["commits_checked"] == 1
362
363 def test_commit_missing_snapshot_fails(self, tmp_path: pathlib.Path) -> None:
364 # Default (local store enabled): snapshot not in mpack → failure
365 commit = self._commit_entry(blob_id(b"c1"), fake_id("missing-snapshot"))
366 bf = tmp_path / "b.muse"
367 bf.write_bytes(_make_bundle(commits=[commit]))
368 r = _vp(tmp_path, ["--file", str(bf), "--json"]) # local check on by default
369 d = json.loads(r.output)
370 assert d["all_ok"] is False
371 assert d["failures"][0]["kind"] == "commit"
372
373 def test_commit_entry_not_dict(self, tmp_path: pathlib.Path) -> None:
374 mpack = {
375 "meta": _FULL_META,
376 "blobs": [],
377 "snapshots": [],
378 "commits": [42],
379 }
380 bf = tmp_path / "b.muse"
381 bf.write_bytes(msgpack.packb(mpack, use_bin_type=True))
382 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
383 d = json.loads(r.output)
384 assert d["all_ok"] is False
385
386
387 # ---------------------------------------------------------------------------
388 # Integration: text format
389 # ---------------------------------------------------------------------------
390
391 class TestTextFormat:
392 def test_text_format_clean(self, tmp_path: pathlib.Path) -> None:
393 bf = tmp_path / "b.muse"
394 bf.write_bytes(_make_bundle())
395 r = _vp(tmp_path, ["--file", str(bf), "--no-local"])
396 assert r.exit_code == 0
397 assert "all_ok=True" in r.output
398
399 def test_text_format_failure_shows_fail(self, tmp_path: pathlib.Path) -> None:
400 bf = tmp_path / "b.muse"
401 bf.write_bytes(_make_bundle([_bad_hash_obj()]))
402 r = _vp(tmp_path, ["--file", str(bf), "--no-local"])
403 assert r.exit_code != 0
404 assert "FAIL" in r.output
405
406 def test_text_format_counts(self, tmp_path: pathlib.Path) -> None:
407 objs = [_good_obj(f"x{i}".encode()) for i in range(5)]
408 bf = tmp_path / "b.muse"
409 bf.write_bytes(_make_bundle(objs))
410 r = _vp(tmp_path, ["--file", str(bf), "--no-local"])
411 assert "blobs=5" in r.output
412
413
414 # ---------------------------------------------------------------------------
415 # Integration: --stat fast-path
416 # ---------------------------------------------------------------------------
417
418 class TestStatMode:
419 def test_stat_json(self, tmp_path: pathlib.Path) -> None:
420 objs = [_good_obj(f"s{i}".encode()) for i in range(7)]
421 bf = tmp_path / "b.muse"
422 bf.write_bytes(_make_bundle(objs))
423 r = _vp(tmp_path, ["--file", str(bf), "--stat", "--json"])
424 assert r.exit_code == 0
425 d = json.loads(r.output)
426 assert d["blobs"] ==7
427 assert d["snapshots"] == 0
428 assert d["commits"] == 0
429
430 def test_stat_text(self, tmp_path: pathlib.Path) -> None:
431 bf = tmp_path / "b.muse"
432 bf.write_bytes(_make_bundle())
433 r = _vp(tmp_path, ["--file", str(bf), "--stat"])
434 assert r.exit_code == 0
435 assert "blobs=0" in r.output
436
437 def test_stat_does_not_verify_hashes(self, tmp_path: pathlib.Path) -> None:
438 """--stat exits 0 even with a corrupted mpack hash."""
439 bf = tmp_path / "b.muse"
440 bf.write_bytes(_make_bundle([_bad_hash_obj()]))
441 r = _vp(tmp_path, ["--file", str(bf), "--stat", "--json"])
442 assert r.exit_code == 0 # no hashing — passes
443 d = json.loads(r.output)
444 assert d["blobs"] ==1
445
446 def test_stat_counts_all_sections(self, tmp_path: pathlib.Path) -> None:
447 snap_id = blob_id(b"s")
448 snap = {"snapshot_id": snap_id, "manifest": {}}
449 commit = {"commit_id": blob_id(b"c"), "snapshot_id": snap_id}
450 objs = [_good_obj(b"obj")]
451 bf = tmp_path / "b.muse"
452 bf.write_bytes(_make_bundle(objs, [snap], [commit]))
453 r = _vp(tmp_path, ["--file", str(bf), "--stat", "--json"])
454 d = json.loads(r.output)
455 assert d["blobs"] ==1
456 assert d["snapshots"] == 1
457 assert d["commits"] == 1
458
459
460 # ---------------------------------------------------------------------------
461 # Integration: --quiet
462 # ---------------------------------------------------------------------------
463
464 class TestQuietMode:
465 def test_quiet_clean_exits_zero_no_output(self, tmp_path: pathlib.Path) -> None:
466 bf = tmp_path / "b.muse"
467 bf.write_bytes(_make_bundle())
468 r = _vp(tmp_path, ["--file", str(bf), "--quiet", "--no-local"])
469 assert r.exit_code == 0
470 assert r.output.strip() == ""
471
472 def test_quiet_failure_exits_nonzero_no_output(self, tmp_path: pathlib.Path) -> None:
473 bf = tmp_path / "b.muse"
474 bf.write_bytes(_make_bundle([_bad_hash_obj()]))
475 r = _vp(tmp_path, ["--file", str(bf), "-q", "--no-local"])
476 assert r.exit_code != 0
477 assert r.output.strip() == ""
478
479
480 # ---------------------------------------------------------------------------
481 # Integration: malformed input
482 # ---------------------------------------------------------------------------
483
484 class TestMalformedInput:
485 def test_malformed_msgpack(self, tmp_path: pathlib.Path) -> None:
486 bf = tmp_path / "b.muse"
487 bf.write_bytes(b"\xff\xff NOT VALID MSGPACK")
488 r = _vp(tmp_path, ["--file", str(bf)])
489 assert r.exit_code != 0
490
491 def test_msgpack_scalar(self, tmp_path: pathlib.Path) -> None:
492 """msgpack-encoded scalar (not a map) must error cleanly."""
493 bf = tmp_path / "b.muse"
494 bf.write_bytes(msgpack.packb(42))
495 r = _vp(tmp_path, ["--file", str(bf)])
496 assert r.exit_code != 0
497 assert "error" in r.stderr.lower() or r.exit_code != 0
498
499 def test_missing_file(self, tmp_path: pathlib.Path) -> None:
500 r = _vp(tmp_path, ["--file", str(tmp_path / "nonexistent.muse")])
501 assert r.exit_code != 0
502
503 def test_objects_not_a_list(self, tmp_path: pathlib.Path) -> None:
504 mpack = {"meta": _FULL_META, "blobs": "bad", "snapshots": [], "commits": []}
505 bf = tmp_path / "b.muse"
506 bf.write_bytes(msgpack.packb(mpack, use_bin_type=True))
507 r = _vp(tmp_path, ["--file", str(bf), "--no-local"])
508 assert r.exit_code != 0
509 # Error goes to stderr
510 assert "blobs" in r.stderr.lower()
511
512 def test_snapshots_not_a_list(self, tmp_path: pathlib.Path) -> None:
513 mpack = {"meta": _FULL_META, "blobs": [], "snapshots": 99, "commits": []}
514 bf = tmp_path / "b.muse"
515 bf.write_bytes(msgpack.packb(mpack, use_bin_type=True))
516 r = _vp(tmp_path, ["--file", str(bf), "--no-local"])
517 assert r.exit_code != 0
518 assert "snapshots" in r.stderr.lower()
519
520 def test_commits_not_a_list(self, tmp_path: pathlib.Path) -> None:
521 mpack = {"meta": _FULL_META, "blobs": [], "snapshots": [], "commits": True}
522 bf = tmp_path / "b.muse"
523 bf.write_bytes(msgpack.packb(mpack, use_bin_type=True))
524 r = _vp(tmp_path, ["--file", str(bf), "--no-local"])
525 assert r.exit_code != 0
526 assert "commits" in r.stderr.lower()
527
528
529 # ---------------------------------------------------------------------------
530 # Security
531 # ---------------------------------------------------------------------------
532
533 class TestSecurity:
534 def test_unrecognized_flag_to_stderr(self, tmp_path: pathlib.Path) -> None:
535 bf = tmp_path / "b.muse"
536 bf.write_bytes(_make_bundle())
537 r = _vp(tmp_path, ["--file", str(bf), "--unknown-flag"])
538 assert r.exit_code != 0
539 assert r.stdout_bytes == b""
540 assert "error" in r.stderr.lower()
541
542 def test_ansi_in_object_id_sanitized_in_text(self, tmp_path: pathlib.Path) -> None:
543 """ANSI in a mpack object_id must not leak to text output."""
544 ansi_oid = f"\x1b[31m{'a' * 58}\x1b[0m"
545 bad: _BadObjectEntry = {"object_id": ansi_oid, "content": b"data"}
546 bf = tmp_path / "b.muse"
547 bf.write_bytes(_make_bundle([bad]))
548 r = _vp(tmp_path, ["--file", str(bf), "--no-local"])
549 # ANSI escape sequences must not appear in text output
550 assert "\x1b" not in r.output
551
552 def test_no_traceback_on_unrecognized_flag(self, tmp_path: pathlib.Path) -> None:
553 bf = tmp_path / "b.muse"
554 bf.write_bytes(_make_bundle())
555 r = _vp(tmp_path, ["--file", str(bf), "--not-a-real-flag"])
556 assert "Traceback" not in r.output
557 assert "Traceback" not in r.stderr
558
559 def test_no_traceback_on_corrupt_bundle(self, tmp_path: pathlib.Path) -> None:
560 bf = tmp_path / "b.muse"
561 bf.write_bytes(b"\x00\x01\x02garbage")
562 r = _vp(tmp_path, ["--file", str(bf)])
563 assert "Traceback" not in r.output
564 assert "Traceback" not in r.stderr
565
566 def test_json_output_encodes_ansi_safely(self, tmp_path: pathlib.Path) -> None:
567 """JSON output must encode ANSI characters, never emit raw escape sequences."""
568 ansi_oid = f"\x1b[31m{'a' * 58}\x1b[0m"
569 bad: _BadObjectEntry = {"object_id": ansi_oid, "content": b"data"}
570 bf = tmp_path / "b.muse"
571 bf.write_bytes(_make_bundle([bad]))
572 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
573 # JSON output must not contain raw ANSI sequences
574 assert "\x1b" not in r.output
575 # But should still be parseable JSON
576 d = json.loads(r.output)
577 assert d["all_ok"] is False
578
579 def test_path_traversal_in_bundle_file(self, tmp_path: pathlib.Path) -> None:
580 """Referencing a path-traversal mpack file must fail cleanly."""
581 r = _vp(tmp_path, ["--file", "/etc/shadow"])
582 assert r.exit_code != 0
583 assert "Traceback" not in r.output
584
585 def test_deeply_nested_msgpack(self, tmp_path: pathlib.Path) -> None:
586 """Deeply nested msgpack should not crash (msgpack raises StackError)."""
587 # Build raw msgpack bytes: 2000 × fixmap(1){"x": …} + nil
588 # Avoids Python-level dict recursion; msgpack may raise StackError on unpack.
589 packed = b"\x81\xa1x" * 2000 + b"\xc0"
590 bf = tmp_path / "b.muse"
591 bf.write_bytes(packed)
592 r = _vp(tmp_path, ["--file", str(bf)])
593 # Either parses or errors — must never traceback
594 assert "Traceback" not in r.output
595 assert "Traceback" not in r.stderr
596
597
598 # ---------------------------------------------------------------------------
599 # Stress
600 # ---------------------------------------------------------------------------
601
602 class TestStress:
603 def test_500_object_bundle(self, tmp_path: pathlib.Path) -> None:
604 objs = [_good_obj(f"stress-object-{i:05d}".encode()) for i in range(500)]
605 bf = tmp_path / "b.muse"
606 bf.write_bytes(_make_bundle(objs))
607 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
608 assert r.exit_code == 0
609 d = json.loads(r.output)
610 assert d["blobs_checked"] == 500
611 assert d["all_ok"] is True
612
613 def test_500_object_bundle_one_corrupt(self, tmp_path: pathlib.Path) -> None:
614 objs = [_good_obj(f"stress-{i}".encode()) for i in range(499)]
615 objs.append(_bad_hash_obj(b"corrupt"))
616 bf = tmp_path / "b.muse"
617 bf.write_bytes(_make_bundle(objs))
618 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
619 assert r.exit_code != 0
620 d = json.loads(r.output)
621 assert d["blobs_checked"] == 500
622 assert d["all_ok"] is False
623
624 def test_200_sequential_verifications(self, tmp_path: pathlib.Path) -> None:
625 bf = tmp_path / "b.muse"
626 bf.write_bytes(_make_bundle([_good_obj(b"seq")]))
627 for _ in range(200):
628 r = _vp(tmp_path, ["--file", str(bf), "--no-local"])
629 assert r.exit_code == 0
630
631 def test_large_object_hashing(self, tmp_path: pathlib.Path) -> None:
632 """1 MiB object should hash correctly without memory issues."""
633 large_data = b"X" * (1024 * 1024)
634 obj = _good_obj(large_data)
635 bf = tmp_path / "b.muse"
636 bf.write_bytes(_make_bundle([obj]))
637 r = _vp(tmp_path, ["--file", str(bf), "--no-local", "--json"])
638 assert r.exit_code == 0
639 d = json.loads(r.output)
640 assert d["all_ok"] is True
641
642 def test_stat_on_large_bundle(self, tmp_path: pathlib.Path) -> None:
643 """--stat on a 500-object mpack must complete quickly."""
644 objs = [_good_obj(f"big-{i}".encode()) for i in range(500)]
645 bf = tmp_path / "b.muse"
646 bf.write_bytes(_make_bundle(objs))
647 r = _vp(tmp_path, ["--file", str(bf), "--stat", "--json"])
648 assert r.exit_code == 0
649 d = json.loads(r.output)
650 assert d["blobs"] ==500
File History 1 commit
sha256:a154bc65916614c833d5a40a10d81ba3eae0d0495b0afddd34dc34f18d5e91b8 fix: test suite alignment and typing audit — zero violations Sonnet 4.6 minor ⚠ 22 days ago