gabriel / muse public
test_cmd_verify_hardening.py python
865 lines 29.1 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 120 days ago
1 """Hardening tests for ``muse verify`` — security, performance, agent UX.
2
3 Covers:
4 Unit (core):
5 - _branch_refs symlink guard (symlinks silently skipped)
6 - _branch_refs size cap (oversized ref file treated as invalid)
7 - _branch_refs branch filter (only named branch returned)
8 - _MAX_COMMITS >= guard (walk stops at budget, not budget+1)
9 - _make_result helper produces correct all_ok
10 - missing-key reported as kind="key_missing" not kind="signature"
11 - fail_fast stops after first failure
12
13 Security:
14 - Symlink inside .muse/refs/heads/ is silently skipped
15 - Oversized ref file content capped — no memory explosion
16 - Invalid ref (bad hex) reported as kind="ref" not passed to BFS
17 - kind column in text output passes through sanitize_display
18
19 Error routing:
20 - I/O error during run_verify goes to stderr
21 - Failures exit with code 1
22
23 JSON schema:
24 - All _VerifyJson fields present: repo_id, branch, fail_fast, check_objects
25 - failures[].kind is one of the documented literals
26 - all_ok=True when failures=[]
27 - --branch reflected in JSON output
28 - --no-objects reflected as check_objects=false in JSON
29
30 New flags:
31 - --branch limits walk to one branch
32 - --fail-fast stops after first failure in text and json mode
33 - --json flag replaces --format json (old flag rejected)
34 - --no-objects flag sets check_objects=False in JSON
35
36 Integration:
37 - Two-branch repo: --branch verifies one, other not touched
38 - Healthy chain + corrupt branch: fail-fast returns exactly one failure
39 - Missing snapshot reported correctly
40 - Multiple failures all listed in JSON
41
42 E2E:
43 - --help shows --json, --branch, --fail-fast flags
44 - Help mentions key_missing in description
45
46 Stress:
47 - 500-commit chain passes with check_objects=True
48 - 500-commit chain with --fail-fast on first corrupt object
49 - Concurrent reads of the same repo (10 threads)
50 """
51
52 from __future__ import annotations
53
54 import datetime
55 import json
56 import pathlib
57 import threading
58
59 import pytest
60 from tests.cli_test_helper import CliRunner, InvokeResult
61
62 from muse.core.object_store import object_path, write_object
63 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
64 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
65
66 from muse.core.types import Manifest, NULL_LONG_ID, blob_id, long_id, fake_id
67 from muse.core.verify import (
68 VerifyFailure,
69 VerifyResult,
70 _MAX_COMMITS,
71 _branch_refs,
72 _make_result,
73 run_verify,
74 )
75 from muse.core.paths import heads_dir, muse_dir, ref_path
76
77 runner = CliRunner()
78 cli = None # argparse migration — CliRunner ignores this arg
79
80 _REPO_ID = "verify-hardening-test"
81
82
83 # ---------------------------------------------------------------------------
84 # TypedDicts for parsing JSON output
85 # ---------------------------------------------------------------------------
86
87
88 from typing import TypedDict
89
90
91 class _FailureOut(TypedDict):
92 kind: str
93 id: str
94 error: str
95
96
97 class _VerifyOut(TypedDict):
98 repo_id: str
99 refs_checked: int
100 commits_checked: int
101 snapshots_checked: int
102 objects_checked: int
103 signatures_checked: int
104 all_ok: bool
105 check_objects: bool
106 branch: str | None
107 fail_fast: bool
108 failures: list[_FailureOut]
109
110
111 # ---------------------------------------------------------------------------
112 # Helpers
113 # ---------------------------------------------------------------------------
114
115
116
117
118 def _init_repo(path: pathlib.Path) -> pathlib.Path:
119 muse = muse_dir(path)
120 for d in ("commits", "snapshots", "objects", "refs/heads", "keys"):
121 (muse / d).mkdir(parents=True, exist_ok=True)
122 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
123 (muse / "repo.json").write_text(
124 json.dumps({"repo_id": _REPO_ID, "domain": "midi"}), encoding="utf-8"
125 )
126 return path
127
128
129 def _env(repo: pathlib.Path) -> Manifest:
130 return {"MUSE_REPO_ROOT": str(repo)}
131
132
133 def _make_commit(
134 root: pathlib.Path,
135 parent_id: str | None = None,
136 content: bytes = b"data",
137 branch: str = "main",
138 idx: int = 0,
139 ) -> str:
140 raw = content + str(idx).encode()
141 obj_id = blob_id(raw)
142 write_object(root, obj_id, raw)
143 manifest = {f"file_{idx}.txt": obj_id}
144 snap_id = compute_snapshot_id(manifest)
145 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
146 committed_at = (
147 datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
148 + datetime.timedelta(hours=idx)
149 )
150 parent_ids = [parent_id] if parent_id else []
151 commit_id = compute_commit_id(
152 parent_ids=parent_ids,
153 snapshot_id=snap_id,
154 message=f"commit {idx}",
155 committed_at_iso=committed_at.isoformat(),
156 )
157 write_commit(
158 root,
159 CommitRecord(
160 repo_id=_REPO_ID,
161 commit_id=commit_id,
162 branch=branch,
163 snapshot_id=snap_id,
164 message=f"commit {idx}",
165 committed_at=committed_at,
166 parent_commit_id=parent_id,
167 ),
168 )
169 # Branch names with '/' require subdirectory creation.
170 branch_ref = ref_path(root, branch)
171 branch_ref.parent.mkdir(parents=True, exist_ok=True)
172 branch_ref.write_text(commit_id, encoding="utf-8")
173 return commit_id
174
175
176 _invoke_lock = threading.Lock()
177
178
179 def _invoke(args: list[str], env: Manifest) -> InvokeResult:
180 with _invoke_lock:
181 return runner.invoke(cli, args, env=env)
182
183
184 def _parse_json(result: InvokeResult) -> _VerifyOut:
185 raw: _VerifyOut = json.loads(result.output)
186 return raw
187
188
189 # ---------------------------------------------------------------------------
190 # Unit: _branch_refs
191 # ---------------------------------------------------------------------------
192
193
194 def test_branch_refs_skips_symlink(tmp_path: pathlib.Path) -> None:
195 _init_repo(tmp_path)
196 heads = heads_dir(tmp_path)
197 real_file = heads / "main"
198 real_file.write_text("a" * 64, encoding="utf-8")
199 link = heads / "malicious"
200 link.symlink_to(real_file)
201 refs = _branch_refs(tmp_path)
202 branch_names = [br for br, _ in refs]
203 assert "malicious" not in branch_names
204
205
206 def test_branch_refs_size_cap(tmp_path: pathlib.Path) -> None:
207 """Oversized ref file is capped; the truncated content fails hex validation
208 in run_verify and is reported as kind='ref' — not read entirely into memory."""
209 _init_repo(tmp_path)
210 heads = heads_dir(tmp_path)
211 # Write 10 MB into a ref file — _branch_refs reads at most 72 bytes.
212 (heads / "main").write_bytes(b"x" * (10 * 1024 * 1024))
213 refs = _branch_refs(tmp_path)
214 # The truncated content (72 'x' chars) is returned but is not a valid ID.
215 # _branch_refs does not validate format — run_verify does.
216 assert len(refs) == 1
217 branch_name, commit_id = refs[0]
218 assert branch_name == "main"
219 # Neither bare hex (64 chars) nor sha256-prefixed (71 chars) — it's garbage.
220 assert commit_id not in (long_id("x" * 64),) and not commit_id.startswith("sha256:")
221 # run_verify must report this as a ref failure.
222 result = run_verify(tmp_path)
223 assert result["all_ok"] is False
224 kinds = [f["kind"] for f in result["failures"]]
225 assert "ref" in kinds
226
227
228 def test_branch_refs_branch_filter_returns_one(tmp_path: pathlib.Path) -> None:
229 _init_repo(tmp_path)
230 _make_commit(tmp_path, content=b"main", branch="main", idx=0)
231 _make_commit(tmp_path, content=b"dev", branch="dev", idx=1)
232 refs = _branch_refs(tmp_path, branch="main")
233 assert len(refs) == 1
234 assert refs[0][0] == "main"
235
236
237 def test_branch_refs_branch_filter_missing_branch(tmp_path: pathlib.Path) -> None:
238 _init_repo(tmp_path)
239 refs = _branch_refs(tmp_path, branch="nonexistent")
240 assert refs == []
241
242
243 def test_branch_refs_branch_filter_skips_symlink(tmp_path: pathlib.Path) -> None:
244 _init_repo(tmp_path)
245 heads = heads_dir(tmp_path)
246 real = heads / "main"
247 real.write_text("a" * 64, encoding="utf-8")
248 link = heads / "malicious"
249 link.symlink_to(real)
250 refs = _branch_refs(tmp_path, branch="malicious")
251 assert refs == []
252
253
254 # ---------------------------------------------------------------------------
255 # Unit: _make_result helper
256 # ---------------------------------------------------------------------------
257
258
259 def test_make_result_all_ok_true_when_no_failures() -> None:
260 result = _make_result(1, 1, 1, 1, 0, [])
261 assert result["all_ok"] is True
262 assert result["failures"] == []
263
264
265 def test_make_result_all_ok_false_when_failures() -> None:
266 failure = VerifyFailure(kind="commit", id="abc", error="missing")
267 result = _make_result(1, 0, 0, 0, 0, [failure])
268 assert result["all_ok"] is False
269 assert len(result["failures"]) == 1
270
271
272 # ---------------------------------------------------------------------------
273 # Unit: _MAX_COMMITS guard — >= not >
274 # ---------------------------------------------------------------------------
275
276
277 def test_max_commits_guard_stops_at_budget(tmp_path: pathlib.Path) -> None:
278 """Walk should stop at _MAX_COMMITS, not _MAX_COMMITS+1."""
279 _init_repo(tmp_path)
280 # Create 3 chained commits and monkey-patch _MAX_COMMITS to 2.
281 import muse.core.verify as verify_mod
282
283 orig = verify_mod._MAX_COMMITS
284 try:
285 verify_mod._MAX_COMMITS = 2
286 prev: str | None = None
287 for i in range(5):
288 prev = _make_commit(tmp_path, parent_id=prev, idx=i)
289 result = run_verify(tmp_path)
290 # Walk stopped early — commits_checked <= 2.
291 assert result["commits_checked"] <= 2
292 finally:
293 verify_mod._MAX_COMMITS = orig
294
295
296 # ---------------------------------------------------------------------------
297 # Unit: missing key → kind="key_missing", not kind="signature"
298 # ---------------------------------------------------------------------------
299
300
301 def test_missing_key_reported_as_key_missing(tmp_path: pathlib.Path) -> None:
302 """A signed commit whose key file is absent → kind='key_missing'."""
303 _init_repo(tmp_path)
304 # Write a commit with a non-empty signature and agent_id but no key file.
305 content = b"signed commit"
306 obj_id = blob_id(content)
307 write_object(tmp_path, obj_id, content)
308 manifest = {"signed.txt": obj_id}
309 snap_id = compute_snapshot_id(manifest)
310 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
311 committed_at = datetime.datetime(2026, 3, 1, tzinfo=datetime.timezone.utc)
312 commit_id = compute_commit_id(
313 parent_ids=[],
314 snapshot_id=snap_id,
315 message="signed",
316 committed_at_iso=committed_at.isoformat(),
317 )
318 write_commit(
319 tmp_path,
320 CommitRecord(
321 repo_id=_REPO_ID,
322 commit_id=commit_id,
323 branch="main",
324 snapshot_id=snap_id,
325 message="signed",
326 committed_at=committed_at,
327 signature="ed25519:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
328 agent_id="agent-42",
329 ),
330 )
331 (heads_dir(tmp_path) / "main").write_text(
332 commit_id, encoding="utf-8"
333 )
334 result = run_verify(tmp_path)
335 kinds = [f["kind"] for f in result["failures"]]
336 assert "key_missing" in kinds
337 assert "signature" not in kinds
338
339
340 # ---------------------------------------------------------------------------
341 # Unit: fail_fast stops after first failure
342 # ---------------------------------------------------------------------------
343
344
345 def test_fail_fast_stops_after_first_failure(tmp_path: pathlib.Path) -> None:
346 _init_repo(tmp_path)
347 # Point main at a nonexistent commit — first failure.
348 fake = "a" * 64
349 (heads_dir(tmp_path) / "main").write_text(
350 fake, encoding="utf-8"
351 )
352 # Point dev at another nonexistent commit — potential second failure.
353 fake2 = "b" * 64
354 (heads_dir(tmp_path) / "dev").write_text(
355 fake2, encoding="utf-8"
356 )
357 result = run_verify(tmp_path, fail_fast=True)
358 # fail_fast: should stop after first failure, not accumulate both.
359 assert not result["all_ok"]
360 assert len(result["failures"]) == 1
361
362
363 def test_fail_fast_still_ok_when_healthy(tmp_path: pathlib.Path) -> None:
364 _init_repo(tmp_path)
365 _make_commit(tmp_path, content=b"all good", idx=0)
366 result = run_verify(tmp_path, fail_fast=True)
367 assert result["all_ok"] is True
368 assert result["failures"] == []
369
370
371 # ---------------------------------------------------------------------------
372 # Security: ANSI in branch name sanitized in text output
373 # ---------------------------------------------------------------------------
374
375
376 def test_ansi_in_branch_name_sanitized_in_text(tmp_path: pathlib.Path) -> None:
377 _init_repo(tmp_path)
378 # Write a bad ref (invalid ID) for a "branch" with ANSI escape in name.
379 heads = heads_dir(tmp_path)
380 malicious_name = "\x1b[31mmalicious\x1b[0m"
381 (heads / malicious_name).write_text("not-valid-hex-id", encoding="utf-8")
382 result = _invoke(["verify"], env=_env(tmp_path))
383 assert "\x1b[" not in result.output
384
385
386 def test_ansi_in_error_sanitized_in_text(tmp_path: pathlib.Path) -> None:
387 _init_repo(tmp_path)
388 # Point main at bad ref — the error text is sanitized.
389 (heads_dir(tmp_path) / "main").write_text(
390 "not-a-valid-commit-id", encoding="utf-8"
391 )
392 result = _invoke(["verify"], env=_env(tmp_path))
393 assert "\x1b[" not in result.output
394
395
396 # ---------------------------------------------------------------------------
397 # Error routing
398 # ---------------------------------------------------------------------------
399
400
401 def test_failure_exits_with_nonzero(tmp_path: pathlib.Path) -> None:
402 _init_repo(tmp_path)
403 (heads_dir(tmp_path) / "main").write_text(
404 "b" * 64, encoding="utf-8"
405 )
406 result = _invoke(["verify"], env=_env(tmp_path))
407 assert result.exit_code != 0
408
409
410 def test_quiet_mode_no_stdout(tmp_path: pathlib.Path) -> None:
411 _init_repo(tmp_path)
412 _make_commit(tmp_path, content=b"quiet", idx=0)
413 result = _invoke(["verify", "--quiet"], env=_env(tmp_path))
414 assert result.exit_code == 0
415 assert result.output.strip() == ""
416
417
418 def test_quiet_mode_fails_silently(tmp_path: pathlib.Path) -> None:
419 _init_repo(tmp_path)
420 (heads_dir(tmp_path) / "main").write_text(
421 "c" * 64, encoding="utf-8"
422 )
423 result = _invoke(["verify", "--quiet"], env=_env(tmp_path))
424 assert result.exit_code != 0
425 assert result.output.strip() == ""
426
427
428 # ---------------------------------------------------------------------------
429 # JSON schema
430 # ---------------------------------------------------------------------------
431
432
433 def test_json_all_fields_present(tmp_path: pathlib.Path) -> None:
434 _init_repo(tmp_path)
435 _make_commit(tmp_path, content=b"schema", idx=0)
436 result = _invoke(["verify", "--json"], env=_env(tmp_path))
437 assert result.exit_code == 0
438 data = _parse_json(result)
439 assert data["repo_id"] == _REPO_ID
440 assert data["all_ok"] is True
441 assert data["failures"] == []
442 assert isinstance(data["refs_checked"], int)
443 assert isinstance(data["commits_checked"], int)
444 assert isinstance(data["snapshots_checked"], int)
445 assert isinstance(data["objects_checked"], int)
446 assert isinstance(data["signatures_checked"], int)
447 assert isinstance(data["check_objects"], bool)
448 assert data["branch"] is None
449 assert data["fail_fast"] is False
450
451
452 def test_json_no_objects_reflected(tmp_path: pathlib.Path) -> None:
453 _init_repo(tmp_path)
454 _make_commit(tmp_path, content=b"no-obj", idx=0)
455 result = _invoke(["verify", "--json", "--no-objects"], env=_env(tmp_path))
456 assert result.exit_code == 0
457 data = _parse_json(result)
458 assert data["check_objects"] is False
459
460
461 def test_json_branch_reflected(tmp_path: pathlib.Path) -> None:
462 _init_repo(tmp_path)
463 _make_commit(tmp_path, content=b"branch-json", branch="feat/x", idx=0)
464 result = _invoke(["verify", "--json", "--branch", "feat/x"], env=_env(tmp_path))
465 assert result.exit_code == 0
466 data = _parse_json(result)
467 assert data["branch"] == "feat/x"
468
469
470 def test_json_fail_fast_reflected(tmp_path: pathlib.Path) -> None:
471 _init_repo(tmp_path)
472 _make_commit(tmp_path, content=b"fail-fast-json", idx=0)
473 result = _invoke(["verify", "--json", "--fail-fast"], env=_env(tmp_path))
474 assert result.exit_code == 0
475 data = _parse_json(result)
476 assert data["fail_fast"] is True
477
478
479 def test_json_failures_have_kind_id_error(tmp_path: pathlib.Path) -> None:
480 _init_repo(tmp_path)
481 (heads_dir(tmp_path) / "main").write_text(
482 "d" * 64, encoding="utf-8"
483 )
484 result = _invoke(["verify", "--json"], env=_env(tmp_path))
485 assert result.exit_code != 0
486 data = _parse_json(result)
487 assert data["all_ok"] is False
488 for failure in data["failures"]:
489 assert "kind" in failure
490 assert "id" in failure
491 assert "error" in failure
492
493
494 def test_json_failure_kind_is_valid_literal(tmp_path: pathlib.Path) -> None:
495 _init_repo(tmp_path)
496 (heads_dir(tmp_path) / "main").write_text(
497 "e" * 64, encoding="utf-8"
498 )
499 result = _invoke(["verify", "--json"], env=_env(tmp_path))
500 data = _parse_json(result)
501 valid_kinds = {"ref", "commit", "snapshot", "object", "signature", "key_missing"}
502 for failure in data["failures"]:
503 assert failure["kind"] in valid_kinds
504
505
506 def test_json_missing_snapshot_kind(tmp_path: pathlib.Path) -> None:
507 """A commit pointing at a nonexistent snapshot shows kind='snapshot'."""
508 _init_repo(tmp_path)
509 snap_id = "f" * 64
510 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
511 commit_id = compute_commit_id(
512 parent_ids=[],
513 snapshot_id=snap_id,
514 message="no snap",
515 committed_at_iso=committed_at.isoformat(),
516 )
517 write_commit(
518 tmp_path,
519 CommitRecord(
520 repo_id=_REPO_ID,
521 commit_id=commit_id,
522 branch="main",
523 snapshot_id=snap_id,
524 message="no snap",
525 committed_at=committed_at,
526 ),
527 )
528 (heads_dir(tmp_path) / "main").write_text(
529 commit_id, encoding="utf-8"
530 )
531 result = _invoke(["verify", "--json"], env=_env(tmp_path))
532 data = _parse_json(result)
533 kinds = [f["kind"] for f in data["failures"]]
534 assert "snapshot" in kinds
535
536
537 def test_json_missing_object_kind(tmp_path: pathlib.Path) -> None:
538 """A manifest referencing a nonexistent object shows kind='object'."""
539 _init_repo(tmp_path)
540 obj_id = NULL_LONG_ID
541 manifest = {"ghost.txt": obj_id}
542 snap_id = compute_snapshot_id(manifest)
543 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
544 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
545 commit_id = compute_commit_id(
546 parent_ids=[],
547 snapshot_id=snap_id,
548 message="ghost",
549 committed_at_iso=committed_at.isoformat(),
550 )
551 write_commit(
552 tmp_path,
553 CommitRecord(
554 repo_id=_REPO_ID,
555 commit_id=commit_id,
556 branch="main",
557 snapshot_id=snap_id,
558 message="ghost",
559 committed_at=committed_at,
560 ),
561 )
562 (heads_dir(tmp_path) / "main").write_text(
563 commit_id, encoding="utf-8"
564 )
565 result = _invoke(["verify", "--json"], env=_env(tmp_path))
566 data = _parse_json(result)
567 kinds = [f["kind"] for f in data["failures"]]
568 assert "object" in kinds
569
570
571 # ---------------------------------------------------------------------------
572 # New flags: --branch
573 # ---------------------------------------------------------------------------
574
575
576 def test_branch_flag_limits_to_named_branch(tmp_path: pathlib.Path) -> None:
577 _init_repo(tmp_path)
578 _make_commit(tmp_path, content=b"main ok", branch="main", idx=0)
579 # dev points at a nonexistent commit — if not limited, would fail.
580 (heads_dir(tmp_path) / "dev").write_text(
581 "1" * 64, encoding="utf-8"
582 )
583 result = _invoke(["verify", "--branch", "main"], env=_env(tmp_path))
584 assert result.exit_code == 0
585
586
587 def test_branch_flag_catches_failure_in_named_branch(tmp_path: pathlib.Path) -> None:
588 _init_repo(tmp_path)
589 (heads_dir(tmp_path) / "main").write_text(
590 "2" * 64, encoding="utf-8"
591 )
592 result = _invoke(["verify", "--branch", "main"], env=_env(tmp_path))
593 assert result.exit_code != 0
594
595
596 def test_branch_flag_missing_branch_is_clean(tmp_path: pathlib.Path) -> None:
597 _init_repo(tmp_path)
598 result = _invoke(["verify", "--branch", "nonexistent"], env=_env(tmp_path))
599 assert result.exit_code == 0
600
601
602 def test_branch_flag_json_branch_field(tmp_path: pathlib.Path) -> None:
603 _init_repo(tmp_path)
604 _make_commit(tmp_path, content=b"b json", branch="main", idx=0)
605 result = _invoke(["verify", "--json", "--branch", "main"], env=_env(tmp_path))
606 data = _parse_json(result)
607 assert data["branch"] == "main"
608 assert data["all_ok"] is True
609
610
611 def test_branch_flag_shown_in_text_output(tmp_path: pathlib.Path) -> None:
612 _init_repo(tmp_path)
613 _make_commit(tmp_path, content=b"text branch", branch="feat/my", idx=0)
614 result = _invoke(["verify", "--branch", "feat/my"], env=_env(tmp_path))
615 assert result.exit_code == 0
616 assert "feat/my" in result.output
617
618
619 # ---------------------------------------------------------------------------
620 # New flags: --fail-fast
621 # ---------------------------------------------------------------------------
622
623
624 def test_fail_fast_cli_stops_early(tmp_path: pathlib.Path) -> None:
625 _init_repo(tmp_path)
626 (heads_dir(tmp_path) / "main").write_text(
627 "3" * 64, encoding="utf-8"
628 )
629 (heads_dir(tmp_path) / "dev").write_text(
630 "4" * 64, encoding="utf-8"
631 )
632 result = _invoke(["verify", "--json", "--fail-fast"], env=_env(tmp_path))
633 assert result.exit_code != 0
634 data = _parse_json(result)
635 assert len(data["failures"]) == 1
636 assert data["fail_fast"] is True
637
638
639 def test_fail_fast_no_effect_on_healthy_repo(tmp_path: pathlib.Path) -> None:
640 _init_repo(tmp_path)
641 _make_commit(tmp_path, content=b"healthy ff", idx=0)
642 result = _invoke(["verify", "--fail-fast"], env=_env(tmp_path))
643 assert result.exit_code == 0
644
645
646 # ---------------------------------------------------------------------------
647 # New flags: --json replaces --format json
648 # ---------------------------------------------------------------------------
649
650
651 def test_json_flag_works(tmp_path: pathlib.Path) -> None:
652 _init_repo(tmp_path)
653 _make_commit(tmp_path, content=b"json flag", idx=0)
654 result = _invoke(["verify", "--json"], env=_env(tmp_path))
655 assert result.exit_code == 0
656 data = _parse_json(result)
657 assert "all_ok" in data
658
659
660 def test_format_flag_rejected(tmp_path: pathlib.Path) -> None:
661 _init_repo(tmp_path)
662 _make_commit(tmp_path, content=b"fmt flag", idx=0)
663 result = _invoke(["verify", "--format", "json"], env=_env(tmp_path))
664 # --format is no longer a valid flag — argparse will reject it.
665 assert result.exit_code != 0
666
667
668 # ---------------------------------------------------------------------------
669 # Integration
670 # ---------------------------------------------------------------------------
671
672
673 def test_two_branch_repo_healthy(tmp_path: pathlib.Path) -> None:
674 _init_repo(tmp_path)
675 _make_commit(tmp_path, content=b"main", branch="main", idx=0)
676 _make_commit(tmp_path, content=b"dev", branch="dev", idx=1)
677 result = run_verify(tmp_path)
678 assert result["all_ok"] is True
679 assert result["refs_checked"] == 2
680
681
682 def test_two_branch_one_broken_full_check(tmp_path: pathlib.Path) -> None:
683 _init_repo(tmp_path)
684 _make_commit(tmp_path, content=b"main ok", branch="main", idx=0)
685 (heads_dir(tmp_path) / "dev").write_text(
686 fake_id("nonexistent-dev-commit"), encoding="utf-8"
687 )
688 result = run_verify(tmp_path)
689 assert result["all_ok"] is False
690 kinds = {f["kind"] for f in result["failures"]}
691 assert "commit" in kinds
692
693
694 def test_two_branch_one_broken_with_branch_filter(tmp_path: pathlib.Path) -> None:
695 _init_repo(tmp_path)
696 _make_commit(tmp_path, content=b"main ok", branch="main", idx=0)
697 (heads_dir(tmp_path) / "dev").write_text(
698 fake_id("nonexistent-dev-6"), encoding="utf-8"
699 )
700 # Limiting to main only — should pass.
701 result = run_verify(tmp_path, branch="main")
702 assert result["all_ok"] is True
703
704
705 def test_corrupt_object_and_fail_fast(tmp_path: pathlib.Path) -> None:
706 """Corrupt the object, run with fail_fast — exactly one failure returned."""
707 import os
708
709 _init_repo(tmp_path)
710 content = b"will be corrupted"
711 obj_id = blob_id(content)
712 write_object(tmp_path, obj_id, content)
713 manifest = {"c.txt": obj_id}
714 snap_id = compute_snapshot_id(manifest)
715 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
716 committed_at = datetime.datetime(2026, 3, 5, tzinfo=datetime.timezone.utc)
717 commit_id = compute_commit_id(
718 parent_ids=[],
719 snapshot_id=snap_id,
720 message="corrupt",
721 committed_at_iso=committed_at.isoformat(),
722 )
723 write_commit(
724 tmp_path,
725 CommitRecord(
726 repo_id=_REPO_ID,
727 commit_id=commit_id,
728 branch="main",
729 snapshot_id=snap_id,
730 message="corrupt",
731 committed_at=committed_at,
732 ),
733 )
734 (heads_dir(tmp_path) / "main").write_text(
735 commit_id, encoding="utf-8"
736 )
737 obj_file = object_path(tmp_path, obj_id)
738 os.chmod(obj_file, 0o644)
739 obj_file.write_bytes(b"bad data!")
740 result = run_verify(tmp_path, check_objects=True, fail_fast=True)
741 assert result["all_ok"] is False
742 assert len(result["failures"]) == 1
743 assert result["failures"][0]["kind"] == "object"
744
745
746 def test_multiple_failures_all_listed(tmp_path: pathlib.Path) -> None:
747 _init_repo(tmp_path)
748 for i in range(5):
749 ref_path(tmp_path, f"br{i}").write_text(
750 chr(ord("a") + i) * 64, encoding="utf-8"
751 )
752 result = run_verify(tmp_path)
753 assert result["all_ok"] is False
754 assert len(result["failures"]) >= 5
755
756
757 # ---------------------------------------------------------------------------
758 # E2E: help output
759 # ---------------------------------------------------------------------------
760
761
762 def test_help_shows_json_flag() -> None:
763 result = runner.invoke(cli, ["verify", "--help"])
764 assert result.exit_code == 0
765 assert "--json" in result.output
766
767
768 def test_help_shows_branch_flag() -> None:
769 result = runner.invoke(cli, ["verify", "--help"])
770 assert result.exit_code == 0
771 assert "--branch" in result.output or "-b" in result.output
772
773
774 def test_help_shows_fail_fast_flag() -> None:
775 result = runner.invoke(cli, ["verify", "--help"])
776 assert result.exit_code == 0
777 assert "--fail-fast" in result.output
778
779
780 def test_help_mentions_key_missing() -> None:
781 result = runner.invoke(cli, ["verify", "--help"])
782 assert result.exit_code == 0
783 assert "key_missing" in result.output or "key" in result.output
784
785
786 # ---------------------------------------------------------------------------
787 # Stress
788 # ---------------------------------------------------------------------------
789
790
791 def test_stress_500_commit_chain(tmp_path: pathlib.Path) -> None:
792 _init_repo(tmp_path)
793 prev: str | None = None
794 for i in range(500):
795 prev = _make_commit(tmp_path, parent_id=prev, content=b"chain", idx=i)
796 result = run_verify(tmp_path, check_objects=True)
797 assert result["all_ok"] is True
798 assert result["commits_checked"] == 500
799
800
801 def test_stress_500_commit_no_objects(tmp_path: pathlib.Path) -> None:
802 _init_repo(tmp_path)
803 prev: str | None = None
804 for i in range(500):
805 prev = _make_commit(tmp_path, parent_id=prev, content=b"fast", idx=i)
806 result = run_verify(tmp_path, check_objects=False)
807 assert result["all_ok"] is True
808 assert result["commits_checked"] == 500
809
810
811 def test_stress_concurrent_reads(tmp_path: pathlib.Path) -> None:
812 _init_repo(tmp_path)
813 prev: str | None = None
814 for i in range(20):
815 prev = _make_commit(tmp_path, parent_id=prev, content=b"conc", idx=i)
816
817 errors: list[str] = []
818 lock = threading.Lock()
819
820 def _read() -> None:
821 res = _invoke(["verify", "--json"], env=_env(tmp_path))
822 with lock:
823 if res.exit_code != 0:
824 errors.append(f"exit_code={res.exit_code}")
825 else:
826 try:
827 data = _parse_json(res)
828 if not data["all_ok"]:
829 errors.append("all_ok=False")
830 except Exception as exc:
831 errors.append(str(exc))
832
833 threads = [threading.Thread(target=_read) for _ in range(10)]
834 for t in threads:
835 t.start()
836 for t in threads:
837 t.join()
838 assert errors == [], f"Concurrent read failures: {errors}"
839
840
841 # ---------------------------------------------------------------------------
842 # Flag registration
843 # ---------------------------------------------------------------------------
844
845
846 class TestRegisterFlags:
847 def _parse(self, *args: str) -> "argparse.Namespace":
848 import argparse
849 from muse.cli.commands.verify import register
850 p = argparse.ArgumentParser()
851 sub = p.add_subparsers()
852 register(sub)
853 return p.parse_args(["verify", *args])
854
855 def test_default_json_out_is_false(self) -> None:
856 ns = self._parse()
857 assert ns.json_out is False
858
859 def test_json_flag_sets_json_out(self) -> None:
860 ns = self._parse("--json")
861 assert ns.json_out is True
862
863 def test_j_shorthand_sets_json_out(self) -> None:
864 ns = self._parse("-j")
865 assert ns.json_out is True
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 120 days ago