"""Tests for ``muse merge-base``. Verifies commit-ID resolution, branch-name resolution, HEAD resolution, text-format output, and error handling for unresolvable refs. """ from __future__ import annotations import datetime import json import pathlib import pytest import argparse from tests.cli_test_helper import CliRunner cli = None # argparse migration — CliRunner ignores this arg from muse.core.errors import ExitCode from muse.core.ids import hash_commit, hash_snapshot from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) from muse.core.types import Manifest, fake_id from muse.core.paths import head_path, muse_dir, ref_path runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _init_repo(path: pathlib.Path, domain: str = "midi") -> pathlib.Path: dot_muse = muse_dir(path) (dot_muse / "commits").mkdir(parents=True) (dot_muse / "snapshots").mkdir(parents=True) (dot_muse / "objects").mkdir(parents=True) (dot_muse / "refs" / "heads").mkdir(parents=True) (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8") (dot_muse / "repo.json").write_text( json.dumps({"repo_id": "test-repo", "domain": domain}), encoding="utf-8" ) return path def _env(repo: pathlib.Path) -> Manifest: return {"MUSE_REPO_ROOT": str(repo)} def _snap(repo: pathlib.Path) -> str: sid = hash_snapshot({}) write_snapshot( repo, SnapshotRecord( snapshot_id=sid, manifest={}, created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc), ), ) return sid def _commit( repo: pathlib.Path, tag: str, snap_id: str, branch: str = "main", parent: str | None = None, ) -> str: committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) parent_ids: list[str] = [parent] if parent else [] cid = hash_commit( parent_ids=parent_ids, snapshot_id=snap_id, message=tag, committed_at_iso=committed_at.isoformat(), author="tester", ) write_commit( repo, CommitRecord( commit_id=cid, branch=branch, snapshot_id=snap_id, message=tag, committed_at=committed_at, author="tester", parent_commit_id=parent, ), ) return cid def _set_branch(repo: pathlib.Path, branch: str, commit_id: str) -> None: ref = ref_path(repo, branch) ref.parent.mkdir(parents=True, exist_ok=True) ref.write_text(commit_id, encoding="utf-8") def _linear_dag(repo: pathlib.Path) -> tuple[str, str, str]: """Build A → B (main) and A → C (feat). Returns (A, B, C).""" sid = _snap(repo) cid_a = _commit(repo, "base", sid) cid_b = _commit(repo, "main-tip", sid, branch="main", parent=cid_a) cid_c = _commit(repo, "feat-tip", sid, branch="feat", parent=cid_a) _set_branch(repo, "main", cid_b) _set_branch(repo, "feat", cid_c) (head_path(repo)).write_text("ref: refs/heads/main", encoding="utf-8") return cid_a, cid_b, cid_c # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- class TestMergeBase: def test_finds_common_ancestor_by_commit_id(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) cid_a, cid_b, cid_c = _linear_dag(repo) result = runner.invoke(cli, ["merge-base", "--json", cid_b, cid_c], env=_env(repo)) assert result.exit_code == 0, result.output data = json.loads(result.stdout) assert data["merge_base"] == cid_a assert data["commit_a"] == cid_b assert data["commit_b"] == cid_c def test_branch_names_resolve_to_correct_base(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) cid_a, _b, _c = _linear_dag(repo) result = runner.invoke(cli, ["merge-base", "--json", "main", "feat"], env=_env(repo)) assert result.exit_code == 0, result.output assert json.loads(result.stdout)["merge_base"] == cid_a def test_head_resolves_to_current_branch(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) cid_a, _b, _c = _linear_dag(repo) result = runner.invoke(cli, ["merge-base", "--json", "HEAD", "feat"], env=_env(repo)) assert result.exit_code == 0, result.output assert json.loads(result.stdout)["merge_base"] == cid_a def test_same_commit_returns_itself(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) cid = _commit(repo, "solo", sid) _set_branch(repo, "main", cid) result = runner.invoke(cli, ["merge-base", "--json", cid, cid], env=_env(repo)) assert result.exit_code == 0, result.output assert json.loads(result.stdout)["merge_base"] == cid def test_text_format_emits_bare_commit_id(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) cid_a, cid_b, cid_c = _linear_dag(repo) # Default (no --json) emits plain text: just the commit ID result = runner.invoke( cli, ["merge-base", cid_b, cid_c], env=_env(repo) ) assert result.exit_code == 0, result.output assert cid_a in result.stdout def test_unresolvable_ref_a_exits_user_error(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) result = runner.invoke( cli, ["merge-base", "--json", "no-such-branch", "also-missing"], env=_env(repo) ) assert result.exit_code == ExitCode.USER_ERROR assert "error" in json.loads(result.stdout) def test_unknown_flag_rejected(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) cid = _commit(repo, "c", sid) _set_branch(repo, "main", cid) result = runner.invoke( cli, ["merge-base", "--format", "yaml", cid, cid], env=_env(repo) ) # --format flag no longer exists; argparse rejects it assert result.exit_code != 0 class TestRegisterFlags: def _parse(self, *args: str) -> "argparse.Namespace": import argparse from muse.cli.commands.merge_base import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) return p.parse_args(["merge-base", fake_id("a"), fake_id("b"), *args]) def test_json_short_flag(self) -> None: args = self._parse("-j") assert args.json_out is True def test_json_long_flag(self) -> None: args = self._parse("--json") assert args.json_out is True def test_default_no_json(self) -> None: args = self._parse() assert args.json_out is False