gabriel / muse public
test_merge_supercharge.py python
364 lines 14.3 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
1 """Supercharge tests for ``muse merge``.
2
3 Coverage tiers
4 --------------
5 - JSON envelope: exit_code and duration_ms always present on all outcome types
6 (merged, fast_forward, up_to_date, conflict)
7 - Error payload: errors go to stdout as JSON in --json mode, no dual stderr prose
8 - TypedDicts: _MergeJson and _MergeErrorJson exist with required annotations
9 - Docstring: module docstring covers exit_code and duration_ms
10 - No-prose pollution: no emoji in JSON stdout on success paths
11 """
12 from __future__ import annotations
13 from collections.abc import Mapping
14
15 import datetime
16 import json
17 import pathlib
18 from typing import get_type_hints
19
20 from muse.core.object_store import write_object
21 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
22 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
23 from tests.cli_test_helper import CliRunner, InvokeResult
24 from muse.core.types import blob_id, fake_id
25 from muse.core.paths import heads_dir, muse_dir, ref_path
26
27 runner = CliRunner()
28
29
30 # ---------------------------------------------------------------------------
31 # Helpers
32 # ---------------------------------------------------------------------------
33
34
35
36 def _env(root: pathlib.Path) -> Mapping[str, str]:
37 return {"MUSE_REPO_ROOT": str(root)}
38
39
40 def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
41 dot_muse = muse_dir(tmp_path)
42 dot_muse.mkdir()
43 repo_id = fake_id("repo")
44 (dot_muse / "repo.json").write_text(json.dumps({
45 "repo_id": repo_id, "domain": "code",
46 "default_branch": "main", "created_at": "2025-01-01T00:00:00+00:00",
47 }), encoding="utf-8")
48 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
49 (dot_muse / "refs" / "heads").mkdir(parents=True)
50 (dot_muse / "snapshots").mkdir()
51 (dot_muse / "commits").mkdir()
52 (dot_muse / "objects").mkdir()
53 return tmp_path, repo_id
54
55
56 def _make_commit(
57 root: pathlib.Path, repo_id: str, branch: str = "main",
58 message: str = "test", manifest: Mapping[str, object] | None = None,
59 ) -> str:
60 ref_file = ref_path(root, branch)
61 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
62 m = manifest or {}
63 snap_id = compute_snapshot_id(m)
64 committed_at = datetime.datetime.now(datetime.timezone.utc)
65 commit_id = compute_commit_id( parent_ids=[parent_id] if parent_id else [],
66 snapshot_id=snap_id, message=message,
67 committed_at_iso=committed_at.isoformat(),
68 )
69 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=m))
70 write_commit(root, CommitRecord(
71 commit_id=commit_id, repo_id=repo_id, branch=branch,
72 snapshot_id=snap_id, message=message, committed_at=committed_at,
73 parent_commit_id=parent_id,
74 ))
75 ref_file.parent.mkdir(parents=True, exist_ok=True)
76 ref_file.write_text(commit_id, encoding="utf-8")
77 return commit_id
78
79
80 def _write_obj(root: pathlib.Path, content: bytes) -> str:
81 oid = blob_id(content)
82 write_object(root, oid, content)
83 return oid
84
85
86 def _merge(root: pathlib.Path, *args: str) -> InvokeResult:
87 from muse.cli.app import main as cli
88 return runner.invoke(cli, ["merge", *args], env=_env(root))
89
90
91 # ---------------------------------------------------------------------------
92 # Repo fixtures
93 # ---------------------------------------------------------------------------
94
95 def _up_to_date_repo(tmp_path: pathlib.Path) -> pathlib.Path:
96 root, repo_id = _init_repo(tmp_path)
97 cid = _make_commit(root, repo_id, branch="main", message="base")
98 (heads_dir(root) / "feature").write_text(cid)
99 return root
100
101
102 def _ff_repo(tmp_path: pathlib.Path) -> pathlib.Path:
103 root, repo_id = _init_repo(tmp_path)
104 base = _make_commit(root, repo_id, branch="main", message="base")
105 (heads_dir(root) / "feature").write_text(base)
106 obj = _write_obj(root, b"new file")
107 _make_commit(root, repo_id, branch="feature", message="feat",
108 manifest={"new.py": obj})
109 return root
110
111
112 def _three_way_clean_repo(tmp_path: pathlib.Path) -> pathlib.Path:
113 root, repo_id = _init_repo(tmp_path)
114 base_obj = _write_obj(root, b"base")
115 base = _make_commit(root, repo_id, branch="main", message="base",
116 manifest={"base.py": base_obj})
117 (heads_dir(root) / "feature").write_text(base)
118 main_obj = _write_obj(root, b"main addition")
119 _make_commit(root, repo_id, branch="main", message="main work",
120 manifest={"base.py": base_obj, "main.py": main_obj})
121 feat_obj = _write_obj(root, b"feat addition")
122 _make_commit(root, repo_id, branch="feature", message="feat work",
123 manifest={"base.py": base_obj, "feat.py": feat_obj})
124 # Write working tree to match main HEAD so require_clean_workdir passes.
125 (root / "base.py").write_bytes(b"base")
126 (root / "main.py").write_bytes(b"main addition")
127 return root
128
129
130 def _conflict_repo(tmp_path: pathlib.Path) -> pathlib.Path:
131 root, repo_id = _init_repo(tmp_path)
132 shared_v1 = _write_obj(root, b"shared v1")
133 base = _make_commit(root, repo_id, branch="main", message="base",
134 manifest={"shared.py": shared_v1})
135 (heads_dir(root) / "feature").write_text(base)
136 shared_main = _write_obj(root, b"shared main version")
137 _make_commit(root, repo_id, branch="main", message="main mod",
138 manifest={"shared.py": shared_main})
139 shared_feat = _write_obj(root, b"shared feature version")
140 _make_commit(root, repo_id, branch="feature", message="feat mod",
141 manifest={"shared.py": shared_feat})
142 # Write working tree to match main HEAD so require_clean_workdir passes.
143 (root / "shared.py").write_bytes(b"shared main version")
144 return root
145
146
147 # ---------------------------------------------------------------------------
148 # JSON envelope — exit_code and duration_ms on all outcomes
149 # ---------------------------------------------------------------------------
150
151 class TestJsonEnvelopeExitCode:
152 """exit_code is present and correct across all merge outcome types."""
153
154 def test_up_to_date_has_exit_code(self, tmp_path: pathlib.Path) -> None:
155 root = _up_to_date_repo(tmp_path)
156 r = _merge(root, "feature", "--json")
157 d = json.loads(r.output)
158 assert "exit_code" in d, "exit_code missing from up_to_date envelope"
159 assert d["exit_code"] == 0
160
161 def test_fast_forward_has_exit_code(self, tmp_path: pathlib.Path) -> None:
162 root = _ff_repo(tmp_path)
163 r = _merge(root, "feature", "--json")
164 d = json.loads(r.output)
165 assert "exit_code" in d, "exit_code missing from fast_forward envelope"
166 assert d["exit_code"] == 0
167
168 def test_three_way_clean_has_exit_code(self, tmp_path: pathlib.Path) -> None:
169 root = _three_way_clean_repo(tmp_path)
170 r = _merge(root, "feature", "--json")
171 d = json.loads(r.output)
172 assert "exit_code" in d, "exit_code missing from merged envelope"
173 assert d["exit_code"] == 0
174
175 def test_conflict_has_exit_code(self, tmp_path: pathlib.Path) -> None:
176 root = _conflict_repo(tmp_path)
177 r = _merge(root, "feature", "--json")
178 d = json.loads(r.output)
179 assert "exit_code" in d, "exit_code missing from conflict envelope"
180 assert d["exit_code"] != 0
181
182 def test_dry_run_merged_has_exit_code(self, tmp_path: pathlib.Path) -> None:
183 root = _three_way_clean_repo(tmp_path)
184 r = _merge(root, "feature", "--dry-run", "--json")
185 d = json.loads(r.output)
186 assert "exit_code" in d
187 assert d["exit_code"] == 0
188
189 def test_dry_run_conflict_has_exit_code(self, tmp_path: pathlib.Path) -> None:
190 root = _conflict_repo(tmp_path)
191 r = _merge(root, "feature", "--dry-run", "--json")
192 d = json.loads(r.output)
193 assert "exit_code" in d
194 assert d["exit_code"] != 0
195
196
197 class TestJsonEnvelopeDurationMs:
198 """duration_ms is present and is a non-negative float on all outcome types."""
199
200 def test_up_to_date_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
201 root = _up_to_date_repo(tmp_path)
202 r = _merge(root, "feature", "--json")
203 d = json.loads(r.output)
204 assert "duration_ms" in d
205 assert isinstance(d["duration_ms"], float)
206 assert d["duration_ms"] >= 0.0
207
208 def test_fast_forward_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
209 root = _ff_repo(tmp_path)
210 r = _merge(root, "feature", "--json")
211 d = json.loads(r.output)
212 assert "duration_ms" in d
213 assert isinstance(d["duration_ms"], float)
214
215 def test_three_way_clean_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
216 root = _three_way_clean_repo(tmp_path)
217 r = _merge(root, "feature", "--json")
218 d = json.loads(r.output)
219 assert "duration_ms" in d
220 assert isinstance(d["duration_ms"], float)
221
222 def test_conflict_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
223 root = _conflict_repo(tmp_path)
224 r = _merge(root, "feature", "--json")
225 d = json.loads(r.output)
226 assert "duration_ms" in d
227 assert isinstance(d["duration_ms"], float)
228
229 def test_dry_run_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
230 root = _ff_repo(tmp_path)
231 r = _merge(root, "feature", "--dry-run", "--json")
232 d = json.loads(r.output)
233 assert "duration_ms" in d
234
235
236 # ---------------------------------------------------------------------------
237 # Error payload — errors go to stdout as JSON in --json mode
238 # ---------------------------------------------------------------------------
239
240 class TestErrorPayload:
241 """Errors emit {status: "error", error: "...", exit_code: N} on stdout in --json mode."""
242
243 def test_no_branch_error_is_json(self, tmp_path: pathlib.Path) -> None:
244 root, _ = _init_repo(tmp_path)
245 r = _merge(root, "--json") # no branch arg
246 assert r.exit_code != 0
247 d = json.loads(r.output)
248 assert d["status"] == "error"
249
250 def test_self_merge_error_is_json(self, tmp_path: pathlib.Path) -> None:
251 root, repo_id = _init_repo(tmp_path)
252 _make_commit(root, repo_id, branch="main")
253 r = _merge(root, "main", "--json")
254 assert r.exit_code != 0
255 d = json.loads(r.output)
256 assert d["status"] == "error"
257
258 def test_self_merge_no_duplicate_stderr_prose(self, tmp_path: pathlib.Path) -> None:
259 """In --json mode, errors should not also print emoji prose to stderr."""
260 root, repo_id = _init_repo(tmp_path)
261 _make_commit(root, repo_id, branch="main")
262 r = _merge(root, "main", "--json")
263 assert "❌" not in r.stderr
264
265 def test_error_payload_has_status_error(self, tmp_path: pathlib.Path) -> None:
266 root, _ = _init_repo(tmp_path)
267 r = _merge(root, "--json")
268 d = json.loads(r.output)
269 assert d["status"] == "error"
270
271 def test_error_payload_has_exit_code(self, tmp_path: pathlib.Path) -> None:
272 root, _ = _init_repo(tmp_path)
273 r = _merge(root, "--json")
274 d = json.loads(r.output)
275 assert "exit_code" in d
276 assert d["exit_code"] != 0
277
278 def test_error_payload_has_error_field(self, tmp_path: pathlib.Path) -> None:
279 root, _ = _init_repo(tmp_path)
280 r = _merge(root, "--json")
281 d = json.loads(r.output)
282 assert "error" in d
283 assert d["error"] # non-empty message
284
285 def test_unknown_branch_error_is_json(self, tmp_path: pathlib.Path) -> None:
286 root, repo_id = _init_repo(tmp_path)
287 _make_commit(root, repo_id, branch="main")
288 r = _merge(root, "no-such-branch", "--json")
289 assert r.exit_code != 0
290 d = json.loads(r.output)
291 assert d["status"] == "error"
292
293
294 # ---------------------------------------------------------------------------
295 # No-prose pollution
296 # ---------------------------------------------------------------------------
297
298 class TestNoProsePollution:
299 def test_up_to_date_stdout_valid_json(self, tmp_path: pathlib.Path) -> None:
300 root = _up_to_date_repo(tmp_path)
301 r = _merge(root, "feature", "--json")
302 json.loads(r.output) # must not raise
303
304 def test_fast_forward_stdout_valid_json(self, tmp_path: pathlib.Path) -> None:
305 root = _ff_repo(tmp_path)
306 r = _merge(root, "feature", "--json")
307 json.loads(r.output)
308
309 def test_merged_stdout_valid_json(self, tmp_path: pathlib.Path) -> None:
310 root = _three_way_clean_repo(tmp_path)
311 r = _merge(root, "feature", "--json")
312 json.loads(r.output)
313
314 def test_no_emoji_in_merged_json(self, tmp_path: pathlib.Path) -> None:
315 root = _three_way_clean_repo(tmp_path)
316 r = _merge(root, "feature", "--json")
317 assert "✅" not in r.output
318 assert "❌" not in r.output
319
320
321 # ---------------------------------------------------------------------------
322 # TypedDicts
323 # ---------------------------------------------------------------------------
324
325 class TestTypedDicts:
326 def test_merge_json_typeddict_exists(self) -> None:
327 from muse.cli.commands.merge import _MergeJson
328 assert _MergeJson is not None
329
330 def test_merge_error_json_typeddict_exists(self) -> None:
331 from muse.cli.commands.merge import _MergeErrorJson
332 assert _MergeErrorJson is not None
333
334 def test_merge_json_has_exit_code_annotation(self) -> None:
335 from muse.cli.commands.merge import _MergeJson
336 hints = get_type_hints(_MergeJson)
337 assert "exit_code" in hints
338
339 def test_merge_json_has_duration_ms_annotation(self) -> None:
340 from muse.cli.commands.merge import _MergeJson
341 hints = get_type_hints(_MergeJson)
342 assert "duration_ms" in hints
343
344 def test_merge_error_json_has_required_fields(self) -> None:
345 from muse.cli.commands.merge import _MergeErrorJson
346 hints = get_type_hints(_MergeErrorJson)
347 for field in ("status", "error", "exit_code"):
348 assert field in hints, f"Missing annotation: {field!r}"
349
350
351 # ---------------------------------------------------------------------------
352 # Docstring coverage
353 # ---------------------------------------------------------------------------
354
355 class TestDocstring:
356 def _doc(self) -> str:
357 import muse.cli.commands.merge as mod
358 return mod.__doc__ or ""
359
360 def test_docstring_documents_exit_code(self) -> None:
361 assert "exit_code" in self._doc()
362
363 def test_docstring_documents_duration_ms(self) -> None:
364 assert "duration_ms" in self._doc()
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago