"""Supercharge tests for ``muse verify``. Every JSON success response must carry ``duration_ms`` (float, ms) and ``exit_code`` (int). Every JSON error response routes to *stdout*, not stderr, so agent pipelines never receive mixed-mode output. Coverage tiers -------------- U — duration_ms / exit_code on all code paths (clean, failures, no-objects, branch-scoped, fail-fast) E — JSON error routing: _emit_error() writes to stdout in JSON mode, stderr in text mode; no traceback on any error path S — Schema completeness: all _VerifyJson fields present in every success response; all _VerifyErrorJson fields present in every error response D — Data integrity: exit_code=0 ↔ all_ok=True; exit_code=1 ↔ all_ok=False; duration_ms > 0; duration_ms is float; counters are non-negative IO — OSError during run_verify → exit_code=3 in JSON, stderr in text mode P — Performance: duration_ms < 5 000 ms for a 50-commit chain; monotone (two runs on same repo differ only by noise) Sec — No traceback on any error; no raw Python exception in stdout C — Concurrent readers produce valid JSON (10 threads, same repo) """ from __future__ import annotations from collections.abc import Mapping import datetime import json import pathlib import threading import unittest.mock as mock import pytest from tests.cli_test_helper import CliRunner, InvokeResult from muse.core.types import blob_id, long_id, short_id from muse.core.object_store import object_path, write_object from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) from muse.core.verify import run_verify from muse.core.paths import muse_dir, heads_dir, ref_path runner = CliRunner() cli = None # argparse migration — CliRunner ignores this arg _REPO_ID = "verify-supercharge-test" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _init_repo(path: pathlib.Path) -> pathlib.Path: muse = muse_dir(path) for d in ("commits", "snapshots", "objects", "refs/heads"): (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) -> Mapping[str, str]: 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 = compute_snapshot_id(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 = compute_commit_id( 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, ), ) (ref_path(root, branch)).write_text(commit_id, encoding="utf-8") return commit_id def _invoke(repo: pathlib.Path, *args: str) -> InvokeResult: from muse.cli.app import main as cli_main return runner.invoke(cli_main, ["verify", *args], env=_env(repo)) # --------------------------------------------------------------------------- # U — duration_ms and exit_code on all code paths # --------------------------------------------------------------------------- class TestElapsedAndExitCode: def test_clean_repo_exit_code_zero(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) r = _invoke(repo, "--json") assert r.exit_code == 0 d = json.loads(r.output) assert d["exit_code"] == 0 def test_clean_repo_duration_ms_present(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) r = _invoke(repo, "--json") d = json.loads(r.output) assert "duration_ms" in d assert isinstance(d["duration_ms"], float) assert d["duration_ms"] > 0 def test_failures_exit_code_one(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) # Write a ref pointing at a non-existent commit (bare hex — invalid ref format) (heads_dir(repo) / "main").write_text("b" * 64) r = _invoke(repo, "--json") assert r.exit_code == 1 d = json.loads(r.output) assert d["exit_code"] == 1 assert d["all_ok"] is False def test_failures_duration_ms_present(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) (heads_dir(repo) / "main").write_text("c" * 64) r = _invoke(repo, "--json") d = json.loads(r.output) assert isinstance(d["duration_ms"], float) assert d["duration_ms"] > 0 def test_no_objects_exit_code_zero(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=1) r = _invoke(repo, "--json", "--no-objects") assert r.exit_code == 0 d = json.loads(r.output) assert d["exit_code"] == 0 assert d["duration_ms"] > 0 def test_branch_scoped_exit_code_zero(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=2) r = _invoke(repo, "--json", "--branch", "main") assert r.exit_code == 0 d = json.loads(r.output) assert d["exit_code"] == 0 assert d["duration_ms"] > 0 def test_fail_fast_exit_code_one(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) (heads_dir(repo) / "main").write_text("d" * 64) r = _invoke(repo, "--json", "--fail-fast") assert r.exit_code == 1 d = json.loads(r.output) assert d["exit_code"] == 1 assert d["duration_ms"] > 0 # --------------------------------------------------------------------------- # E — Error routing: JSON mode → stdout; text mode → stderr # --------------------------------------------------------------------------- class TestErrorRouting: def test_io_error_json_to_stdout(self, tmp_path: pathlib.Path) -> None: """OSError during run_verify → JSON error on stdout in JSON mode.""" repo = _init_repo(tmp_path) _make_commit(repo, idx=0) with mock.patch( "muse.cli.commands.verify.run_verify", side_effect=OSError("disk full"), ): r = _invoke(repo, "--json") assert r.stderr.strip() == "", f"Expected empty stderr, got: {r.stderr!r}" d = json.loads(r.output) assert d["error"] == "io_error" assert "disk full" in d["message"] assert d["exit_code"] == 3 assert isinstance(d["duration_ms"], float) def test_io_error_text_to_stderr(self, tmp_path: pathlib.Path) -> None: """OSError during run_verify → error on stderr in text mode.""" repo = _init_repo(tmp_path) _make_commit(repo, idx=0) with mock.patch( "muse.cli.commands.verify.run_verify", side_effect=OSError("disk full"), ): r = _invoke(repo) assert r.exit_code == 3 assert r.stderr.strip() != "" assert "disk full" in r.stderr def test_io_error_quiet_no_output(self, tmp_path: pathlib.Path) -> None: """OSError in quiet mode → no output at all, exit 3.""" repo = _init_repo(tmp_path) _make_commit(repo, idx=0) with mock.patch( "muse.cli.commands.verify.run_verify", side_effect=OSError("disk full"), ): r = _invoke(repo, "--quiet") assert r.exit_code == 3 assert r.output.strip() == "" assert r.stderr.strip() == "" def test_io_error_json_no_traceback(self, tmp_path: pathlib.Path) -> None: """No Python traceback lands on stdout in JSON mode.""" repo = _init_repo(tmp_path) _make_commit(repo, idx=0) with mock.patch( "muse.cli.commands.verify.run_verify", side_effect=OSError("broken pipe"), ): r = _invoke(repo, "--json") assert "Traceback" not in r.output assert "Traceback" not in r.stderr def test_io_error_json_schema(self, tmp_path: pathlib.Path) -> None: """JSON error payload has exactly the documented keys.""" repo = _init_repo(tmp_path) _make_commit(repo, idx=0) with mock.patch( "muse.cli.commands.verify.run_verify", side_effect=OSError("nfs timeout"), ): r = _invoke(repo, "--json") d = json.loads(r.output) assert set(d) >= {"error", "message", "duration_ms", "exit_code"} # --------------------------------------------------------------------------- # S — Schema completeness # --------------------------------------------------------------------------- _SUCCESS_KEYS = { "repo_id", "refs_checked", "commits_checked", "snapshots_checked", "objects_checked", "signatures_checked", "all_ok", "nothing_checked", "check_objects", "branch", "fail_fast", "duration_ms", "exit_code", "failures", } _ERROR_KEYS = {"error", "message", "duration_ms", "exit_code"} class TestSchemaCompleteness: def test_all_success_keys_present_clean(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) d = json.loads(_invoke(repo, "--json").output) assert _SUCCESS_KEYS <= set(d), f"Missing keys: {_SUCCESS_KEYS - set(d)}" def test_all_success_keys_present_with_failures(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) (heads_dir(repo) / "main").write_text("e" * 64) d = json.loads(_invoke(repo, "--json").output) assert _SUCCESS_KEYS <= set(d), f"Missing keys: {_SUCCESS_KEYS - set(d)}" def test_all_error_keys_present(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) with mock.patch( "muse.cli.commands.verify.run_verify", side_effect=OSError("fail") ): d = json.loads(_invoke(repo, "--json").output) assert _ERROR_KEYS <= set(d), f"Missing keys: {_ERROR_KEYS - set(d)}" def test_failures_list_schema(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) # Missing commit: write a ref pointing to a valid-format but missing commit. cid = long_id("f" * 64) (heads_dir(repo) / "main").write_text(cid) d = json.loads(_invoke(repo, "--json").output) assert len(d["failures"]) >= 1 for f in d["failures"]: assert {"kind", "id", "error"} <= set(f) def test_failures_kind_is_documented_literal(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) cid = long_id("a" * 64) (heads_dir(repo) / "main").write_text(cid) d = json.loads(_invoke(repo, "--json").output) valid_kinds = {"ref", "commit", "snapshot", "object", "signature", "key_missing"} for f in d["failures"]: assert f["kind"] in valid_kinds, f"Unexpected kind: {f['kind']!r}" # --------------------------------------------------------------------------- # D — Data integrity # --------------------------------------------------------------------------- class TestDataIntegrity: def test_exit_code_zero_iff_all_ok_true(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) d = json.loads(_invoke(repo, "--json").output) assert (d["exit_code"] == 0) == (d["all_ok"] is True) def test_exit_code_one_iff_all_ok_false(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) cid = long_id("b" * 64) (heads_dir(repo) / "main").write_text(cid) d = json.loads(_invoke(repo, "--json").output) assert d["exit_code"] == 1 assert d["all_ok"] is False def test_counters_non_negative(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) d = json.loads(_invoke(repo, "--json").output) for key in ("refs_checked", "commits_checked", "snapshots_checked", "objects_checked", "signatures_checked"): assert d[key] >= 0, f"{key} is negative: {d[key]}" def test_check_objects_reflected_true(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) d = json.loads(_invoke(repo, "--json").output) assert d["check_objects"] is True def test_check_objects_reflected_false(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) d = json.loads(_invoke(repo, "--json", "--no-objects").output) assert d["check_objects"] is False def test_branch_reflected_in_json(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) d = json.loads(_invoke(repo, "--json", "--branch", "main").output) assert d["branch"] == "main" def test_branch_none_when_not_specified(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) d = json.loads(_invoke(repo, "--json").output) assert d["branch"] is None def test_fail_fast_reflected_true(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) d = json.loads(_invoke(repo, "--json", "--fail-fast").output) assert d["fail_fast"] is True def test_fail_fast_reflected_false(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) d = json.loads(_invoke(repo, "--json").output) assert d["fail_fast"] is False def test_nothing_checked_false_when_commits_exist(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) d = json.loads(_invoke(repo, "--json").output) assert d["nothing_checked"] is False def test_nothing_checked_true_empty_repo(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) d = json.loads(_invoke(repo, "--json").output) assert d["nothing_checked"] is True def test_duration_ms_is_float(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) d = json.loads(_invoke(repo, "--json").output) assert isinstance(d["duration_ms"], float) def test_blob_id_unique_per_content(self, tmp_path: pathlib.Path) -> None: """blob_id produces distinct IDs for distinct byte sequences.""" ids = {blob_id(b"content-a"), blob_id(b"content-b"), blob_id(b"content-c")} assert len(ids) == 3 def test_long_id_round_trips(self) -> None: """long_id strips sha256: prefix correctly for comparison.""" hex_val = "a" * 64 full = long_id(hex_val) assert full == "sha256:" + hex_val assert full == f"sha256:{hex_val}" def test_short_id_abbreviates(self) -> None: full = long_id("f" * 64) s = short_id(full) assert s.startswith("sha256:") assert len(s) < len(full) # --------------------------------------------------------------------------- # IO — OSError handling # --------------------------------------------------------------------------- class TestIOErrorHandling: def test_exit_code_3_on_io_error_json(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) with mock.patch( "muse.cli.commands.verify.run_verify", side_effect=OSError("io fail") ): r = _invoke(repo, "--json") assert r.exit_code == 3 d = json.loads(r.output) assert d["exit_code"] == 3 def test_exit_code_3_on_io_error_text(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) with mock.patch( "muse.cli.commands.verify.run_verify", side_effect=OSError("io fail") ): r = _invoke(repo) assert r.exit_code == 3 def test_io_error_json_no_stderr(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) with mock.patch( "muse.cli.commands.verify.run_verify", side_effect=OSError("io fail") ): r = _invoke(repo, "--json") assert r.stderr.strip() == "" def test_io_error_text_has_stderr(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) with mock.patch( "muse.cli.commands.verify.run_verify", side_effect=OSError("io fail") ): r = _invoke(repo) assert r.stderr.strip() != "" # --------------------------------------------------------------------------- # P — Performance # --------------------------------------------------------------------------- class TestPerformance: def test_duration_ms_positive(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) _make_commit(repo, idx=0) d = json.loads(_invoke(repo, "--json").output) assert d["duration_ms"] > 0 def test_50_commit_chain_under_5000ms(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) prev: str | None = None for i in range(50): prev = _make_commit(repo, parent_id=prev, idx=i) d = json.loads(_invoke(repo, "--json").output) assert d["duration_ms"] < 5_000, f"Too slow: {d['duration_ms']} ms" assert d["all_ok"] is True def test_50_commit_chain_no_objects_faster(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) prev: str | None = None for i in range(50): prev = _make_commit(repo, parent_id=prev, idx=i) full = json.loads(_invoke(repo, "--json").output)["duration_ms"] fast = json.loads(_invoke(repo, "--json", "--no-objects").output)["duration_ms"] # --no-objects should generally be faster; we allow some timing noise # but cap both under 10 s to prevent runaway assert fast < 10_000 assert full < 10_000 # --------------------------------------------------------------------------- # Sec — Security # --------------------------------------------------------------------------- class TestSecurity: def test_no_traceback_on_json_io_error(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) with mock.patch( "muse.cli.commands.verify.run_verify", side_effect=OSError("fail") ): r = _invoke(repo, "--json") assert "Traceback" not in r.output assert "Traceback" not in r.stderr def test_no_traceback_on_text_io_error(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) with mock.patch( "muse.cli.commands.verify.run_verify", side_effect=OSError("fail") ): r = _invoke(repo) assert "Traceback" not in r.output assert "Traceback" not in r.stderr def test_no_raw_exception_in_stdout(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) with mock.patch( "muse.cli.commands.verify.run_verify", side_effect=OSError("secret path") ): r = _invoke(repo, "--json") # The exception message may appear in the JSON "message" field — that's # intentional. What we check is that no raw Python exception string # (e.g. "OSError:") leaks outside the JSON structure. assert "OSError:" not in r.output assert "OSError:" not in r.stderr # --------------------------------------------------------------------------- # C — Concurrent readers # --------------------------------------------------------------------------- class TestConcurrent: def test_10_concurrent_reads_all_valid_json(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) prev: str | None = None for i in range(10): prev = _make_commit(repo, parent_id=prev, idx=i) results: list[dict] = [] errors: list[Exception] = [] lock = threading.Lock() def _read() -> None: try: r = _invoke(repo, "--json") d = json.loads(r.output) with lock: results.append(d) except Exception as exc: with lock: errors.append(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"Thread errors: {errors}" assert len(results) == 10 for d in results: assert d["all_ok"] is True assert d["exit_code"] == 0 assert isinstance(d["duration_ms"], float)