"""Comprehensive tests for ``muse plumbing rev-parse``. Coverage tiers -------------- - Integration: branch, HEAD, SHA prefix, full SHA, --abbrev-ref, --format text - Edge cases: empty repo (no commits), empty ref, ambiguous prefix, HEAD→branch - Security: ANSI/control chars in ref → JSON-escaped, empty ref clean error - Stress: 200 rapid resolves """ from __future__ import annotations import datetime import json import pathlib from muse.core.errors import ExitCode from muse.core.object_store import write_object from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.store import CommitRecord, SnapshotRecord, 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, branch: str = "main") -> 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(f"ref: refs/heads/{branch}") (muse / "repo.json").write_text(json.dumps({"repo_id": "test", "domain": "code"})) return repo _TS = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) def _store_snap(repo: pathlib.Path, manifest: Manifest | None = None) -> str: sid = compute_snapshot_id(manifest or {}) write_snapshot(repo, SnapshotRecord( snapshot_id=sid, manifest=manifest or {}, created_at=_TS, )) return sid def _make_commit( repo: pathlib.Path, snapshot_id: str, *, branch: str = "main", parent: str | None = None, message: str = "test", ) -> str: parents = [parent] if parent else [] cid = compute_commit_id(parents, snapshot_id, message, _TS.isoformat()) rec = CommitRecord( commit_id=cid, repo_id="test-repo-id", branch=branch, snapshot_id=snapshot_id, message=message, committed_at=_TS, author="tester", parent_commit_id=parent, ) write_commit(repo, rec) return cid def _set_head(repo: pathlib.Path, branch: str, commit_id: str) -> None: ref = repo / ".muse" / "refs" / "heads" / branch ref.parent.mkdir(parents=True, exist_ok=True) ref.write_text(commit_id) def _rev(repo: pathlib.Path, *args: str) -> InvokeResult: from muse.cli.app import main as cli return runner.invoke( cli, ["rev-parse", *args], env={"MUSE_REPO_ROOT": str(repo)}, ) def _populated_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]: """Return (repo, commit_id) with one commit on main, using real content-addressed IDs.""" repo = _make_repo(tmp_path) sid = _store_snap(repo) cid = _make_commit(repo, sid) _set_head(repo, "main", cid) return repo, cid # --------------------------------------------------------------------------- # Integration — branch resolution # --------------------------------------------------------------------------- class TestBranchResolution: def test_resolve_branch_json(self, tmp_path: pathlib.Path) -> None: repo, cid = _populated_repo(tmp_path) result = _rev(repo, "main") assert result.exit_code == 0 data = json.loads(result.output) assert data["commit_id"] == cid assert data["ref"] == "main" def test_resolve_branch_text(self, tmp_path: pathlib.Path) -> None: repo, cid = _populated_repo(tmp_path) result = _rev(repo, "--format", "text", "main") assert result.exit_code == 0 assert result.output.strip() == cid def test_json_flag_shorthand(self, tmp_path: pathlib.Path) -> None: repo, cid = _populated_repo(tmp_path) result = _rev(repo, "--json", "main") assert result.exit_code == 0 data = json.loads(result.output) assert data["commit_id"] == cid def test_unknown_branch_not_found(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _rev(repo, "nonexistent-branch") assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert data["commit_id"] is None assert data["error"] == "not found" # --------------------------------------------------------------------------- # Integration — HEAD resolution # --------------------------------------------------------------------------- class TestHeadResolution: def test_resolve_head(self, tmp_path: pathlib.Path) -> None: repo, cid = _populated_repo(tmp_path) result = _rev(repo, "HEAD") assert result.exit_code == 0 data = json.loads(result.output) assert data["commit_id"] == cid def test_head_lowercase_also_resolves(self, tmp_path: pathlib.Path) -> None: """HEAD resolution is case-insensitive (matches git behaviour).""" repo, cid = _populated_repo(tmp_path) result = _rev(repo, "head") assert result.exit_code == 0 data = json.loads(result.output) assert data["commit_id"] == cid def test_head_on_empty_repo_errors(self, tmp_path: pathlib.Path) -> None: """HEAD on a repo with no commits should error cleanly.""" repo = _make_repo(tmp_path) result = _rev(repo, "HEAD") assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert data["commit_id"] is None assert "no commits" in data["error"] # --------------------------------------------------------------------------- # Integration — SHA prefix resolution # --------------------------------------------------------------------------- class TestShaResolution: def test_resolve_full_sha(self, tmp_path: pathlib.Path) -> None: repo, cid = _populated_repo(tmp_path) result = _rev(repo, cid) assert result.exit_code == 0 data = json.loads(result.output) assert data["commit_id"] == cid def test_resolve_8char_prefix(self, tmp_path: pathlib.Path) -> None: repo, cid = _populated_repo(tmp_path) result = _rev(repo, cid[:8]) assert result.exit_code == 0 data = json.loads(result.output) assert data["commit_id"] == cid def test_ambiguous_prefix_returns_candidates(self, tmp_path: pathlib.Path) -> None: """Two commits sharing a prefix → error with candidates list.""" # Messages "commit-search-165" and "commit-search-106" produce IDs # sharing the 4-char prefix "9f7c" (same snapshot, same timestamp). _AMBIG_MSG_1 = "commit-search-165" _AMBIG_MSG_2 = "commit-search-106" _AMBIG_PREFIX = "9f7c" repo = _make_repo(tmp_path) sid = _store_snap(repo) cid1 = _make_commit(repo, sid, branch="main", message=_AMBIG_MSG_1) cid2 = _make_commit(repo, sid, branch="dev", message=_AMBIG_MSG_2) assert cid1[:4] == cid2[:4] == _AMBIG_PREFIX _set_head(repo, "main", cid1) _set_head(repo, "dev", cid2) result = _rev(repo, _AMBIG_PREFIX) assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert data["error"] == "ambiguous" assert set(data["candidates"]) == {cid1, cid2} def test_nonexistent_full_sha_not_found(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _rev(repo, "f" * 64) assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert data["error"] == "not found" # --------------------------------------------------------------------------- # Integration — --abbrev-ref # --------------------------------------------------------------------------- class TestAbbrevRef: def test_abbrev_ref_head_returns_branch_name(self, tmp_path: pathlib.Path) -> None: """The canonical agent UX: what branch am I on?""" repo = _make_repo(tmp_path, branch="feat/my-feature") result = _rev(repo, "--abbrev-ref", "HEAD") assert result.exit_code == 0 data = json.loads(result.output) assert data["branch"] == "feat/my-feature" assert data["ref"] == "HEAD" def test_abbrev_ref_text_format(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path, branch="dev") result = _rev(repo, "--abbrev-ref", "--format", "text", "HEAD") assert result.exit_code == 0 assert result.output.strip() == "dev" def test_abbrev_ref_main(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path, branch="main") result = _rev(repo, "--abbrev-ref", "HEAD") assert result.exit_code == 0 assert json.loads(result.output)["branch"] == "main" # --------------------------------------------------------------------------- # Edge cases # --------------------------------------------------------------------------- class TestEdgeCases: def test_empty_ref_clean_error(self, tmp_path: pathlib.Path) -> None: """Empty string ref must give a clear 'ref must not be empty' error.""" repo = _make_repo(tmp_path) result = _rev(repo, "") assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert "empty" in data["error"] def test_invalid_format_errors(self, tmp_path: pathlib.Path) -> None: repo, _ = _populated_repo(tmp_path) result = _rev(repo, "--format", "xml", "main") assert result.exit_code == ExitCode.USER_ERROR def test_branch_with_slash_resolves(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path, branch="feat/my-feature") sid = _store_snap(repo) cid = _make_commit(repo, sid, branch="feat/my-feature", message="feat-init") _set_head(repo, "feat/my-feature", cid) result = _rev(repo, "feat/my-feature") assert result.exit_code == 0 assert json.loads(result.output)["commit_id"] == cid # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: def test_ansi_in_ref_is_json_escaped(self, tmp_path: pathlib.Path) -> None: """ANSI escape in ref is safely JSON-encoded, never echoed raw.""" repo = _make_repo(tmp_path) evil = "\x1b[31mevil\x1b[0m" result = _rev(repo, evil) assert result.exit_code == ExitCode.USER_ERROR # Output is JSON — ANSI must be encoded as \u001b, not emitted raw assert "\x1b" not in result.output data = json.loads(result.output) assert data["error"] == "not found" def test_path_traversal_ref_gives_not_found(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _rev(repo, "../../../etc/passwd") assert result.exit_code == ExitCode.USER_ERROR def test_null_byte_in_ref(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _rev(repo, "branch\x00null") 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 = _rev(repo, "") assert "Traceback" not in result.output # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestStress: def test_200_rapid_resolves(self, tmp_path: pathlib.Path) -> None: repo, cid = _populated_repo(tmp_path) for i in range(200): result = _rev(repo, "main") assert result.exit_code == 0, f"failed at iteration {i}" assert json.loads(result.output)["commit_id"] == cid def test_200_abbrev_ref_resolves(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path, branch="dev") for i in range(200): result = _rev(repo, "--abbrev-ref", "HEAD") assert result.exit_code == 0, f"failed at iteration {i}" assert json.loads(result.output)["branch"] == "dev"