gabriel / muse public
test_cmd_merge_base.py python
200 lines 6.9 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 122 days ago
1 """Tests for ``muse merge-base``.
2
3 Verifies commit-ID resolution, branch-name resolution, HEAD resolution,
4 text-format output, and error handling for unresolvable refs.
5 """
6
7 from __future__ import annotations
8
9 import datetime
10 import json
11 import pathlib
12
13 import pytest
14 import argparse
15 from tests.cli_test_helper import CliRunner
16
17 cli = None # argparse migration — CliRunner ignores this arg
18 from muse.core.errors import ExitCode
19 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
20 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
21 from muse.core.types import Manifest, fake_id
22 from muse.core.paths import head_path, muse_dir, ref_path
23
24 runner = CliRunner()
25
26
27 # ---------------------------------------------------------------------------
28 # Helpers
29 # ---------------------------------------------------------------------------
30
31
32
33 def _init_repo(path: pathlib.Path, domain: str = "midi") -> pathlib.Path:
34 dot_muse = muse_dir(path)
35 (dot_muse / "commits").mkdir(parents=True)
36 (dot_muse / "snapshots").mkdir(parents=True)
37 (dot_muse / "objects").mkdir(parents=True)
38 (dot_muse / "refs" / "heads").mkdir(parents=True)
39 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
40 (dot_muse / "repo.json").write_text(
41 json.dumps({"repo_id": "test-repo", "domain": domain}), encoding="utf-8"
42 )
43 return path
44
45
46 def _env(repo: pathlib.Path) -> Manifest:
47 return {"MUSE_REPO_ROOT": str(repo)}
48
49
50 def _snap(repo: pathlib.Path) -> str:
51 sid = compute_snapshot_id({})
52 write_snapshot(
53 repo,
54 SnapshotRecord(
55 snapshot_id=sid,
56 manifest={},
57 created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
58 ),
59 )
60 return sid
61
62
63 def _commit(
64 repo: pathlib.Path,
65 tag: str,
66 snap_id: str,
67 branch: str = "main",
68 parent: str | None = None,
69 ) -> str:
70 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
71 parent_ids: list[str] = [parent] if parent else []
72 cid = compute_commit_id( parent_ids=parent_ids,
73 snapshot_id=snap_id,
74 message=tag,
75 committed_at_iso=committed_at.isoformat(),
76 author="tester",
77 )
78 write_commit(
79 repo,
80 CommitRecord(
81 commit_id=cid,
82 repo_id="test-repo",
83 branch=branch,
84 snapshot_id=snap_id,
85 message=tag,
86 committed_at=committed_at,
87 author="tester",
88 parent_commit_id=parent,
89 ),
90 )
91 return cid
92
93
94 def _set_branch(repo: pathlib.Path, branch: str, commit_id: str) -> None:
95 ref = ref_path(repo, branch)
96 ref.parent.mkdir(parents=True, exist_ok=True)
97 ref.write_text(commit_id, encoding="utf-8")
98
99
100 def _linear_dag(repo: pathlib.Path) -> tuple[str, str, str]:
101 """Build A → B (main) and A → C (feat). Returns (A, B, C)."""
102 sid = _snap(repo)
103 cid_a = _commit(repo, "base", sid)
104 cid_b = _commit(repo, "main-tip", sid, branch="main", parent=cid_a)
105 cid_c = _commit(repo, "feat-tip", sid, branch="feat", parent=cid_a)
106 _set_branch(repo, "main", cid_b)
107 _set_branch(repo, "feat", cid_c)
108 (head_path(repo)).write_text("ref: refs/heads/main", encoding="utf-8")
109 return cid_a, cid_b, cid_c
110
111
112 # ---------------------------------------------------------------------------
113 # Tests
114 # ---------------------------------------------------------------------------
115
116
117 class TestMergeBase:
118 def test_finds_common_ancestor_by_commit_id(self, tmp_path: pathlib.Path) -> None:
119 repo = _init_repo(tmp_path)
120 cid_a, cid_b, cid_c = _linear_dag(repo)
121 result = runner.invoke(cli, ["merge-base", "--json", cid_b, cid_c], env=_env(repo))
122 assert result.exit_code == 0, result.output
123 data = json.loads(result.stdout)
124 assert data["merge_base"] == cid_a
125 assert data["commit_a"] == cid_b
126 assert data["commit_b"] == cid_c
127
128 def test_branch_names_resolve_to_correct_base(self, tmp_path: pathlib.Path) -> None:
129 repo = _init_repo(tmp_path)
130 cid_a, _b, _c = _linear_dag(repo)
131 result = runner.invoke(cli, ["merge-base", "--json", "main", "feat"], env=_env(repo))
132 assert result.exit_code == 0, result.output
133 assert json.loads(result.stdout)["merge_base"] == cid_a
134
135 def test_head_resolves_to_current_branch(self, tmp_path: pathlib.Path) -> None:
136 repo = _init_repo(tmp_path)
137 cid_a, _b, _c = _linear_dag(repo)
138 result = runner.invoke(cli, ["merge-base", "--json", "HEAD", "feat"], env=_env(repo))
139 assert result.exit_code == 0, result.output
140 assert json.loads(result.stdout)["merge_base"] == cid_a
141
142 def test_same_commit_returns_itself(self, tmp_path: pathlib.Path) -> None:
143 repo = _init_repo(tmp_path)
144 sid = _snap(repo)
145 cid = _commit(repo, "solo", sid)
146 _set_branch(repo, "main", cid)
147 result = runner.invoke(cli, ["merge-base", "--json", cid, cid], env=_env(repo))
148 assert result.exit_code == 0, result.output
149 assert json.loads(result.stdout)["merge_base"] == cid
150
151 def test_text_format_emits_bare_commit_id(self, tmp_path: pathlib.Path) -> None:
152 repo = _init_repo(tmp_path)
153 cid_a, cid_b, cid_c = _linear_dag(repo)
154 # Default (no --json) emits plain text: just the commit ID
155 result = runner.invoke(
156 cli, ["merge-base", cid_b, cid_c], env=_env(repo)
157 )
158 assert result.exit_code == 0, result.output
159 assert cid_a in result.stdout
160
161 def test_unresolvable_ref_a_exits_user_error(self, tmp_path: pathlib.Path) -> None:
162 repo = _init_repo(tmp_path)
163 result = runner.invoke(
164 cli, ["merge-base", "--json", "no-such-branch", "also-missing"], env=_env(repo)
165 )
166 assert result.exit_code == ExitCode.USER_ERROR
167 assert "error" in json.loads(result.stdout)
168
169 def test_unknown_flag_rejected(self, tmp_path: pathlib.Path) -> None:
170 repo = _init_repo(tmp_path)
171 sid = _snap(repo)
172 cid = _commit(repo, "c", sid)
173 _set_branch(repo, "main", cid)
174 result = runner.invoke(
175 cli, ["merge-base", "--format", "yaml", cid, cid], env=_env(repo)
176 )
177 # --format flag no longer exists; argparse rejects it
178 assert result.exit_code != 0
179
180
181 class TestRegisterFlags:
182 def _parse(self, *args: str) -> "argparse.Namespace":
183 import argparse
184 from muse.cli.commands.merge_base import register
185 p = argparse.ArgumentParser()
186 subs = p.add_subparsers()
187 register(subs)
188 return p.parse_args(["merge-base", fake_id("a"), fake_id("b"), *args])
189
190 def test_json_short_flag(self) -> None:
191 args = self._parse("-j")
192 assert args.json_out is True
193
194 def test_json_long_flag(self) -> None:
195 args = self._parse("--json")
196 assert args.json_out is True
197
198 def test_default_no_json(self) -> None:
199 args = self._parse()
200 assert args.json_out is False
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 122 days ago