gabriel / muse public
test_merge_tree_supercharge.py python
389 lines 14.9 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago
1 """Supercharge tests for ``muse merge-tree``.
2
3 Coverage tiers
4 --------------
5 - JSON envelope: exit_code and duration_ms present on clean and conflict outcomes
6 - Error payload: errors go to stdout as JSON in --json mode, no dual stderr prose
7 - Data integrity: sha256: OID prefix preserved in merged_manifest; --base with
8 sha256:-prefixed commit ID accepted
9 - TypedDicts: _MergeTreeJson and _MergeTreeErrorJson with required annotations
10 - Docstring: module docstring covers exit_code and duration_ms
11 - No-prose pollution: JSON stdout is valid on all non-error paths
12 - Stress: 100-file manifest, 40% conflict rate — correct counts, correct exit code
13 """
14 from __future__ import annotations
15
16 import datetime
17 import hashlib
18 import json
19 import pathlib
20 from typing import get_type_hints
21
22 from muse.core.object_store import write_object
23 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
24 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
25 from tests.cli_test_helper import CliRunner
26 from muse.core._types import long_id
27
28 runner = CliRunner()
29
30 _REPO_ID = "mt-supercharge"
31 _DT = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
32
33
34 # ---------------------------------------------------------------------------
35 # Helpers
36 # ---------------------------------------------------------------------------
37
38
39 def _sha(data: bytes) -> str:
40 return long_id(hashlib.sha256(data).hexdigest())
41
42
43 def _init_repo(tmp_path: pathlib.Path) -> pathlib.Path:
44 muse = tmp_path / ".muse"
45 for sub in ("objects", "commits", "snapshots", "refs/heads"):
46 (muse / sub).mkdir(parents=True)
47 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
48 (muse / "repo.json").write_text(
49 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
50 )
51 return tmp_path
52
53
54 def _env(root: pathlib.Path) -> dict[str, str]:
55 return {"MUSE_REPO_ROOT": str(root)}
56
57
58 def _write_obj(root: pathlib.Path, content: bytes) -> str:
59 oid = _sha(content)
60 write_object(root, oid, content)
61 return oid
62
63
64 def _make_commit(
65 root: pathlib.Path,
66 manifest: dict[str, str],
67 branch: str,
68 parent: str | None = None,
69 msg: str = "test",
70 ) -> str:
71 snap_id = compute_snapshot_id(manifest)
72 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest, created_at=_DT))
73 parent_ids = [parent] if parent else []
74 cid = compute_commit_id(parent_ids, snap_id, msg, _DT.isoformat())
75 write_commit(root, CommitRecord(
76 commit_id=cid, repo_id=_REPO_ID, branch=branch,
77 snapshot_id=snap_id, message=msg, committed_at=_DT,
78 parent_commit_id=parent,
79 ))
80 ref = root / ".muse" / "refs" / "heads" / branch
81 ref.parent.mkdir(parents=True, exist_ok=True)
82 ref.write_text(cid, encoding="utf-8")
83 return cid
84
85
86 def _mt(root: pathlib.Path, *args: str):
87 from muse.cli.app import main as cli
88 return runner.invoke(cli, ["merge-tree", *args], env=_env(root))
89
90
91 def _diverged(root: pathlib.Path):
92 """base → branch-a (adds a.py) and base → branch-b (adds b.py). No conflict."""
93 base_oid = _write_obj(root, b"base")
94 base_cid = _make_commit(root, {"base.py": base_oid}, "main")
95 a_oid = _write_obj(root, b"a content")
96 a_cid = _make_commit(root, {"base.py": base_oid, "a.py": a_oid}, "branch-a", parent=base_cid)
97 b_oid = _write_obj(root, b"b content")
98 b_cid = _make_commit(root, {"base.py": base_oid, "b.py": b_oid}, "branch-b", parent=base_cid)
99 return base_cid, a_cid, b_cid
100
101
102 def _conflicted(root: pathlib.Path):
103 """base → branch-a and branch-b both modify shared.py differently."""
104 v1 = _write_obj(root, b"v1")
105 base_cid = _make_commit(root, {"shared.py": v1}, "main")
106 va = _write_obj(root, b"version-a")
107 a_cid = _make_commit(root, {"shared.py": va}, "branch-a", parent=base_cid)
108 vb = _write_obj(root, b"version-b")
109 b_cid = _make_commit(root, {"shared.py": vb}, "branch-b", parent=base_cid)
110 return base_cid, a_cid, b_cid
111
112
113 # ---------------------------------------------------------------------------
114 # JSON envelope — exit_code
115 # ---------------------------------------------------------------------------
116
117
118 class TestJsonEnvelopeExitCode:
119 def test_clean_merge_has_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
120 root = _init_repo(tmp_path)
121 _diverged(root)
122 r = _mt(root, "branch-a", "branch-b", "--json")
123 assert r.exit_code == 0
124 d = json.loads(r.output)
125 assert "exit_code" in d, "exit_code missing from clean merge envelope"
126 assert d["exit_code"] == 0
127
128 def test_conflict_merge_has_exit_code_nonzero(self, tmp_path: pathlib.Path) -> None:
129 root = _init_repo(tmp_path)
130 _conflicted(root)
131 r = _mt(root, "branch-a", "branch-b", "--json")
132 assert r.exit_code != 0
133 d = json.loads(r.output)
134 assert "exit_code" in d, "exit_code missing from conflict envelope"
135 assert d["exit_code"] != 0
136
137
138 # ---------------------------------------------------------------------------
139 # JSON envelope — duration_ms
140 # ---------------------------------------------------------------------------
141
142
143 class TestJsonEnvelopeDurationMs:
144 def test_clean_merge_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
145 root = _init_repo(tmp_path)
146 _diverged(root)
147 r = _mt(root, "branch-a", "branch-b", "--json")
148 d = json.loads(r.output)
149 assert "duration_ms" in d, "duration_ms missing from clean merge envelope"
150 assert isinstance(d["duration_ms"], float)
151 assert d["duration_ms"] >= 0.0
152
153 def test_conflict_merge_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
154 root = _init_repo(tmp_path)
155 _conflicted(root)
156 r = _mt(root, "branch-a", "branch-b", "--json")
157 d = json.loads(r.output)
158 assert "duration_ms" in d, "duration_ms missing from conflict envelope"
159 assert isinstance(d["duration_ms"], float)
160
161 def test_write_objects_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
162 root = _init_repo(tmp_path)
163 _diverged(root)
164 r = _mt(root, "branch-a", "branch-b", "--write-objects", "--json")
165 d = json.loads(r.output)
166 assert "duration_ms" in d
167
168
169 # ---------------------------------------------------------------------------
170 # Error payload — errors route to stdout as JSON in --json mode
171 # ---------------------------------------------------------------------------
172
173
174 class TestErrorPayload:
175 def test_bad_branch1_error_goes_to_stdout(self, tmp_path: pathlib.Path) -> None:
176 root = _init_repo(tmp_path)
177 r = _mt(root, "no-such", "also-no", "--json")
178 assert r.exit_code != 0
179 assert not r.stderr.strip(), f"unexpected stderr: {r.stderr!r}"
180 d = json.loads(r.output)
181 assert d["status"] == "error"
182
183 def test_bad_branch2_error_goes_to_stdout(self, tmp_path: pathlib.Path) -> None:
184 root = _init_repo(tmp_path)
185 base_oid = _write_obj(root, b"x")
186 _make_commit(root, {"x.py": base_oid}, "main")
187 r = _mt(root, "main", "nonexistent", "--json")
188 assert r.exit_code != 0
189 assert not r.stderr.strip(), f"unexpected stderr: {r.stderr!r}"
190 d = json.loads(r.output)
191 assert d["status"] == "error"
192
193 def test_no_common_ancestor_error_goes_to_stdout(self, tmp_path: pathlib.Path) -> None:
194 root = _init_repo(tmp_path)
195 oid = _write_obj(root, b"x")
196 _make_commit(root, {"x.py": oid}, "orphan-a")
197 oid2 = _write_obj(root, b"y")
198 _make_commit(root, {"y.py": oid2}, "orphan-b")
199 r = _mt(root, "orphan-a", "orphan-b", "--json")
200 assert r.exit_code != 0
201 assert not r.stderr.strip(), f"unexpected stderr: {r.stderr!r}"
202 d = json.loads(r.output)
203 assert d["status"] == "error"
204
205 def test_error_payload_has_status_error(self, tmp_path: pathlib.Path) -> None:
206 root = _init_repo(tmp_path)
207 r = _mt(root, "ghost", "phantom", "--json")
208 d = json.loads(r.output)
209 assert d["status"] == "error"
210
211 def test_error_payload_has_exit_code(self, tmp_path: pathlib.Path) -> None:
212 root = _init_repo(tmp_path)
213 r = _mt(root, "ghost", "phantom", "--json")
214 d = json.loads(r.output)
215 assert "exit_code" in d
216 assert d["exit_code"] != 0
217
218 def test_error_payload_has_error_field(self, tmp_path: pathlib.Path) -> None:
219 root = _init_repo(tmp_path)
220 r = _mt(root, "ghost", "phantom", "--json")
221 d = json.loads(r.output)
222 assert "error" in d
223 assert d["error"]
224
225 def test_no_duplicate_stderr_prose(self, tmp_path: pathlib.Path) -> None:
226 """In --json mode, errors must not also print ❌ prose to stderr."""
227 root = _init_repo(tmp_path)
228 r = _mt(root, "ghost", "phantom", "--json")
229 assert "❌" not in r.stderr
230
231
232 # ---------------------------------------------------------------------------
233 # Data integrity
234 # ---------------------------------------------------------------------------
235
236
237 class TestDataIntegrity:
238 def test_merged_manifest_oids_are_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
239 """All non-null object IDs in merged_manifest must carry sha256: prefix."""
240 root = _init_repo(tmp_path)
241 _diverged(root)
242 r = _mt(root, "branch-a", "branch-b", "--json")
243 d = json.loads(r.output)
244 for path, oid in d["merged_manifest"].items():
245 if oid is not None:
246 assert oid.startswith("sha256:"), (
247 f"OID for '{path}' missing sha256: prefix: {oid!r}"
248 )
249
250 def test_base_with_sha256_prefixed_commit_id(self, tmp_path: pathlib.Path) -> None:
251 """--base accepts sha256:-prefixed commit IDs (not just branch names)."""
252 root = _init_repo(tmp_path)
253 base_cid, a_cid, b_cid = _diverged(root)
254 r = _mt(root, "branch-a", "branch-b", "--base", base_cid, "--json")
255 assert r.exit_code == 0
256 d = json.loads(r.output)
257 assert d["base"] == base_cid
258
259 def test_branch_ids_echoed_in_response(self, tmp_path: pathlib.Path) -> None:
260 root = _init_repo(tmp_path)
261 base_cid, a_cid, b_cid = _diverged(root)
262 r = _mt(root, "branch-a", "branch-b", "--json")
263 d = json.loads(r.output)
264 assert d["branch1"] == a_cid
265 assert d["branch2"] == b_cid
266
267 def test_conflict_paths_have_null_oid(self, tmp_path: pathlib.Path) -> None:
268 root = _init_repo(tmp_path)
269 _conflicted(root)
270 r = _mt(root, "branch-a", "branch-b", "--json")
271 d = json.loads(r.output)
272 for path in d["conflicts"]:
273 assert d["merged_manifest"][path] is None
274
275 def test_snapshot_id_is_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
276 """--write-objects snapshot_id must carry sha256: prefix."""
277 root = _init_repo(tmp_path)
278 _diverged(root)
279 r = _mt(root, "branch-a", "branch-b", "--write-objects", "--json")
280 d = json.loads(r.output)
281 assert "snapshot_id" in d
282 assert d["snapshot_id"].startswith("sha256:"), (
283 f"snapshot_id missing sha256: prefix: {d['snapshot_id']!r}"
284 )
285
286
287 # ---------------------------------------------------------------------------
288 # No-prose pollution
289 # ---------------------------------------------------------------------------
290
291
292 class TestNoProsePollution:
293 def test_clean_merge_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None:
294 root = _init_repo(tmp_path)
295 _diverged(root)
296 r = _mt(root, "branch-a", "branch-b", "--json")
297 json.loads(r.output) # must not raise
298
299 def test_conflict_merge_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None:
300 root = _init_repo(tmp_path)
301 _conflicted(root)
302 json.loads(_mt(root, "branch-a", "branch-b", "--json").output)
303
304 def test_no_emoji_in_clean_json(self, tmp_path: pathlib.Path) -> None:
305 root = _init_repo(tmp_path)
306 _diverged(root)
307 r = _mt(root, "branch-a", "branch-b", "--json")
308 assert "✅" not in r.output
309 assert "❌" not in r.output
310
311
312 # ---------------------------------------------------------------------------
313 # TypedDicts
314 # ---------------------------------------------------------------------------
315
316
317 class TestTypedDicts:
318 def test_merge_tree_json_typeddict_exists(self) -> None:
319 from muse.cli.commands.merge_tree import _MergeTreeJson
320 assert _MergeTreeJson is not None
321
322 def test_merge_tree_error_json_typeddict_exists(self) -> None:
323 from muse.cli.commands.merge_tree import _MergeTreeErrorJson
324 assert _MergeTreeErrorJson is not None
325
326 def test_merge_tree_json_has_exit_code_annotation(self) -> None:
327 from muse.cli.commands.merge_tree import _MergeTreeJson
328 hints = get_type_hints(_MergeTreeJson)
329 assert "exit_code" in hints
330
331 def test_merge_tree_json_has_duration_ms_annotation(self) -> None:
332 from muse.cli.commands.merge_tree import _MergeTreeJson
333 hints = get_type_hints(_MergeTreeJson)
334 assert "duration_ms" in hints
335
336 def test_merge_tree_error_json_has_required_fields(self) -> None:
337 from muse.cli.commands.merge_tree import _MergeTreeErrorJson
338 hints = get_type_hints(_MergeTreeErrorJson)
339 for field in ("status", "error", "exit_code"):
340 assert field in hints, f"Missing annotation: {field!r}"
341
342
343 # ---------------------------------------------------------------------------
344 # Docstring coverage
345 # ---------------------------------------------------------------------------
346
347
348 class TestDocstring:
349 def _doc(self) -> str:
350 import muse.cli.commands.merge_tree as mod
351 return mod.__doc__ or ""
352
353 def test_docstring_documents_exit_code(self) -> None:
354 assert "exit_code" in self._doc()
355
356 def test_docstring_documents_duration_ms(self) -> None:
357 assert "duration_ms" in self._doc()
358
359
360 # ---------------------------------------------------------------------------
361 # Stress
362 # ---------------------------------------------------------------------------
363
364
365 class TestStress:
366 def test_100_files_40_pct_conflicts(self, tmp_path: pathlib.Path) -> None:
367 root = _init_repo(tmp_path)
368 n = 100
369 conflict_n = 40
370
371 base_manifest = {f"f{i:03d}.py": _write_obj(root, f"base-{i}".encode()) for i in range(n)}
372 base_cid = _make_commit(root, base_manifest, "main", msg="base")
373
374 a_manifest = dict(base_manifest)
375 for i in range(conflict_n):
376 a_manifest[f"f{i:03d}.py"] = _write_obj(root, f"a-{i}".encode())
377 _make_commit(root, a_manifest, "stress-a", parent=base_cid)
378
379 b_manifest = dict(base_manifest)
380 for i in range(conflict_n):
381 b_manifest[f"f{i:03d}.py"] = _write_obj(root, f"b-{i}".encode())
382 _make_commit(root, b_manifest, "stress-b", parent=base_cid)
383
384 r = _mt(root, "stress-a", "stress-b", "--json")
385 assert r.exit_code != 0
386 d = json.loads(r.output)
387 assert len(d["conflicts"]) == conflict_n
388 assert d["exit_code"] != 0
389 assert "duration_ms" in d
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago