"""Comprehensive tests for ``muse commit-tree``. Coverage tiers -------------- - Unit: _FORMAT_CHOICES - Integration: basic creation, --parent chain, merge commit, text format, --agent-id/--model-id/--toolchain-id provenance, --branch override - Security: >2 parents silently rejected, errors to stderr, no traceback - Stress: 200 sequential commit-tree calls """ from __future__ import annotations import datetime import json import pathlib from muse.core.errors import ExitCode from muse.core.ids import hash_commit, hash_snapshot from muse.core.commits import ( CommitRecord, read_commit, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) from muse.core.types import Manifest, fake_id from muse.core.paths import muse_dir from tests.cli_test_helper import CliRunner, InvokeResult runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: repo = tmp_path / "repo" dot_muse = muse_dir(repo) for sub in ("objects", "commits", "snapshots", "refs/heads"): (dot_muse / sub).mkdir(parents=True) (dot_muse / "HEAD").write_text("ref: refs/heads/main") (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"})) return repo def _snap(repo: pathlib.Path) -> str: manifest: Manifest = {} sid = hash_snapshot(manifest) write_snapshot(repo, SnapshotRecord( snapshot_id=sid, manifest=manifest, created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc), )) return sid def _commit( repo: pathlib.Path, snap_id: str, parent: str | None = None, message: str = "parent", ) -> str: committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) parent_ids: list[str] = [parent] if parent else [] commit_id = hash_commit( parent_ids=parent_ids, snapshot_id=snap_id, message=message, committed_at_iso=committed_at.isoformat(), ) write_commit(repo, CommitRecord( commit_id=commit_id, branch="main", snapshot_id=snap_id, message=message, committed_at=committed_at, parent_commit_id=parent, )) return commit_id def _ct(repo: pathlib.Path, *args: str) -> InvokeResult: from muse.cli.app import main as cli return runner.invoke( cli, ["commit-tree", "--json", *args], env={"MUSE_REPO_ROOT": str(repo)}, ) # --------------------------------------------------------------------------- # Unit # --------------------------------------------------------------------------- class TestUnit: def test_json_flag_registered(self) -> None: from muse.cli.commands.commit_tree import register import argparse p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) ns = p.parse_args(["commit-tree", "--snapshot", fake_id("a"), "--json"]) assert ns.json_out is True # --------------------------------------------------------------------------- # Integration — basic creation # --------------------------------------------------------------------------- class TestBasicCreation: def test_creates_commit(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) result = _ct(repo, "--snapshot", sid, "--message", "first commit") assert result.exit_code == 0 data = json.loads(result.output) assert "commit_id" in data assert data["commit_id"].startswith("sha256:") assert len(data["commit_id"]) == 71 def test_commit_persisted_in_store(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid).output) cid = data["commit_id"] rec = read_commit(repo, cid) assert rec is not None assert rec.snapshot_id == sid def test_json_flag_shorthand(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) result = _ct(repo, "--json", "--snapshot", sid) assert result.exit_code == 0 assert "commit_id" in json.loads(result.output) def test_text_format_bare_commit_id(self, tmp_path: pathlib.Path) -> None: from muse.cli.app import main as cli repo = _make_repo(tmp_path) sid = _snap(repo) result = runner.invoke( cli, ["commit-tree", "--snapshot", sid], env={"MUSE_REPO_ROOT": str(repo)}, ) assert result.exit_code == 0 line = result.output.strip() assert line.startswith("sha256:") assert len(line) == 71 def test_message_stored(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid, "--message", "my msg").output) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.message == "my msg" def test_author_stored(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid, "--author", "gabriel").output) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.author == "gabriel" # --------------------------------------------------------------------------- # Integration — parent chain # --------------------------------------------------------------------------- class TestParentChain: def test_single_parent_stored(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) p1_id = _commit(repo, sid) data = json.loads(_ct(repo, "--snapshot", sid, "--parent", p1_id).output) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.parent_commit_id == p1_id assert rec.parent2_commit_id is None def test_merge_commit_two_parents(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) p1 = _commit(repo, sid, message="parent1") p2 = _commit(repo, sid, message="parent2") data = json.loads( _ct(repo, "--snapshot", sid, "--parent", p1, "--parent", p2).output ) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.parent_commit_id == p1 assert rec.parent2_commit_id == p2 def test_three_parents_rejected(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) p = _commit(repo, sid) result = _ct( repo, "--snapshot", sid, "--parent", p, "--parent", p, "--parent", p, ) assert result.exit_code == ExitCode.USER_ERROR def test_missing_parent_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) result = _ct(repo, "--snapshot", sid, "--parent", f"dead{'beef' * 15}") assert result.exit_code == ExitCode.USER_ERROR # --------------------------------------------------------------------------- # Integration — agent provenance flags # --------------------------------------------------------------------------- class TestAgentProvenance: def test_agent_id_stored(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads( _ct(repo, "--snapshot", sid, "--agent-id", "my-bot").output ) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.agent_id == "my-bot" def test_model_id_stored(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads( _ct(repo, "--snapshot", sid, "--model-id", "claude-opus-4").output ) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.model_id == "claude-opus-4" def test_toolchain_id_stored(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads( _ct(repo, "--snapshot", sid, "--toolchain-id", "cursor-agent-v2").output ) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.toolchain_id == "cursor-agent-v2" def test_full_provenance_round_trip(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct( repo, "--snapshot", sid, "--agent-id", "audit-bot", "--model-id", "claude-4", "--toolchain-id", "muse-agent-v1", ).output) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.agent_id == "audit-bot" assert rec.model_id == "claude-4" assert rec.toolchain_id == "muse-agent-v1" def test_defaults_to_empty_strings(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid).output) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.agent_id == "" assert rec.model_id == "" assert rec.toolchain_id == "" # --------------------------------------------------------------------------- # Integration — --branch override # --------------------------------------------------------------------------- class TestBranchOverride: def test_branch_override_stored(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid, "--branch", "feat/x").output) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.branch == "feat/x" # --------------------------------------------------------------------------- # Error cases # --------------------------------------------------------------------------- class TestErrors: def test_missing_snapshot_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ct(repo, "--snapshot", f"dead{'beef' * 15}") assert result.exit_code == ExitCode.USER_ERROR def test_invalid_snapshot_id_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ct(repo, "--snapshot", "not-hex") assert result.exit_code == ExitCode.USER_ERROR def test_invalid_parent_id_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) result = _ct(repo, "--snapshot", sid, "--parent", "bad-hex") assert result.exit_code == ExitCode.USER_ERROR def test_missing_snapshot_arg_errors(self, tmp_path: pathlib.Path) -> None: from muse.cli.app import main as cli repo = _make_repo(tmp_path) result = runner.invoke( cli, ["commit-tree"], env={"MUSE_REPO_ROOT": str(repo)}, ) assert result.exit_code != 0 # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: def test_no_traceback_on_bad_snapshot(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ct(repo, "--snapshot", "bad") assert "Traceback" not in result.output def test_no_traceback_on_too_many_parents(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) p = _commit(repo, sid) result = _ct(repo, "--snapshot", sid, "--parent", p, "--parent", p, "--parent", p) assert "Traceback" not in result.output # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestStress: def test_200_sequential_commits(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) for i in range(200): result = _ct(repo, "--snapshot", sid, "--message", f"commit {i}") assert result.exit_code == 0, f"failed at iteration {i}" data = json.loads(result.output) assert data["commit_id"].startswith("sha256:") assert len(data["commit_id"]) == 71 # --------------------------------------------------------------------------- # Supercharge — full JSON schema # --------------------------------------------------------------------------- _FULL_KEYS = frozenset({ "commit_id", "snapshot_id", "branch", "message", "committed_at", "author", "agent_id", "model_id", "toolchain_id", "parent_commit_id", "parent2_commit_id", "duration_ms", "exit_code", }) class TestJsonSchemaComplete: """JSON output must carry the full commit record so agents need no follow-up read.""" def test_all_keys_present_on_success(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid, "--message", "m").output) assert _FULL_KEYS <= set(data.keys()), ( f"Missing keys: {_FULL_KEYS - set(data.keys())}" ) def test_snapshot_id_echoed(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid).output) assert data["snapshot_id"] == sid def test_branch_echoed(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid, "--branch", "feat/x").output) assert data["branch"] == "feat/x" def test_message_echoed(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid, "--message", "hello world").output) assert data["message"] == "hello world" def test_author_echoed(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid, "--author", "gabriel").output) assert data["author"] == "gabriel" def test_agent_id_echoed(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid, "--agent-id", "bot-x").output) assert data["agent_id"] == "bot-x" def test_model_id_echoed(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid, "--model-id", "claude-opus-4").output) assert data["model_id"] == "claude-opus-4" def test_toolchain_id_echoed(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid, "--toolchain-id", "v2").output) assert data["toolchain_id"] == "v2" def test_parent_commit_id_null_when_no_parent(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid).output) assert data["parent_commit_id"] is None def test_parent_commit_id_present(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) p1 = _commit(repo, sid) data = json.loads(_ct(repo, "--snapshot", sid, "--parent", p1).output) assert data["parent_commit_id"] == p1 def test_parent2_commit_id_null_when_no_second_parent(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid).output) assert data["parent2_commit_id"] is None def test_parent2_commit_id_present_for_merge(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) p1 = _commit(repo, sid, message="p1") p2 = _commit(repo, sid, message="p2") data = json.loads(_ct(repo, "--snapshot", sid, "--parent", p1, "--parent", p2).output) assert data["parent2_commit_id"] == p2 def test_committed_at_is_iso_string(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid).output) ts = data["committed_at"] assert isinstance(ts, str) # Must parse as ISO datetime datetime.datetime.fromisoformat(ts) def test_exit_code_zero(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid).output) assert data["exit_code"] == 0 # --------------------------------------------------------------------------- # Supercharge — duration_ms # --------------------------------------------------------------------------- class TestElapsed: def test_elapsed_present(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid).output) assert "duration_ms" in data def test_elapsed_is_float(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid).output) assert isinstance(data["duration_ms"], float) def test_elapsed_non_negative(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid).output) assert data["duration_ms"] >= 0.0 # --------------------------------------------------------------------------- # Supercharge — exit_code # --------------------------------------------------------------------------- class TestExitCode: def test_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid).output) assert data["exit_code"] == 0 def test_process_exit_zero_on_success(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) result = _ct(repo, "--snapshot", sid) assert result.exit_code == 0 # --------------------------------------------------------------------------- # Supercharge — data integrity # --------------------------------------------------------------------------- class TestDataIntegrity: def test_commit_id_roundtrips_via_store(self, tmp_path: pathlib.Path) -> None: """commit_id in JSON matches what was actually written to the store.""" repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid, "--message", "integrity check").output) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.commit_id == data["commit_id"] assert rec.snapshot_id == data["snapshot_id"] assert rec.message == data["message"] def test_snapshot_id_matches_store(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid).output) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.snapshot_id == sid def test_parent_id_matches_store(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) p1 = _commit(repo, sid) data = json.loads(_ct(repo, "--snapshot", sid, "--parent", p1).output) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.parent_commit_id == data["parent_commit_id"] == p1 def test_provenance_matches_store(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct( repo, "--snapshot", sid, "--agent-id", "integrity-bot", "--model-id", "claude-sonnet-4-6", "--toolchain-id", "test-chain", ).output) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.agent_id == data["agent_id"] == "integrity-bot" assert rec.model_id == data["model_id"] == "claude-sonnet-4-6" assert rec.toolchain_id == data["toolchain_id"] == "test-chain" # --------------------------------------------------------------------------- # Supercharge — security (ANSI injection) # --------------------------------------------------------------------------- class TestSecurityAnsi: def test_ansi_in_message_does_not_corrupt_json(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) malicious_message = "feat: \x1b[31mred\x1b[0m alert" result = _ct(repo, "--snapshot", sid, "--message", malicious_message) assert result.exit_code == 0 data = json.loads(result.output) assert data["message"] == malicious_message def test_ansi_in_author_does_not_corrupt_json(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) malicious_author = "\x1b[32mgabriel\x1b[0m" result = _ct(repo, "--snapshot", sid, "--author", malicious_author) assert result.exit_code == 0 data = json.loads(result.output) assert data["author"] == malicious_author def test_long_message_handled(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) long_msg = "x" * 10_000 data = json.loads(_ct(repo, "--snapshot", sid, "--message", long_msg).output) rec = read_commit(repo, data["commit_id"]) assert rec is not None assert rec.message == long_msg # --------------------------------------------------------------------------- # Supercharge — performance # --------------------------------------------------------------------------- class TestPerformance: def test_single_commit_under_500ms(self, tmp_path: pathlib.Path) -> None: import time repo = _make_repo(tmp_path) sid = _snap(repo) t0 = time.monotonic() result = _ct(repo, "--snapshot", sid) elapsed = time.monotonic() - t0 assert result.exit_code == 0 assert elapsed < 0.5, f"commit-tree took {elapsed:.3f}s — too slow" def test_duration_ms_reasonable(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) data = json.loads(_ct(repo, "--snapshot", sid).output) assert data["duration_ms"] < 50.0, ( f"reported elapsed {data['duration_ms']}ms — implausibly slow" ) # --------------------------------------------------------------------------- # Flag registration tests # --------------------------------------------------------------------------- import argparse as _argparse from muse.cli.commands.commit_tree import register as _register_commit_tree def _parse_ct(*args: str) -> _argparse.Namespace: root_p = _argparse.ArgumentParser() subs = root_p.add_subparsers(dest="cmd") _register_commit_tree(subs) return root_p.parse_args(["commit-tree", *args]) class TestRegisterFlags: def test_default_json_out_is_false(self) -> None: ns = _parse_ct("--snapshot", fake_id("a")) assert ns.json_out is False def test_json_flag_sets_json_out(self) -> None: ns = _parse_ct("--snapshot", fake_id("a"), "--json") assert ns.json_out is True def test_j_shorthand_sets_json_out(self) -> None: ns = _parse_ct("--snapshot", fake_id("a"), "-j") assert ns.json_out is True def test_format_flag_no_longer_exists(self) -> None: import pytest with pytest.raises(SystemExit): _parse_ct("--snapshot", fake_id("a"), "--format", "json")