"""Comprehensive tests for ``muse plumbing 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.snapshot import compute_commit_id, compute_snapshot_id from muse.core.store import CommitRecord, SnapshotRecord, read_commit, write_commit, write_snapshot from muse.core._types import Manifest from tests.cli_test_helper import CliRunner, InvokeResult runner = CliRunner() # --------------------------------------------------------------------------- # 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 _snap(repo: pathlib.Path) -> str: manifest: Manifest = {} sid = compute_snapshot_id(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 = compute_commit_id(parent_ids, snap_id, message, committed_at.isoformat()) write_commit(repo, CommitRecord( commit_id=commit_id, repo_id="test-repo", 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", *args], env={"MUSE_REPO_ROOT": str(repo)}, ) # --------------------------------------------------------------------------- # Unit # --------------------------------------------------------------------------- class TestUnit: def test_format_choices(self) -> None: from muse.cli.commands.plumbing.commit_tree import _FORMAT_CHOICES assert "json" in _FORMAT_CHOICES assert "text" in _FORMAT_CHOICES # --------------------------------------------------------------------------- # 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 len(data["commit_id"]) == 64 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: repo = _make_repo(tmp_path) sid = _snap(repo) result = _ct(repo, "--format", "text", "--snapshot", sid) assert result.exit_code == 0 line = result.output.strip() assert len(line) == 64 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", "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", "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_unknown_format_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) sid = _snap(repo) result = _ct(repo, "--snapshot", sid, "--format", "msgpack") assert result.exit_code == ExitCode.USER_ERROR # --------------------------------------------------------------------------- # 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 len(data["commit_id"]) == 64