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