"""Hardening tests for ``muse verify`` — security, performance, agent UX. Covers: Unit (core): - _branch_refs symlink guard (symlinks silently skipped) - _branch_refs size cap (oversized ref file treated as invalid) - _branch_refs branch filter (only named branch returned) - _MAX_COMMITS >= guard (walk stops at budget, not budget+1) - _make_result helper produces correct all_ok - missing-key reported as kind="key_missing" not kind="signature" - fail_fast stops after first failure Security: - Symlink inside .muse/refs/heads/ is silently skipped - Oversized ref file content capped — no memory explosion - Invalid ref (bad hex) reported as kind="ref" not passed to BFS - kind column in text output passes through sanitize_display Error routing: - I/O error during run_verify goes to stderr - Failures exit with code 1 JSON schema: - All _VerifyJson fields present: repo_id, branch, fail_fast, check_objects - failures[].kind is one of the documented literals - all_ok=True when failures=[] - --branch reflected in JSON output - --no-objects reflected as check_objects=false in JSON New flags: - --branch limits walk to one branch - --fail-fast stops after first failure in text and json mode - --json flag replaces --format json (old flag rejected) - --no-objects flag sets check_objects=False in JSON Integration: - Two-branch repo: --branch verifies one, other not touched - Healthy chain + corrupt branch: fail-fast returns exactly one failure - Missing snapshot reported correctly - Multiple failures all listed in JSON E2E: - --help shows --json, --branch, --fail-fast flags - Help mentions key_missing in description Stress: - 500-commit chain passes with check_objects=True - 500-commit chain with --fail-fast on first corrupt object - Concurrent reads of the same repo (10 threads) """ from __future__ import annotations import datetime import json import pathlib import threading import pytest from tests.cli_test_helper import CliRunner, InvokeResult from muse.core.object_store import object_path, write_object from muse.core.ids import hash_commit, hash_snapshot from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) from muse.core.types import Manifest, NULL_LONG_ID, blob_id, long_id, fake_id from muse.core.verify import ( VerifyFailure, VerifyResult, _MAX_COMMITS, _branch_refs, _make_result, run_verify, ) from muse.core.paths import heads_dir, muse_dir, ref_path runner = CliRunner() cli = None # argparse migration — CliRunner ignores this arg _REPO_ID = "verify-hardening-test" # --------------------------------------------------------------------------- # TypedDicts for parsing JSON output # --------------------------------------------------------------------------- from typing import TypedDict class _FailureOut(TypedDict): kind: str id: str error: str class _VerifyOut(TypedDict): repo_id: str refs_checked: int commits_checked: int snapshots_checked: int objects_checked: int signatures_checked: int all_ok: bool check_objects: bool branch: str | None fail_fast: bool failures: list[_FailureOut] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _init_repo(path: pathlib.Path) -> pathlib.Path: muse = muse_dir(path) for d in ("commits", "snapshots", "objects", "refs/heads", "keys"): (muse / d).mkdir(parents=True, exist_ok=True) (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8") (muse / "repo.json").write_text( json.dumps({"repo_id": _REPO_ID, "domain": "midi"}), encoding="utf-8" ) return path def _env(repo: pathlib.Path) -> Manifest: return {"MUSE_REPO_ROOT": str(repo)} def _make_commit( root: pathlib.Path, parent_id: str | None = None, content: bytes = b"data", branch: str = "main", idx: int = 0, ) -> str: raw = content + str(idx).encode() obj_id = blob_id(raw) write_object(root, obj_id, raw) manifest = {f"file_{idx}.txt": obj_id} snap_id = hash_snapshot(manifest) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) committed_at = ( datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) + datetime.timedelta(hours=idx) ) parent_ids = [parent_id] if parent_id else [] commit_id = hash_commit( parent_ids=parent_ids, snapshot_id=snap_id, message=f"commit {idx}", committed_at_iso=committed_at.isoformat(), ) write_commit( root, CommitRecord( commit_id=commit_id, branch=branch, snapshot_id=snap_id, message=f"commit {idx}", committed_at=committed_at, parent_commit_id=parent_id, ), ) # Branch names with '/' require subdirectory creation. branch_ref = ref_path(root, branch) branch_ref.parent.mkdir(parents=True, exist_ok=True) branch_ref.write_text(commit_id, encoding="utf-8") return commit_id _invoke_lock = threading.Lock() def _invoke(args: list[str], env: Manifest) -> InvokeResult: with _invoke_lock: return runner.invoke(cli, args, env=env) def _parse_json(result: InvokeResult) -> _VerifyOut: raw: _VerifyOut = json.loads(result.output) return raw # --------------------------------------------------------------------------- # Unit: _branch_refs # --------------------------------------------------------------------------- def test_branch_refs_skips_symlink(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) heads = heads_dir(tmp_path) real_file = heads / "main" real_file.write_text("a" * 64, encoding="utf-8") link = heads / "malicious" link.symlink_to(real_file) refs = _branch_refs(tmp_path) branch_names = [br for br, _ in refs] assert "malicious" not in branch_names def test_branch_refs_size_cap(tmp_path: pathlib.Path) -> None: """Oversized ref file is capped; the truncated content fails hex validation in run_verify and is reported as kind='ref' — not read entirely into memory.""" _init_repo(tmp_path) heads = heads_dir(tmp_path) # Write 10 MB into a ref file — _branch_refs reads at most 72 bytes. (heads / "main").write_bytes(b"x" * (10 * 1024 * 1024)) refs = _branch_refs(tmp_path) # The truncated content (72 'x' chars) is returned but is not a valid ID. # _branch_refs does not validate format — run_verify does. assert len(refs) == 1 branch_name, commit_id = refs[0] assert branch_name == "main" # Neither bare hex (64 chars) nor sha256-prefixed (71 chars) — it's garbage. assert commit_id not in (long_id("x" * 64),) and not commit_id.startswith("sha256:") # run_verify must report this as a ref failure. result = run_verify(tmp_path) assert result["all_ok"] is False kinds = [f["kind"] for f in result["failures"]] assert "ref" in kinds def test_branch_refs_branch_filter_returns_one(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"main", branch="main", idx=0) _make_commit(tmp_path, content=b"dev", branch="dev", idx=1) refs = _branch_refs(tmp_path, branch="main") assert len(refs) == 1 assert refs[0][0] == "main" def test_branch_refs_branch_filter_missing_branch(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) refs = _branch_refs(tmp_path, branch="nonexistent") assert refs == [] def test_branch_refs_branch_filter_skips_symlink(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) heads = heads_dir(tmp_path) real = heads / "main" real.write_text("a" * 64, encoding="utf-8") link = heads / "malicious" link.symlink_to(real) refs = _branch_refs(tmp_path, branch="malicious") assert refs == [] # --------------------------------------------------------------------------- # Unit: _make_result helper # --------------------------------------------------------------------------- def test_make_result_all_ok_true_when_no_failures() -> None: result = _make_result(1, 1, 1, 1, 0, []) assert result["all_ok"] is True assert result["failures"] == [] def test_make_result_all_ok_false_when_failures() -> None: failure = VerifyFailure(kind="commit", id="abc", error="missing") result = _make_result(1, 0, 0, 0, 0, [failure]) assert result["all_ok"] is False assert len(result["failures"]) == 1 # --------------------------------------------------------------------------- # Unit: _MAX_COMMITS guard — >= not > # --------------------------------------------------------------------------- def test_max_commits_guard_stops_at_budget(tmp_path: pathlib.Path) -> None: """Walk should stop at _MAX_COMMITS, not _MAX_COMMITS+1.""" _init_repo(tmp_path) # Create 3 chained commits and monkey-patch _MAX_COMMITS to 2. import muse.core.verify as verify_mod orig = verify_mod._MAX_COMMITS try: verify_mod._MAX_COMMITS = 2 prev: str | None = None for i in range(5): prev = _make_commit(tmp_path, parent_id=prev, idx=i) result = run_verify(tmp_path) # Walk stopped early — commits_checked <= 2. assert result["commits_checked"] <= 2 finally: verify_mod._MAX_COMMITS = orig # --------------------------------------------------------------------------- # Unit: missing key → kind="key_missing", not kind="signature" # --------------------------------------------------------------------------- def test_missing_key_reported_as_key_missing(tmp_path: pathlib.Path) -> None: """A signed commit whose key file is absent → kind='key_missing'.""" _init_repo(tmp_path) # Write a commit with a non-empty signature and agent_id but no key file. content = b"signed commit" obj_id = blob_id(content) write_object(tmp_path, obj_id, content) manifest = {"signed.txt": obj_id} snap_id = hash_snapshot(manifest) write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) committed_at = datetime.datetime(2026, 3, 1, tzinfo=datetime.timezone.utc) commit_id = hash_commit( parent_ids=[], snapshot_id=snap_id, message="signed", committed_at_iso=committed_at.isoformat(), ) write_commit( tmp_path, CommitRecord( commit_id=commit_id, branch="main", snapshot_id=snap_id, message="signed", committed_at=committed_at, signature="ed25519:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", agent_id="agent-42", ), ) (heads_dir(tmp_path) / "main").write_text( commit_id, encoding="utf-8" ) result = run_verify(tmp_path) kinds = [f["kind"] for f in result["failures"]] assert "key_missing" in kinds assert "signature" not in kinds # --------------------------------------------------------------------------- # Unit: fail_fast stops after first failure # --------------------------------------------------------------------------- def test_fail_fast_stops_after_first_failure(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) # Point main at a nonexistent commit — first failure. fake = "a" * 64 (heads_dir(tmp_path) / "main").write_text( fake, encoding="utf-8" ) # Point dev at another nonexistent commit — potential second failure. fake2 = "b" * 64 (heads_dir(tmp_path) / "dev").write_text( fake2, encoding="utf-8" ) result = run_verify(tmp_path, fail_fast=True) # fail_fast: should stop after first failure, not accumulate both. assert not result["all_ok"] assert len(result["failures"]) == 1 def test_fail_fast_still_ok_when_healthy(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"all good", idx=0) result = run_verify(tmp_path, fail_fast=True) assert result["all_ok"] is True assert result["failures"] == [] # --------------------------------------------------------------------------- # Security: ANSI in branch name sanitized in text output # --------------------------------------------------------------------------- def test_ansi_in_branch_name_sanitized_in_text(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) # Write a bad ref (invalid ID) for a "branch" with ANSI escape in name. heads = heads_dir(tmp_path) malicious_name = "\x1b[31mmalicious\x1b[0m" (heads / malicious_name).write_text("not-valid-hex-id", encoding="utf-8") result = _invoke(["verify"], env=_env(tmp_path)) assert "\x1b[" not in result.output def test_ansi_in_error_sanitized_in_text(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) # Point main at bad ref — the error text is sanitized. (heads_dir(tmp_path) / "main").write_text( "not-a-valid-commit-id", encoding="utf-8" ) result = _invoke(["verify"], env=_env(tmp_path)) assert "\x1b[" not in result.output # --------------------------------------------------------------------------- # Error routing # --------------------------------------------------------------------------- def test_failure_exits_with_nonzero(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) (heads_dir(tmp_path) / "main").write_text( "b" * 64, encoding="utf-8" ) result = _invoke(["verify"], env=_env(tmp_path)) assert result.exit_code != 0 def test_quiet_mode_no_stdout(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"quiet", idx=0) result = _invoke(["verify", "--quiet"], env=_env(tmp_path)) assert result.exit_code == 0 assert result.output.strip() == "" def test_quiet_mode_fails_silently(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) (heads_dir(tmp_path) / "main").write_text( "c" * 64, encoding="utf-8" ) result = _invoke(["verify", "--quiet"], env=_env(tmp_path)) assert result.exit_code != 0 assert result.output.strip() == "" # --------------------------------------------------------------------------- # JSON schema # --------------------------------------------------------------------------- def test_json_all_fields_present(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"schema", idx=0) result = _invoke(["verify", "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert data["repo_id"] == _REPO_ID assert data["all_ok"] is True assert data["failures"] == [] assert isinstance(data["refs_checked"], int) assert isinstance(data["commits_checked"], int) assert isinstance(data["snapshots_checked"], int) assert isinstance(data["objects_checked"], int) assert isinstance(data["signatures_checked"], int) assert isinstance(data["check_objects"], bool) assert data["branch"] is None assert data["fail_fast"] is False def test_json_no_objects_reflected(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"no-obj", idx=0) result = _invoke(["verify", "--json", "--no-objects"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert data["check_objects"] is False def test_json_branch_reflected(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"branch-json", branch="feat/x", idx=0) result = _invoke(["verify", "--json", "--branch", "feat/x"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert data["branch"] == "feat/x" def test_json_fail_fast_reflected(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"fail-fast-json", idx=0) result = _invoke(["verify", "--json", "--fail-fast"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert data["fail_fast"] is True def test_json_failures_have_kind_id_error(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) (heads_dir(tmp_path) / "main").write_text( "d" * 64, encoding="utf-8" ) result = _invoke(["verify", "--json"], env=_env(tmp_path)) assert result.exit_code != 0 data = _parse_json(result) assert data["all_ok"] is False for failure in data["failures"]: assert "kind" in failure assert "id" in failure assert "error" in failure def test_json_failure_kind_is_valid_literal(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) (heads_dir(tmp_path) / "main").write_text( "e" * 64, encoding="utf-8" ) result = _invoke(["verify", "--json"], env=_env(tmp_path)) data = _parse_json(result) valid_kinds = {"ref", "commit", "snapshot", "object", "signature", "key_missing"} for failure in data["failures"]: assert failure["kind"] in valid_kinds def test_json_missing_snapshot_kind(tmp_path: pathlib.Path) -> None: """A commit pointing at a nonexistent snapshot shows kind='snapshot'.""" _init_repo(tmp_path) snap_id = long_id("f" * 64) committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) commit_id = hash_commit( parent_ids=[], snapshot_id=snap_id, message="no snap", committed_at_iso=committed_at.isoformat(), ) write_commit( tmp_path, CommitRecord( commit_id=commit_id, branch="main", snapshot_id=snap_id, message="no snap", committed_at=committed_at, ), ) (heads_dir(tmp_path) / "main").write_text( commit_id, encoding="utf-8" ) result = _invoke(["verify", "--json"], env=_env(tmp_path)) data = _parse_json(result) kinds = [f["kind"] for f in data["failures"]] assert "snapshot" in kinds def test_json_missing_object_kind(tmp_path: pathlib.Path) -> None: """A manifest referencing a nonexistent object shows kind='object'.""" _init_repo(tmp_path) obj_id = NULL_LONG_ID manifest = {"ghost.txt": obj_id} snap_id = hash_snapshot(manifest) write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) commit_id = hash_commit( parent_ids=[], snapshot_id=snap_id, message="ghost", committed_at_iso=committed_at.isoformat(), ) write_commit( tmp_path, CommitRecord( commit_id=commit_id, branch="main", snapshot_id=snap_id, message="ghost", committed_at=committed_at, ), ) (heads_dir(tmp_path) / "main").write_text( commit_id, encoding="utf-8" ) result = _invoke(["verify", "--json"], env=_env(tmp_path)) data = _parse_json(result) kinds = [f["kind"] for f in data["failures"]] assert "object" in kinds # --------------------------------------------------------------------------- # New flags: --branch # --------------------------------------------------------------------------- def test_branch_flag_limits_to_named_branch(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"main ok", branch="main", idx=0) # dev points at a nonexistent commit — if not limited, would fail. (heads_dir(tmp_path) / "dev").write_text( "1" * 64, encoding="utf-8" ) result = _invoke(["verify", "--branch", "main"], env=_env(tmp_path)) assert result.exit_code == 0 def test_branch_flag_catches_failure_in_named_branch(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) (heads_dir(tmp_path) / "main").write_text( "2" * 64, encoding="utf-8" ) result = _invoke(["verify", "--branch", "main"], env=_env(tmp_path)) assert result.exit_code != 0 def test_branch_flag_missing_branch_is_clean(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(["verify", "--branch", "nonexistent"], env=_env(tmp_path)) assert result.exit_code == 0 def test_branch_flag_json_branch_field(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"b json", branch="main", idx=0) result = _invoke(["verify", "--json", "--branch", "main"], env=_env(tmp_path)) data = _parse_json(result) assert data["branch"] == "main" assert data["all_ok"] is True def test_branch_flag_shown_in_text_output(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"text branch", branch="feat/my", idx=0) result = _invoke(["verify", "--branch", "feat/my"], env=_env(tmp_path)) assert result.exit_code == 0 assert "feat/my" in result.output # --------------------------------------------------------------------------- # New flags: --fail-fast # --------------------------------------------------------------------------- def test_fail_fast_cli_stops_early(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) (heads_dir(tmp_path) / "main").write_text( "3" * 64, encoding="utf-8" ) (heads_dir(tmp_path) / "dev").write_text( "4" * 64, encoding="utf-8" ) result = _invoke(["verify", "--json", "--fail-fast"], env=_env(tmp_path)) assert result.exit_code != 0 data = _parse_json(result) assert len(data["failures"]) == 1 assert data["fail_fast"] is True def test_fail_fast_no_effect_on_healthy_repo(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"healthy ff", idx=0) result = _invoke(["verify", "--fail-fast"], env=_env(tmp_path)) assert result.exit_code == 0 # --------------------------------------------------------------------------- # New flags: --json replaces --format json # --------------------------------------------------------------------------- def test_json_flag_works(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"json flag", idx=0) result = _invoke(["verify", "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert "all_ok" in data def test_format_flag_rejected(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"fmt flag", idx=0) result = _invoke(["verify", "--format", "json"], env=_env(tmp_path)) # --format is no longer a valid flag — argparse will reject it. assert result.exit_code != 0 # --------------------------------------------------------------------------- # Integration # --------------------------------------------------------------------------- def test_two_branch_repo_healthy(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"main", branch="main", idx=0) _make_commit(tmp_path, content=b"dev", branch="dev", idx=1) result = run_verify(tmp_path) assert result["all_ok"] is True assert result["refs_checked"] == 2 def test_two_branch_one_broken_full_check(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"main ok", branch="main", idx=0) (heads_dir(tmp_path) / "dev").write_text( fake_id("nonexistent-dev-commit"), encoding="utf-8" ) result = run_verify(tmp_path) assert result["all_ok"] is False kinds = {f["kind"] for f in result["failures"]} assert "commit" in kinds def test_two_branch_one_broken_with_branch_filter(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"main ok", branch="main", idx=0) (heads_dir(tmp_path) / "dev").write_text( fake_id("nonexistent-dev-6"), encoding="utf-8" ) # Limiting to main only — should pass. result = run_verify(tmp_path, branch="main") assert result["all_ok"] is True def test_corrupt_object_and_fail_fast(tmp_path: pathlib.Path) -> None: """Corrupt the object, run with fail_fast — exactly one failure returned.""" import os _init_repo(tmp_path) content = b"will be corrupted" obj_id = blob_id(content) write_object(tmp_path, obj_id, content) manifest = {"c.txt": obj_id} snap_id = hash_snapshot(manifest) write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) committed_at = datetime.datetime(2026, 3, 5, tzinfo=datetime.timezone.utc) commit_id = hash_commit( parent_ids=[], snapshot_id=snap_id, message="corrupt", committed_at_iso=committed_at.isoformat(), ) write_commit( tmp_path, CommitRecord( commit_id=commit_id, branch="main", snapshot_id=snap_id, message="corrupt", committed_at=committed_at, ), ) (heads_dir(tmp_path) / "main").write_text( commit_id, encoding="utf-8" ) obj_file = object_path(tmp_path, obj_id) os.chmod(obj_file, 0o644) obj_file.write_bytes(b"bad data!") result = run_verify(tmp_path, check_objects=True, fail_fast=True) assert result["all_ok"] is False assert len(result["failures"]) == 1 assert result["failures"][0]["kind"] == "object" def test_multiple_failures_all_listed(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) for i in range(5): ref_path(tmp_path, f"br{i}").write_text( chr(ord("a") + i) * 64, encoding="utf-8" ) result = run_verify(tmp_path) assert result["all_ok"] is False assert len(result["failures"]) >= 5 # --------------------------------------------------------------------------- # E2E: help output # --------------------------------------------------------------------------- def test_help_shows_json_flag() -> None: result = runner.invoke(cli, ["verify", "--help"]) assert result.exit_code == 0 assert "--json" in result.output def test_help_shows_branch_flag() -> None: result = runner.invoke(cli, ["verify", "--help"]) assert result.exit_code == 0 assert "--branch" in result.output or "-b" in result.output def test_help_shows_fail_fast_flag() -> None: result = runner.invoke(cli, ["verify", "--help"]) assert result.exit_code == 0 assert "--fail-fast" in result.output def test_help_mentions_key_missing() -> None: result = runner.invoke(cli, ["verify", "--help"]) assert result.exit_code == 0 assert "key_missing" in result.output or "key" in result.output # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- def test_stress_500_commit_chain(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) prev: str | None = None for i in range(500): prev = _make_commit(tmp_path, parent_id=prev, content=b"chain", idx=i) result = run_verify(tmp_path, check_objects=True) assert result["all_ok"] is True assert result["commits_checked"] == 500 def test_stress_500_commit_no_objects(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) prev: str | None = None for i in range(500): prev = _make_commit(tmp_path, parent_id=prev, content=b"fast", idx=i) result = run_verify(tmp_path, check_objects=False) assert result["all_ok"] is True assert result["commits_checked"] == 500 def test_stress_concurrent_reads(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) prev: str | None = None for i in range(20): prev = _make_commit(tmp_path, parent_id=prev, content=b"conc", idx=i) errors: list[str] = [] lock = threading.Lock() def _read() -> None: res = _invoke(["verify", "--json"], env=_env(tmp_path)) with lock: if res.exit_code != 0: errors.append(f"exit_code={res.exit_code}") else: try: data = _parse_json(res) if not data["all_ok"]: errors.append("all_ok=False") except Exception as exc: errors.append(str(exc)) threads = [threading.Thread(target=_read) for _ in range(10)] for t in threads: t.start() for t in threads: t.join() assert errors == [], f"Concurrent read failures: {errors}" # --------------------------------------------------------------------------- # Flag registration # --------------------------------------------------------------------------- class TestRegisterFlags: def _parse(self, *args: str) -> "argparse.Namespace": import argparse from muse.cli.commands.verify import register p = argparse.ArgumentParser() sub = p.add_subparsers() register(sub) return p.parse_args(["verify", *args]) def test_default_json_out_is_false(self) -> None: ns = self._parse() assert ns.json_out is False def test_json_flag_sets_json_out(self) -> None: ns = self._parse("--json") assert ns.json_out is True def test_j_shorthand_sets_json_out(self) -> None: ns = self._parse("-j") assert ns.json_out is True