"""Comprehensive tests for ``muse plumbing read-commit``. Coverage tiers -------------- - Unit: _ALL_FIELDS completeness - Integration: JSON/text format, prefix resolution, --fields filter, parent chain - Security: ANSI in branch/author/message stripped in text mode - Stress: 200 sequential reads, --fields on large schema """ from __future__ import annotations import datetime import json import pathlib from muse.core.errors import ExitCode from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot from tests.cli_test_helper import CliRunner, InvokeResult runner = CliRunner() # Module-level constants so every test uses the same deterministic inputs. _SNAP_ID: str = compute_snapshot_id({}) _COMMITTED_AT: datetime.datetime = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: repo = tmp_path / "repo" muse = repo / ".muse" for sub in ("objects", "commits", "snapshots", "refs/heads"): (muse / sub).mkdir(parents=True) (muse / "HEAD").write_text("ref: refs/heads/main") (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"})) return repo def _commit( repo: pathlib.Path, *, branch: str = "main", message: str = "test commit", author: str = "tester", parent: str | None = None, agent_id: str = "", model_id: str = "", ) -> str: """Write a commit with a real content-addressed ID; return the commit_id.""" parent_ids: list[str] = [parent] if parent else [] commit_id = compute_commit_id(parent_ids, _SNAP_ID, message, _COMMITTED_AT.isoformat()) write_snapshot(repo, SnapshotRecord( snapshot_id=_SNAP_ID, manifest={}, created_at=_COMMITTED_AT, )) rec = CommitRecord( commit_id=commit_id, repo_id="test-repo", branch=branch, snapshot_id=_SNAP_ID, message=message, committed_at=_COMMITTED_AT, author=author, parent_commit_id=parent, agent_id=agent_id, model_id=model_id, ) write_commit(repo, rec) return commit_id def _rc(repo: pathlib.Path, *args: str) -> InvokeResult: from muse.cli.app import main as cli return runner.invoke( cli, ["read-commit", *args], env={"MUSE_REPO_ROOT": str(repo)}, ) # --------------------------------------------------------------------------- # Unit — _ALL_FIELDS # --------------------------------------------------------------------------- class TestAllFields: def test_all_fields_matches_commitdict_annotations(self) -> None: """_ALL_FIELDS must be exactly the keys in CommitDict.__annotations__.""" from muse.cli.commands.plumbing.read_commit import _ALL_FIELDS from muse.core.store import CommitDict assert _ALL_FIELDS == frozenset(CommitDict.__annotations__.keys()) def test_required_fields_present(self) -> None: from muse.cli.commands.plumbing.read_commit import _ALL_FIELDS for field in ("commit_id", "branch", "message", "committed_at", "agent_id", "model_id", "format_version", "reviewed_by"): assert field in _ALL_FIELDS # --------------------------------------------------------------------------- # Integration — JSON format # --------------------------------------------------------------------------- class TestJsonFormat: def test_full_schema_returned(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="hello world") result = _rc(repo, cid) assert result.exit_code == 0 data = json.loads(result.output) assert data["commit_id"] == cid assert data["message"] == "hello world" assert data["branch"] == "main" assert "format_version" in data def test_json_flag_shorthand(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="shorthand test") result = _rc(repo, "--json", cid) assert result.exit_code == 0 assert "commit_id" in json.loads(result.output) def test_agent_provenance_fields_present(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, agent_id="my-agent", model_id="claude-4") data = json.loads(_rc(repo, cid).output) assert data["agent_id"] == "my-agent" assert data["model_id"] == "claude-4" def test_parent_commit_id_null_for_root(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="root commit") data = json.loads(_rc(repo, cid).output) assert data["parent_commit_id"] is None def test_parent_commit_id_set_for_child(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) parent = _commit(repo, message="parent commit") child = _commit(repo, message="child commit", parent=parent) data = json.loads(_rc(repo, child).output) assert data["parent_commit_id"] == parent def test_committed_at_is_iso8601(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="iso date test") data = json.loads(_rc(repo, cid).output) # Should parse without error datetime.datetime.fromisoformat(data["committed_at"]) def test_snapshot_id_in_output(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="snapshot test") data = json.loads(_rc(repo, cid).output) assert len(data["snapshot_id"]) == 64 # --------------------------------------------------------------------------- # Integration — text format # --------------------------------------------------------------------------- class TestTextFormat: def test_text_format_contains_commit_prefix(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="text test") result = _rc(repo, "--format", "text", cid) assert result.exit_code == 0 line = result.output.strip() assert cid[:12] in line def test_text_format_contains_branch(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, branch="main", message="branch test") result = _rc(repo, "--format", "text", cid) assert "main" in result.output def test_text_format_contains_message(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="my commit message") result = _rc(repo, "--format", "text", cid) assert "my commit message" in result.output def test_text_multiline_message_flattened(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="line one\nline two") result = _rc(repo, "--format", "text", cid) # Newline replaced with space — output stays on one line assert "\n" not in result.output.strip() assert "line one" in result.output # --------------------------------------------------------------------------- # Integration — prefix resolution # --------------------------------------------------------------------------- class TestPrefixResolution: def test_8char_prefix_resolves(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="prefix resolve test") result = _rc(repo, cid[:8]) assert result.exit_code == 0 assert json.loads(result.output)["commit_id"] == cid def test_ambiguous_prefix_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) # "msg 121" and "msg 127" produce commit IDs sharing the "3f47" 4-char # prefix when hashed with an empty manifest and 2026-01-01T00:00:00+00:00. # Verified by precomputation; changing _SNAP_ID or _COMMITTED_AT requires # updating these message strings. cid1 = _commit(repo, message="msg 121") cid2 = _commit(repo, message="msg 127") result = _rc(repo, "3f47") assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert "ambiguous" in data["error"] assert set(data["candidates"]) == {cid1, cid2} def test_missing_commit_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) # Valid hex ID that doesn't exist in the store result = _rc(repo, "dead" + "beef" * 15) assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert "not found" in data["error"] def test_invalid_commit_id_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _rc(repo, "ZZZZ" + "a" * 60) assert result.exit_code == ExitCode.USER_ERROR # --------------------------------------------------------------------------- # Integration — --fields filter # --------------------------------------------------------------------------- class TestFieldsFilter: def test_single_field(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="filtered") result = _rc(repo, "--fields", "message", cid) assert result.exit_code == 0 data = json.loads(result.output) assert list(data.keys()) == ["message"] assert data["message"] == "filtered" def test_multiple_fields(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, branch="dev", message="multi field test") result = _rc(repo, "--fields", "commit_id,branch,message", cid) data = json.loads(result.output) assert set(data.keys()) == {"commit_id", "branch", "message"} assert data["commit_id"] == cid assert data["branch"] == "dev" def test_agent_fields_filter(self, tmp_path: pathlib.Path) -> None: """Agents extracting provenance fields should get exactly what they asked for.""" repo = _make_repo(tmp_path) cid = _commit(repo, agent_id="audit-bot", model_id="claude-4") result = _rc(repo, "--fields", "agent_id,model_id,format_version", cid) data = json.loads(result.output) assert set(data.keys()) == {"agent_id", "model_id", "format_version"} assert data["agent_id"] == "audit-bot" assert data["model_id"] == "claude-4" def test_unknown_field_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="unknown field test") result = _rc(repo, "--fields", "nonexistent_field", cid) assert result.exit_code == ExitCode.USER_ERROR def test_fields_with_text_format_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="fields text error test") result = _rc(repo, "--fields", "commit_id", "--format", "text", cid) assert result.exit_code == ExitCode.USER_ERROR def test_fields_whitespace_trimmed(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="whitespace trim test") result = _rc(repo, "--fields", " commit_id , message ", cid) assert result.exit_code == 0 data = json.loads(result.output) assert "commit_id" in data assert "message" in data # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: def test_ansi_in_branch_stripped_in_text(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _commit(repo, branch="main") # Write an evil commit directly, bypassing the normal helper. # The commit_id must be a real hash of the stored fields for read_commit # to pass content-hash verification. from muse.core.store import write_commit as _wc evil_message = "test" evil_cid = compute_commit_id([], _SNAP_ID, evil_message, _COMMITTED_AT.isoformat()) evil_rec = CommitRecord( commit_id=evil_cid, repo_id="test-repo", branch="\x1b[31mevil\x1b[0m", snapshot_id=_SNAP_ID, message=evil_message, committed_at=_COMMITTED_AT, ) _wc(repo, evil_rec) result = _rc(repo, "--format", "text", evil_cid) assert result.exit_code == 0 assert "\x1b" not in result.output def test_ansi_in_message_stripped_in_text(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) # snapshot_id does not need to be a valid hex string for read_commit to # succeed — _verify_commit_id uses it as an opaque string in the hash. evil_snap_id = "s" * 64 evil_message = "\x1b[31mmalicious\x1b[0m" evil_committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) evil_cid = compute_commit_id([], evil_snap_id, evil_message, evil_committed_at.isoformat()) evil_rec = CommitRecord( commit_id=evil_cid, repo_id="test-repo", branch="main", snapshot_id=evil_snap_id, message=evil_message, committed_at=evil_committed_at, ) write_commit(repo, evil_rec) result = _rc(repo, "--format", "text", evil_cid) assert result.exit_code == 0 assert "\x1b" not in result.output def test_ansi_in_commit_id_rejected(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _rc(repo, "\x1b[31m" + "a" * 64) assert result.exit_code == ExitCode.USER_ERROR def test_no_traceback_on_bad_input(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _rc(repo, "not-valid") assert "Traceback" not in result.output # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestStress: def test_200_sequential_reads(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, message="stable") for i in range(200): result = _rc(repo, cid) assert result.exit_code == 0, f"failed at iteration {i}" assert json.loads(result.output)["message"] == "stable" def test_fields_filter_200_iterations(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, agent_id="bot") for i in range(200): result = _rc(repo, "--fields", "commit_id,agent_id", cid) assert result.exit_code == 0, f"failed at iteration {i}" data = json.loads(result.output) assert data["agent_id"] == "bot"