gabriel / muse public
test_cmd_plan_merge.py python
1,530 lines 70.7 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 coord plan-merge``.
2
3 Coverage matrix
4 ---------------
5 Unit
6 ~~~~
7 * :func:`_classify_change` — all six change classifications
8 * :func:`_classify_conflict` — all three-way matrix cells
9 (base/ours/theirs combinations: 8 cases + rename, symbol_edit_overlap)
10 * :func:`_find_renames_and_moves` — rename vs move discrimination, trivial body skipped
11 * :func:`_find_delete_use_conflicts` — new callers detected, call graph errors handled
12 * :func:`_find_dependency_conflicts` — transitive dependency detection, errors handled
13 * :class:`_MergeItem` — to_dict, slots
14
15 Integration (mock-based — no real commits required)
16 ~~~~~~~~~~~
17 * ``plan-merge OURS THEIRS`` — ref not found → exit 1
18 * ``plan-merge OURS THEIRS`` — theirs ref not found → exit 1
19 * ``plan-merge OURS THEIRS --base MISSING`` — base ref not found → exit 1
20 * ``plan-merge OURS THEIRS`` — base auto-computed
21 * ``plan-merge OURS THEIRS --base BASE_REF``
22 * ``plan-merge OURS THEIRS --skip-call-graph``
23 * ``plan-merge OURS THEIRS --format json`` — schema complete
24 * ``plan-merge OURS THEIRS --json`` — shorthand
25 * JSON: ``base_auto_computed``, ``call_graph_available``, ``warnings``,
26 ``duration_ms``, full commit IDs, ``conflicts_by_type`` breakdown
27 * JSON output is compact (no indent — single line)
28 * Text output: conflict summary, warnings visible, elapsed present
29 * ``symbol_edit_overlap`` detected (both changed, content differs)
30 * ``rename_edit`` detected via body_hash matching
31 * ``move_edit`` detected via body_hash + file prefix mismatch
32 * ``delete_use`` detected via forward call graph
33 * ``dependency_conflict`` detected via reverse call graph
34 * ``no_conflict`` for unilateral changes (three-way correctness)
35 * call_graph unavailable → warning, not crash
36 * unexpected call graph exception propagates
37
38 Error shapes
39 ~~~~~~~~~~~~
40 * JSON error has ``{"error": ..., "status": "error"}``
41 * Text error uses ``❌`` prefix on stderr
42 * theirs-ref not-found JSON error shape
43 * base-ref not-found JSON error shape
44
45 Security
46 ~~~~~~~~
47 * ANSI sequences in ref names, addresses, change descriptions stripped
48 * Path traversal in ref args → resolve_commit_ref handles it (no FS access)
49
50 Stress
51 ~~~~~~
52 * 1000 symbols across ours + theirs → Pass 1 in < 2 s
53 * 200 renames → Pass 2 in < 1 s
54 * delete_use with 100 deleted symbols → < 2 s
55 * conflicts_by_type counts correct with mixed conflict types at scale
56
57 E2E
58 ~~~
59 * Same commit for ours and theirs → 0 conflicts
60 * Three distinct conflict types in one plan → all appear in conflicts_by_type
61 """
62
63 from __future__ import annotations
64
65 import argparse
66 import json
67 import os
68 import pathlib
69 import sys
70 import time
71 from unittest.mock import MagicMock, patch
72
73 import pytest
74
75 from muse.core.types import Manifest, fake_id
76 from muse.core.paths import muse_dir
77 from muse.plugins.code.ast_parser import SymbolRecord, SymbolTree
78 type SymbolMap = dict[str, SymbolRecord]
79 type SymbolsByFile = dict[str, SymbolMap]
80 from muse.plugins.code._callgraph import ForwardGraph
81
82
83 # ── Helpers ───────────────────────────────────────────────────────────────────
84
85
86 def _sym(
87 name: str,
88 content_id: str | None = None,
89 body_hash: str | None = None,
90 signature_id: str | None = None,
91 metadata_id: str = "",
92 ) -> SymbolRecord:
93 """Build a minimal SymbolRecord for testing."""
94 h = content_id or f"cid-{name}"
95 b = body_hash or f"bh-{name}"
96 s = signature_id or f"sid-{name}"
97 return {
98 "kind": "function",
99 "name": name,
100 "qualified_name": name,
101 "content_id": h,
102 "body_hash": b,
103 "signature_id": s,
104 "metadata_id": metadata_id,
105 "canonical_key": f"f.py##{name}",
106 }
107
108
109 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
110 dot_muse = muse_dir(tmp_path)
111 dot_muse.mkdir()
112 (dot_muse / "HEAD").write_text("ref: refs/heads/main\n")
113 (dot_muse / "repo.json").write_text(
114 json.dumps({"repo_id": fake_id("repo"), "name": "test-repo"})
115 )
116 return tmp_path
117
118
119 @pytest.fixture()
120 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
121 return _make_repo(tmp_path)
122
123
124 def _mock_commit(cid: str | None = None) -> MagicMock:
125 m = MagicMock()
126 m.commit_id = cid or fake_id("commit")
127 return m
128
129
130 def _run_plan_merge(
131 repo: pathlib.Path,
132 ours_ref: str = "HEAD",
133 theirs_ref: str = "main",
134 base_ref: str | None = None,
135 skip_call_graph: bool = True,
136 fmt: str = "json",
137 mock_ours_commit: MagicMock | None = None,
138 mock_theirs_commit: MagicMock | None = None,
139 mock_base_cid: str | None = None,
140 mock_ours_syms: SymbolMap | None = None,
141 mock_theirs_syms: SymbolMap | None = None,
142 mock_base_syms: SymbolMap | None = None,
143 ) -> tuple[int, str]:
144 """Run plan-merge with mocked commits/symbols. Returns (exit_code, stdout)."""
145 from muse.cli.commands.plan_merge import run as pm_run
146
147 ours_c = mock_ours_commit or _mock_commit(f"aaa{'0' * 61}")
148 theirs_c = mock_theirs_commit or _mock_commit(f"bbb{'0' * 61}")
149 base_cid = mock_base_cid or f"ccc{'0' * 61}"
150
151 ours_syms = mock_ours_syms or {}
152 theirs_syms = mock_theirs_syms or {}
153 base_syms = mock_base_syms or {}
154
155 def _sym_for_snapshot(root: pathlib.Path, manifest: Manifest, **kwargs: str) -> SymbolsByFile:
156 if manifest.get("_branch") == "ours":
157 return {"f.py": ours_syms}
158 if manifest.get("_branch") == "theirs":
159 return {"f.py": theirs_syms}
160 if manifest.get("_branch") == "base":
161 return {"f.py": base_syms}
162 return {}
163
164 ns = argparse.Namespace(
165 ours_ref=ours_ref,
166 theirs_ref=theirs_ref,
167 base_ref=base_ref,
168 skip_call_graph=skip_call_graph,
169 fmt=fmt,
170 json_out=(fmt == "json"),
171 )
172 old = os.getcwd()
173 os.chdir(repo)
174 import io, sys
175 captured = io.StringIO()
176 old_stdout = sys.stdout
177 sys.stdout = captured
178
179 exit_code = 0
180 try:
181 with (
182 patch("muse.cli.commands.plan_merge.require_repo", return_value=repo),
183 patch("muse.cli.commands.plan_merge.read_repo_id", return_value="test-repo"),
184 patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"),
185 patch("muse.cli.commands.plan_merge.resolve_commit_ref",
186 side_effect=lambda root, repo_id, branch, ref: (
187 ours_c if ref in (ours_ref, None) else
188 theirs_c if ref == theirs_ref else
189 _mock_commit(mock_base_cid) if ref == base_ref else None
190 )),
191 patch("muse.cli.commands.plan_merge.find_merge_base",
192 return_value=base_cid),
193 patch("muse.cli.commands.plan_merge.get_commit_snapshot_manifest",
194 side_effect=lambda root, cid: (
195 {"_branch": "ours"} if cid == ours_c.commit_id else
196 {"_branch": "theirs"} if cid == theirs_c.commit_id else
197 {"_branch": "base"}
198 )),
199 patch("muse.cli.commands.plan_merge.symbols_for_snapshot",
200 side_effect=_sym_for_snapshot),
201 ):
202 pm_run(ns)
203 except SystemExit as exc:
204 exit_code = exc.code or 0
205 finally:
206 sys.stdout = old_stdout
207 os.chdir(old)
208
209 return exit_code, captured.getvalue()
210
211
212 # ─────────────────────────────────────────────────────────────────────────────
213 # Unit tests — _classify_change
214 # ─────────────────────────────────────────────────────────────────────────────
215
216
217 class TestClassifyChange:
218 def test_unchanged(self) -> None:
219 from muse.cli.commands.plan_merge import _classify_change
220 s = _sym("fn", content_id="X")
221 assert _classify_change(s, s) == "unchanged"
222
223 def test_metadata_only(self) -> None:
224 from muse.cli.commands.plan_merge import _classify_change
225 base = _sym("fn", content_id="X", body_hash="BH", signature_id="SIG")
226 target = _sym("fn", content_id="Y", body_hash="BH", signature_id="SIG")
227 assert _classify_change(base, target) == "metadata_only"
228
229 def test_signature_only(self) -> None:
230 from muse.cli.commands.plan_merge import _classify_change
231 base = _sym("fn", body_hash="BH", signature_id="SIG1")
232 target = _sym("fn", content_id="X2", body_hash="BH", signature_id="SIG2")
233 assert _classify_change(base, target) == "signature_only"
234
235 def test_impl_only(self) -> None:
236 from muse.cli.commands.plan_merge import _classify_change
237 base = _sym("fn", body_hash="BH1", signature_id="SIG")
238 target = _sym("fn", content_id="X2", body_hash="BH2", signature_id="SIG")
239 assert _classify_change(base, target) == "impl_only"
240
241 def test_rename_modify(self) -> None:
242 from muse.cli.commands.plan_merge import _classify_change
243 base = {**_sym("fn_old"), "name": "fn_old"}
244 target = {**_sym("fn_new", content_id="X2", body_hash="BH2"), "name": "fn_new"}
245 assert _classify_change(base, target) == "rename+modify"
246
247 def test_full_rewrite(self) -> None:
248 from muse.cli.commands.plan_merge import _classify_change
249 base = _sym("fn")
250 target = _sym("fn", content_id="X2", body_hash="BH2", signature_id="SIG2")
251 assert _classify_change(base, target) == "full_rewrite"
252
253
254 # ─────────────────────────────────────────────────────────────────────────────
255 # Unit tests — _classify_conflict (three-way matrix)
256 # ─────────────────────────────────────────────────────────────────────────────
257
258
259 class TestClassifyConflict:
260 def test_both_absent_no_conflict(self) -> None:
261 from muse.cli.commands.plan_merge import _classify_conflict
262 item = _classify_conflict("x.py::fn", None, None, None)
263 assert item.conflict_type == "no_conflict"
264
265 def test_ours_new_no_base_no_conflict(self) -> None:
266 from muse.cli.commands.plan_merge import _classify_conflict
267 item = _classify_conflict("x.py::fn", None, _sym("fn"), None)
268 assert item.conflict_type == "no_conflict"
269 assert item.ours_change == "added"
270
271 def test_theirs_new_no_base_no_conflict(self) -> None:
272 from muse.cli.commands.plan_merge import _classify_conflict
273 item = _classify_conflict("x.py::fn", None, None, _sym("fn"))
274 assert item.conflict_type == "no_conflict"
275 assert item.theirs_change == "added"
276
277 def test_only_ours_changed_three_way(self) -> None:
278 """base=X, ours=Y, theirs=X → only ours changed → no_conflict."""
279 from muse.cli.commands.plan_merge import _classify_conflict
280 base = _sym("fn", content_id="X", body_hash="BH1", signature_id="SIG")
281 ours = _sym("fn", content_id="Y", body_hash="BH2", signature_id="SIG")
282 theirs = _sym("fn", content_id="X", body_hash="BH1", signature_id="SIG")
283 item = _classify_conflict("x.py::fn", base, ours, theirs)
284 assert item.conflict_type == "no_conflict"
285 assert "fast-forward" in item.recommendation
286
287 def test_only_theirs_changed_three_way(self) -> None:
288 """base=X, ours=X, theirs=Y → only theirs changed → no_conflict."""
289 from muse.cli.commands.plan_merge import _classify_conflict
290 base = _sym("fn", content_id="X", body_hash="BH1", signature_id="SIG")
291 ours = _sym("fn", content_id="X", body_hash="BH1", signature_id="SIG")
292 theirs = _sym("fn", content_id="Y", body_hash="BH2", signature_id="SIG")
293 item = _classify_conflict("x.py::fn", base, ours, theirs)
294 assert item.conflict_type == "no_conflict"
295
296 def test_both_changed_three_way_overlap(self) -> None:
297 """base=X, ours=Y, theirs=Z → both changed → symbol_edit_overlap."""
298 from muse.cli.commands.plan_merge import _classify_conflict
299 base = _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG0")
300 ours = _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG0")
301 theirs = _sym("fn", content_id="Z", body_hash="BH2", signature_id="SIG0")
302 item = _classify_conflict("x.py::fn", base, ours, theirs)
303 assert item.conflict_type == "symbol_edit_overlap"
304
305 def test_identical_on_both_no_conflict(self) -> None:
306 from muse.cli.commands.plan_merge import _classify_conflict
307 s = _sym("fn", content_id="SAME")
308 item = _classify_conflict("x.py::fn", _sym("fn"), s, s)
309 assert item.conflict_type == "no_conflict"
310 assert "identical" in item.recommendation
311
312 def test_ours_deleted_theirs_unchanged_no_conflict(self) -> None:
313 """base=X, ours=None, theirs=X (unchanged) → no_conflict."""
314 from muse.cli.commands.plan_merge import _classify_conflict
315 base = _sym("fn", content_id="X", body_hash="BH", signature_id="SIG")
316 theirs = _sym("fn", content_id="X", body_hash="BH", signature_id="SIG")
317 item = _classify_conflict("x.py::fn", base, None, theirs)
318 assert item.conflict_type == "no_conflict"
319
320 def test_ours_deleted_theirs_modified_review(self) -> None:
321 """base=X, ours=None, theirs=Y → potentially delete_use → review."""
322 from muse.cli.commands.plan_merge import _classify_conflict
323 base = _sym("fn", content_id="X", body_hash="BH", signature_id="SIG")
324 theirs = _sym("fn", content_id="Y", body_hash="BH2", signature_id="SIG")
325 item = _classify_conflict("x.py::fn", base, None, theirs)
326 # delete_use is detected in Pass 3; this is a "review" no_conflict.
327 assert item.conflict_type == "no_conflict"
328 assert "review" in item.recommendation
329
330 def test_rename_edit_detected_ours_renamed(self) -> None:
331 """Ours renamed (same body, different name) + theirs modified → rename_edit."""
332 from muse.cli.commands.plan_merge import _classify_conflict
333 base_rec = {**_sym("fn_old", body_hash="BH", signature_id="SIG"), "name": "fn_old"}
334 # Ours: same body hash, but name changed (signature_only = rename)
335 ours_rec = {
336 **_sym("fn_new", content_id="C2", body_hash="BH", signature_id="SIG2"),
337 "name": "fn_new",
338 }
339 theirs_rec = {
340 **_sym("fn_old", content_id="C3", body_hash="BH3", signature_id="SIG"),
341 "name": "fn_old",
342 }
343 item = _classify_conflict("x.py::fn_old", base_rec, ours_rec, theirs_rec)
344 assert item.conflict_type == "rename_edit"
345
346 def test_both_added_same_content_no_conflict(self) -> None:
347 """No base, both added same symbol → no_conflict."""
348 from muse.cli.commands.plan_merge import _classify_conflict
349 s = _sym("fn", content_id="SAME")
350 item = _classify_conflict("x.py::fn", None, s, s)
351 assert item.conflict_type == "no_conflict"
352
353 def test_both_added_different_content_overlap(self) -> None:
354 """No base, both added different content → symbol_edit_overlap."""
355 from muse.cli.commands.plan_merge import _classify_conflict
356 item = _classify_conflict(
357 "x.py::fn", None,
358 _sym("fn", content_id="X"),
359 _sym("fn", content_id="Y"),
360 )
361 assert item.conflict_type == "symbol_edit_overlap"
362
363 def test_merge_item_slots_and_to_dict(self) -> None:
364 from muse.cli.commands.plan_merge import _MergeItem
365 m = _MergeItem("addr", "no_conflict", "a", "b", "rec")
366 d = m.to_dict()
367 assert set(d.keys()) == {"address", "conflict_type", "ours_change", "theirs_change", "recommendation"}
368 assert d["conflict_type"] == "no_conflict"
369
370
371 # ─────────────────────────────────────────────────────────────────────────────
372 # Unit tests — _find_renames_and_moves
373 # ─────────────────────────────────────────────────────────────────────────────
374
375
376 class TestFindRenamesAndMoves:
377 def test_rename_detected_same_file(self) -> None:
378 from muse.cli.commands.plan_merge import _find_renames_and_moves
379 base = {"file.py::fn_old": _sym("fn_old", body_hash="BODYHASH123456")}
380 branch = {"file.py::fn_new": {**_sym("fn_new", body_hash="BODYHASH123456"), "name": "fn_new"}}
381 renames, moves = _find_renames_and_moves(base, branch)
382 assert "file.py::fn_old" in renames
383 assert renames["file.py::fn_old"] == "file.py::fn_new"
384 assert not moves
385
386 def test_move_detected_different_file(self) -> None:
387 from muse.cli.commands.plan_merge import _find_renames_and_moves
388 base = {"old/file.py::fn": _sym("fn", body_hash="BODYHASH123456")}
389 branch = {"new/file.py::fn": _sym("fn", body_hash="BODYHASH123456")}
390 renames, moves = _find_renames_and_moves(base, branch)
391 assert not renames
392 assert "old/file.py::fn" in moves
393
394 def test_same_file_preferred_over_other_file(self) -> None:
395 from muse.cli.commands.plan_merge import _find_renames_and_moves
396 base = {"file.py::fn_old": _sym("fn_old", body_hash="BODYHASH123456")}
397 branch = {
398 "file.py::fn_new": _sym("fn_new", body_hash="BODYHASH123456"),
399 "other.py::fn_copy": _sym("fn_copy", body_hash="BODYHASH123456"),
400 }
401 renames, moves = _find_renames_and_moves(base, branch)
402 # Same file candidate should win → rename, not move.
403 assert "file.py::fn_old" in renames
404 assert "file.py::fn_old" not in moves
405
406 def test_trivial_body_hash_skipped(self) -> None:
407 """Very short body hashes are skipped to avoid false positives."""
408 from muse.cli.commands.plan_merge import _find_renames_and_moves
409 base = {"file.py::fn": _sym("fn", body_hash="X")} # Too short.
410 branch = {"file.py::fn_new": _sym("fn_new", body_hash="X")}
411 renames, moves = _find_renames_and_moves(base, branch)
412 assert not renames
413 assert not moves
414
415 def test_still_present_not_a_rename(self) -> None:
416 from muse.cli.commands.plan_merge import _find_renames_and_moves
417 sym = _sym("fn", body_hash="BODY1234567890")
418 base = {"file.py::fn": sym}
419 branch = {"file.py::fn": sym, "file.py::fn2": _sym("fn2", body_hash="BODY1234567890")}
420 renames, moves = _find_renames_and_moves(base, branch)
421 # Original still present → not a rename.
422 assert "file.py::fn" not in renames
423
424
425 # ─────────────────────────────────────────────────────────────────────────────
426 # Unit tests — _find_delete_use_conflicts
427 # ─────────────────────────────────────────────────────────────────────────────
428
429
430 class TestFindDeleteUseConflicts:
431 def _make_manifests(self) -> tuple[Manifest, Manifest, Manifest]:
432 return {"base": "m"}, {"ours": "m"}, {"theirs": "m"}
433
434 def test_delete_use_detected(self, repo: pathlib.Path) -> None:
435 from muse.cli.commands.plan_merge import _find_delete_use_conflicts
436 base_syms = {"src/api.py::fn": _sym("fn")}
437 ours_syms = {} # deleted on ours
438 theirs_syms = {"src/api.py::fn": _sym("fn")} # still in theirs
439
440 # Theirs has a new caller of fn that wasn't in base.
441 base_fg = {"caller_base.py::existing": frozenset()}
442 ours_fg = {}
443 theirs_fg = {
444 "caller_new.py::new_caller": frozenset({"fn"}), # new!
445 }
446 with (
447 patch("muse.plugins.code._callgraph.build_forward_graph",
448 side_effect=[base_fg, ours_fg, theirs_fg]),
449 ):
450 items, ok, warn = _find_delete_use_conflicts(
451 repo, {}, {}, {}, base_syms, ours_syms, theirs_syms,
452 )
453
454 assert ok is True
455 assert warn is None
456 assert len(items) == 1
457 assert items[0].conflict_type == "delete_use"
458 assert "caller_new.py::new_caller" in items[0].theirs_change
459
460 def test_no_delete_use_when_no_deletions(self, repo: pathlib.Path) -> None:
461 from muse.cli.commands.plan_merge import _find_delete_use_conflicts
462 base_syms = {"src/api.py::fn": _sym("fn")}
463 ours_syms = {"src/api.py::fn": _sym("fn")}
464 theirs_syms = {"src/api.py::fn": _sym("fn")}
465 items, ok, warn = _find_delete_use_conflicts(
466 repo, {}, {}, {}, base_syms, ours_syms, theirs_syms,
467 )
468 assert ok is True
469 assert items == []
470
471 def test_call_graph_unavailable_warns(self, repo: pathlib.Path) -> None:
472 from muse.cli.commands.plan_merge import _find_delete_use_conflicts
473 base_syms = {"src/api.py::fn": _sym("fn")}
474 ours_syms = {}
475 theirs_syms = {"src/api.py::fn": _sym("fn")}
476
477 with patch("muse.plugins.code._callgraph.build_forward_graph",
478 side_effect=OSError("no index")):
479 items, ok, warn = _find_delete_use_conflicts(
480 repo, {}, {}, {}, base_syms, ours_syms, theirs_syms,
481 )
482
483 assert ok is False
484 assert warn is not None
485 assert "call graph unavailable" in warn
486 assert items == []
487
488 def test_keyerror_from_call_graph_warns(self, repo: pathlib.Path) -> None:
489 from muse.cli.commands.plan_merge import _find_delete_use_conflicts
490 base_syms = {"src/api.py::fn": _sym("fn")}
491 ours_syms = {}
492 theirs_syms = {"src/api.py::fn": _sym("fn")}
493
494 with patch("muse.plugins.code._callgraph.build_forward_graph",
495 side_effect=KeyError("missing")):
496 items, ok, warn = _find_delete_use_conflicts(
497 repo, {}, {}, {}, base_syms, ours_syms, theirs_syms,
498 )
499
500 assert ok is False
501
502 def test_unexpected_exception_propagates(self, repo: pathlib.Path) -> None:
503 from muse.cli.commands.plan_merge import _find_delete_use_conflicts
504 base_syms = {"src/api.py::fn": _sym("fn")}
505 ours_syms = {}
506 theirs_syms = {"src/api.py::fn": _sym("fn")}
507
508 with patch("muse.plugins.code._callgraph.build_forward_graph",
509 side_effect=MemoryError("OOM")):
510 with pytest.raises(MemoryError):
511 _find_delete_use_conflicts(
512 repo, {}, {}, {}, base_syms, ours_syms, theirs_syms,
513 )
514
515 def test_caller_preview_truncated_at_3(self, repo: pathlib.Path) -> None:
516 from muse.cli.commands.plan_merge import _find_delete_use_conflicts
517 base_syms = {"src/api.py::fn": _sym("fn")}
518 ours_syms = {}
519 theirs_syms = {"src/api.py::fn": _sym("fn")}
520
521 theirs_fg = {f"mod{i}.py::caller{i}": frozenset({"fn"}) for i in range(10)}
522 base_fg = {}
523 ours_fg = {}
524
525 with patch("muse.plugins.code._callgraph.build_forward_graph",
526 side_effect=[base_fg, ours_fg, theirs_fg]):
527 items, ok, _ = _find_delete_use_conflicts(
528 repo, {}, {}, {}, base_syms, ours_syms, theirs_syms,
529 )
530
531 assert len(items) == 1
532 assert "+7 more" in items[0].theirs_change
533
534
535 # ─────────────────────────────────────────────────────────────────────────────
536 # Unit tests — _find_dependency_conflicts
537 # ─────────────────────────────────────────────────────────────────────────────
538
539
540 class TestFindDependencyConflicts:
541 def test_dependency_detected(self, repo: pathlib.Path) -> None:
542 from muse.cli.commands.plan_merge import _find_dependency_conflicts
543 # Ours changed fn_a; theirs changed fn_b which calls fn_a.
544 ours_changed = {"src/x.py::fn_a"}
545 theirs_changed = {"src/y.py::fn_b"}
546 reverse = {"fn_a": ["src/y.py::fn_b"]} # fn_b calls fn_a.
547
548 with (
549 patch("muse.plugins.code._callgraph.build_reverse_graph", return_value=reverse),
550 patch("muse.plugins.code._callgraph.transitive_callers",
551 return_value={1: ["src/y.py::fn_b"]}),
552 ):
553 items, ok, warn = _find_dependency_conflicts(
554 repo, {}, {}, ours_changed, theirs_changed,
555 )
556
557 assert ok is True
558 assert len(items) == 1
559 assert items[0].conflict_type == "dependency_conflict"
560 assert "src/x.py::fn_a" == items[0].address
561
562 def test_no_conflict_when_no_changes(self, repo: pathlib.Path) -> None:
563 from muse.cli.commands.plan_merge import _find_dependency_conflicts
564 items, ok, warn = _find_dependency_conflicts(repo, {}, {}, set(), set())
565 assert items == []
566 assert ok is True
567
568 def test_call_graph_error_warns(self, repo: pathlib.Path) -> None:
569 from muse.cli.commands.plan_merge import _find_dependency_conflicts
570 with patch("muse.plugins.code._callgraph.build_reverse_graph",
571 side_effect=ValueError("bad data")):
572 items, ok, warn = _find_dependency_conflicts(
573 repo, {}, {}, {"x.py::fn"}, {"y.py::fn"},
574 )
575 assert ok is False
576 assert warn is not None
577
578 def test_deduplication(self, repo: pathlib.Path) -> None:
579 """Same (ours_addr, theirs_addr) pair not added twice."""
580 from muse.cli.commands.plan_merge import _find_dependency_conflicts
581 ours_changed = {"x.py::fn_a"}
582 theirs_changed = {"y.py::fn_b"}
583 reverse = {"fn_a": ["y.py::fn_b", "y.py::fn_b"]} # Duplicate.
584
585 with (
586 patch("muse.plugins.code._callgraph.build_reverse_graph", return_value=reverse),
587 patch("muse.plugins.code._callgraph.transitive_callers",
588 return_value={1: ["y.py::fn_b", "y.py::fn_b"]}),
589 ):
590 items, _, _ = _find_dependency_conflicts(repo, {}, {}, ours_changed, theirs_changed)
591
592 assert len(items) == 1 # Not duplicated.
593
594
595 # ─────────────────────────────────────────────────────────────────────────────
596 # Integration tests — full CLI (mock-based)
597 # ─────────────────────────────────────────────────────────────────────────────
598
599
600 class TestPlanMergeIntegration:
601 def test_ref_not_found_exits_1(self, repo: pathlib.Path) -> None:
602 from muse.cli.commands.plan_merge import run as pm_run
603 ns = argparse.Namespace(
604 ours_ref="nonexistent", theirs_ref="main",
605 base_ref=None, skip_call_graph=True, fmt="json",
606 json_out=True,
607 )
608 old = os.getcwd()
609 os.chdir(repo)
610 try:
611 with (
612 patch("muse.cli.commands.plan_merge.require_repo", return_value=repo),
613 patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"),
614 patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"),
615 patch("muse.cli.commands.plan_merge.resolve_commit_ref", return_value=None),
616 ):
617 with pytest.raises(SystemExit) as exc:
618 pm_run(ns)
619 finally:
620 os.chdir(old)
621 assert exc.value.code == 1
622
623 def test_json_schema_complete(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
624 code, out = _run_plan_merge(repo)
625 data = json.loads(out)
626 required = {
627 "schema", "ours", "theirs", "base", "base_auto_computed",
628 "call_graph_available", "call_graph_skipped", "warnings",
629 "total_symbols", "conflicts", "clean", "items", "duration_ms",
630 }
631 assert required.issubset(data.keys())
632
633 def test_duration_ms_is_non_negative_float(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
634 _, out = _run_plan_merge(repo)
635 data = json.loads(out)
636 assert isinstance(data["duration_ms"], float)
637 assert data["duration_ms"] >= 0
638
639 def test_full_commit_ids_in_json(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
640 ours_c = _mock_commit("a" * 64)
641 theirs_c = _mock_commit("b" * 64)
642 _, out = _run_plan_merge(
643 repo,
644 mock_ours_commit=ours_c,
645 mock_theirs_commit=theirs_c,
646 )
647 data = json.loads(out)
648 assert data["ours"] == "a" * 64
649 assert data["theirs"] == "b" * 64
650
651 def test_base_auto_computed_true_when_no_base_arg(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
652 _, out = _run_plan_merge(repo)
653 data = json.loads(out)
654 assert data["base_auto_computed"] is True
655
656 def test_base_auto_computed_false_when_base_arg_given(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
657 from muse.cli.commands.plan_merge import run as pm_run
658 ours_c = _mock_commit("a" * 64)
659 theirs_c = _mock_commit("b" * 64)
660 base_c = _mock_commit("c" * 64)
661
662 ns = argparse.Namespace(
663 ours_ref="HEAD", theirs_ref="main",
664 base_ref="base-ref", skip_call_graph=True, fmt="json",
665 json_out=True,
666 )
667 import io, sys
668 captured = io.StringIO()
669 old_stdout = sys.stdout
670 sys.stdout = captured
671 old = os.getcwd()
672 os.chdir(repo)
673 try:
674 with (
675 patch("muse.cli.commands.plan_merge.require_repo", return_value=repo),
676 patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"),
677 patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"),
678 patch("muse.cli.commands.plan_merge.resolve_commit_ref",
679 side_effect=lambda *a, **kw: (
680 ours_c if a[3] in ("HEAD", None) else
681 theirs_c if a[3] == "main" else
682 base_c if a[3] == "base-ref" else None
683 )),
684 patch("muse.cli.commands.plan_merge.find_merge_base", return_value="c" * 64),
685 patch("muse.cli.commands.plan_merge.get_commit_snapshot_manifest", return_value={}),
686 patch("muse.cli.commands.plan_merge.symbols_for_snapshot", return_value={}),
687 ):
688 pm_run(ns)
689 except SystemExit:
690 pass
691 finally:
692 sys.stdout = old_stdout
693 os.chdir(old)
694
695 data = json.loads(captured.getvalue())
696 assert data["base_auto_computed"] is False
697
698 def test_call_graph_skipped_flag(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
699 _, out = _run_plan_merge(repo, skip_call_graph=True)
700 data = json.loads(out)
701 assert data["call_graph_skipped"] is True
702
703 def test_warnings_list_in_json(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
704 _, out = _run_plan_merge(repo)
705 data = json.loads(out)
706 assert isinstance(data["warnings"], list)
707
708 def test_no_conflicts_empty_swarm(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
709 _, out = _run_plan_merge(repo)
710 data = json.loads(out)
711 assert data["conflicts"] == 0
712 assert data["items"] == []
713
714 def test_symbol_edit_overlap_detected(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
715 """Three-way: base=X, ours=Y, theirs=Z → symbol_edit_overlap."""
716 base = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")}
717 ours = {"f.py::fn": _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")}
718 theirs = {"f.py::fn": _sym("fn", content_id="Z", body_hash="BH2", signature_id="SIG")}
719 _, out = _run_plan_merge(
720 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
721 )
722 data = json.loads(out)
723 assert data["conflicts"] == 1
724 assert data["items"][0]["conflict_type"] == "symbol_edit_overlap"
725
726 def test_no_false_positive_unilateral_change(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
727 """Three-way: base=X, ours=Y, theirs=X → only ours changed → no_conflict."""
728 base = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")}
729 ours = {"f.py::fn": _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")}
730 theirs = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")}
731 _, out = _run_plan_merge(
732 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
733 )
734 data = json.loads(out)
735 assert data["conflicts"] == 0, (
736 "Three-way should detect no conflict when only ours changed"
737 )
738
739 def test_rename_edit_via_pass2_body_hash(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
740 """Pass 2 upgrades symbol_edit_overlap → rename_edit via body_hash matching."""
741 body = "REAL_BODY_HASH_12345"
742 base = {"f.py::fn_old": {**_sym("fn_old", content_id="X", body_hash=body), "name": "fn_old"}}
743 # Ours: fn_old deleted, fn_new added with same body (rename)
744 ours = {"f.py::fn_new": {**_sym("fn_new", content_id="Y", body_hash=body), "name": "fn_new"}}
745 # Theirs: fn_old modified (impl change)
746 theirs = {"f.py::fn_old": {**_sym("fn_old", content_id="Z", body_hash="OTHER"), "name": "fn_old"}}
747 _, out = _run_plan_merge(
748 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
749 )
750 data = json.loads(out)
751 conflict_types = [i["conflict_type"] for i in data["items"]]
752 assert "rename_edit" in conflict_types
753
754 def test_move_edit_via_pass2(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
755 """Pass 2 detects move_edit when ours moved to different file and theirs modified."""
756 body = "MOVED_BODY_HASH_12345"
757 base = {"old/file.py::fn": _sym("fn", content_id="X", body_hash=body)}
758 # Ours: fn moved to new/file.py
759 ours = {"new/file.py::fn": _sym("fn", content_id="X", body_hash=body)}
760 # Theirs: fn modified in original location
761 theirs = {"old/file.py::fn": _sym("fn", content_id="Z", body_hash="OTHER")}
762 _, out = _run_plan_merge(
763 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
764 )
765 data = json.loads(out)
766 conflict_types = [i["conflict_type"] for i in data["items"]]
767 assert "move_edit" in conflict_types
768
769 def test_text_output_format(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
770 _, out = _run_plan_merge(repo, fmt="text")
771 assert "Semantic merge plan" in out
772 assert "base:" in out
773
774 def test_text_output_shows_conflicts(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
775 base = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")}
776 ours = {"f.py::fn": _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")}
777 theirs = {"f.py::fn": _sym("fn", content_id="Z", body_hash="BH2", signature_id="SIG")}
778 _, out = _run_plan_merge(
779 repo, fmt="text",
780 mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
781 )
782 assert "symbol_edit_overlap" in out
783 assert "ours:" in out
784 assert "theirs:" in out
785
786 def test_text_output_shows_elapsed(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
787 _, out = _run_plan_merge(repo, fmt="text")
788 assert "s)" in out
789
790 def test_skip_call_graph_omits_delete_use(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
791 base = {"f.py::fn": _sym("fn")}
792 ours = {} # deleted
793 theirs = {"f.py::fn": _sym("fn")}
794 _, out = _run_plan_merge(
795 repo, skip_call_graph=True,
796 mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
797 )
798 data = json.loads(out)
799 conflict_types = [i["conflict_type"] for i in data["items"]]
800 assert "delete_use" not in conflict_types
801
802 def test_base_none_warning_in_json(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
803 """When find_merge_base returns None, warnings includes a notice."""
804 from muse.cli.commands.plan_merge import run as pm_run
805 ours_c = _mock_commit("a" * 64)
806 theirs_c = _mock_commit("b" * 64)
807 ns = argparse.Namespace(
808 ours_ref="HEAD", theirs_ref="main",
809 base_ref=None, skip_call_graph=True, fmt="json",
810 json_out=True,
811 )
812 import io, sys
813 captured = io.StringIO()
814 old = os.getcwd()
815 os.chdir(repo)
816 sys.stdout = captured
817 try:
818 with (
819 patch("muse.cli.commands.plan_merge.require_repo", return_value=repo),
820 patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"),
821 patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"),
822 patch("muse.cli.commands.plan_merge.resolve_commit_ref",
823 side_effect=lambda *a: ours_c if a[3] in ("HEAD", None) else theirs_c),
824 patch("muse.cli.commands.plan_merge.find_merge_base", return_value=None),
825 patch("muse.cli.commands.plan_merge.get_commit_snapshot_manifest", return_value={}),
826 patch("muse.cli.commands.plan_merge.symbols_for_snapshot", return_value={}),
827 ):
828 pm_run(ns)
829 except SystemExit:
830 pass
831 finally:
832 sys.stdout = sys.__stdout__
833 os.chdir(old)
834
835 data = json.loads(captured.getvalue())
836 assert any("no common ancestor" in w for w in data["warnings"])
837 assert data["base"] is None
838
839 def test_format_json_shorthand(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
840 """--json is equivalent to --format json."""
841 from muse.cli.commands.plan_merge import run as pm_run
842 ours_c = _mock_commit("a" * 64)
843 theirs_c = _mock_commit("b" * 64)
844 ns = argparse.Namespace(
845 ours_ref="HEAD", theirs_ref="main",
846 base_ref=None, skip_call_graph=True, fmt="json", # same as --json
847 json_out=True,
848 )
849 import io, sys
850 captured = io.StringIO()
851 old = os.getcwd()
852 os.chdir(repo)
853 sys.stdout = captured
854 try:
855 with (
856 patch("muse.cli.commands.plan_merge.require_repo", return_value=repo),
857 patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"),
858 patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"),
859 patch("muse.cli.commands.plan_merge.resolve_commit_ref",
860 side_effect=lambda *a: ours_c if a[3] in ("HEAD", None) else theirs_c),
861 patch("muse.cli.commands.plan_merge.find_merge_base", return_value="c" * 64),
862 patch("muse.cli.commands.plan_merge.get_commit_snapshot_manifest", return_value={}),
863 patch("muse.cli.commands.plan_merge.symbols_for_snapshot", return_value={}),
864 ):
865 pm_run(ns)
866 except SystemExit:
867 pass
868 finally:
869 sys.stdout = sys.__stdout__
870 os.chdir(old)
871
872 data = json.loads(captured.getvalue())
873 assert "conflicts" in data
874
875
876 # ─────────────────────────────────────────────────────────────────────────────
877 # Security tests
878 # ─────────────────────────────────────────────────────────────────────────────
879
880
881 class TestPlanMergeSecurity:
882 def test_ansi_in_address_stripped_text_output(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
883 """ANSI escape codes in symbol addresses are stripped before display."""
884 ansi_addr = "\x1b[31msrc/malicious.py::fn\x1b[0m"
885 base = {ansi_addr: _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")}
886 ours = {ansi_addr: _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")}
887 theirs = {ansi_addr: _sym("fn", content_id="Z", body_hash="BH2", signature_id="SIG")}
888 _, out = _run_plan_merge(
889 repo, fmt="text",
890 mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
891 )
892 assert "\x1b[" not in out
893 assert "src/malicious.py::fn" in out # sanitized content still shown
894
895 def test_ansi_in_recommendation_stripped(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
896 base = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")}
897 ours = {"f.py::fn": _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")}
898 theirs = {"f.py::fn": _sym("fn", content_id="Z", body_hash="BH2", signature_id="SIG")}
899 _, out = _run_plan_merge(
900 repo, fmt="text",
901 mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
902 )
903 assert "\x1b[" not in out
904
905 def test_control_chars_in_ref_not_escape_fs(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
906 """Control characters in ref names are sanitised before display in error output."""
907 from muse.cli.commands.plan_merge import run as pm_run
908 import io, sys
909 captured = io.StringIO()
910 malicious_ref = "ref\x00/../../../etc"
911 ns = argparse.Namespace(
912 ours_ref=malicious_ref, theirs_ref="main",
913 base_ref=None, skip_call_graph=True, fmt="json",
914 json_out=True,
915 )
916 old = os.getcwd()
917 os.chdir(repo)
918 sys.stdout = captured
919 try:
920 with (
921 patch("muse.cli.commands.plan_merge.require_repo", return_value=repo),
922 patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"),
923 patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"),
924 patch("muse.cli.commands.plan_merge.resolve_commit_ref", return_value=None),
925 ):
926 with pytest.raises(SystemExit) as exc:
927 pm_run(ns)
928 finally:
929 sys.stdout = sys.__stdout__
930 os.chdir(old)
931 assert exc.value.code == 1
932 out = captured.getvalue()
933 # Output must not contain raw null bytes.
934 assert "\x00" not in out
935
936
937 # ─────────────────────────────────────────────────────────────────────────────
938 # Stress tests
939 # ─────────────────────────────────────────────────────────────────────────────
940
941
942 class TestPlanMergeStress:
943 def test_1000_symbols_pass1_under_2s(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
944 """Pass 1 with 1000 symbols completes in < 2 s."""
945 N = 1000
946 base = {f"f{i}.py::fn{i}": _sym(f"fn{i}", content_id=f"X{i}") for i in range(N)}
947 # Ours and theirs both change half the symbols differently.
948 ours = {
949 f"f{i}.py::fn{i}": (
950 _sym(f"fn{i}", content_id=f"Y{i}") if i % 2 == 0
951 else _sym(f"fn{i}", content_id=f"X{i}")
952 )
953 for i in range(N)
954 }
955 theirs = {
956 f"f{i}.py::fn{i}": (
957 _sym(f"fn{i}", content_id=f"X{i}") if i % 2 == 0
958 else _sym(f"fn{i}", content_id=f"Z{i}")
959 )
960 for i in range(N)
961 }
962
963 t0 = time.monotonic()
964 _, out = _run_plan_merge(
965 repo,
966 mock_ours_syms=ours,
967 mock_theirs_syms=theirs,
968 mock_base_syms=base,
969 )
970 elapsed = time.monotonic() - t0
971
972 assert elapsed < 2.0, f"Pass 1 took {elapsed:.2f}s — too slow"
973 data = json.loads(out)
974 assert data["total_symbols"] == N
975
976 def test_200_renames_pass2_under_1s(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
977 """Pass 2 rename detection with 200 renames completes in < 1 s."""
978 N = 200
979 body_hashes = [f"BODYHASH{i:08d}" for i in range(N)]
980 base = {
981 f"f.py::fn_old_{i}": {**_sym(f"fn_old_{i}", body_hash=body_hashes[i]), "name": f"fn_old_{i}"}
982 for i in range(N)
983 }
984 # Ours: all renamed
985 ours = {
986 f"f.py::fn_new_{i}": {**_sym(f"fn_new_{i}", body_hash=body_hashes[i]), "name": f"fn_new_{i}"}
987 for i in range(N)
988 }
989 # Theirs: all modified (different body)
990 theirs = {
991 f"f.py::fn_old_{i}": {**_sym(f"fn_old_{i}", content_id=f"Z{i}", body_hash=f"DIFF{i}"), "name": f"fn_old_{i}"}
992 for i in range(N)
993 }
994
995 t0 = time.monotonic()
996 _, out = _run_plan_merge(
997 repo,
998 mock_ours_syms=ours,
999 mock_theirs_syms=theirs,
1000 mock_base_syms=base,
1001 )
1002 elapsed = time.monotonic() - t0
1003
1004 assert elapsed < 1.0, f"Pass 2 took {elapsed:.2f}s — too slow"
1005 data = json.loads(out)
1006 conflict_types = [i["conflict_type"] for i in data["items"]]
1007 assert "rename_edit" in conflict_types
1008
1009 def test_100_deletes_delete_use_detection_under_2s(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1010 """delete_use detection with 100 deleted symbols completes in < 2 s."""
1011 from muse.cli.commands.plan_merge import _find_delete_use_conflicts
1012
1013 N = 100
1014 base_syms = {f"f{i}.py::fn{i}": _sym(f"fn{i}") for i in range(N)}
1015 ours_syms: SymbolTree = {} # All deleted on ours.
1016 theirs_syms = dict(base_syms) # All present on theirs.
1017
1018 # Mock: each fn has a new caller on theirs.
1019 base_fg: ForwardGraph = {}
1020 ours_fg: ForwardGraph = {}
1021 theirs_fg = {f"new_caller{i}.py::caller{i}": frozenset({f"fn{i}"}) for i in range(N)}
1022
1023 t0 = time.monotonic()
1024 with patch("muse.plugins.code._callgraph.build_forward_graph",
1025 side_effect=[base_fg, ours_fg, theirs_fg]):
1026 items, ok, warn = _find_delete_use_conflicts(
1027 repo, {}, {}, {}, base_syms, ours_syms, theirs_syms,
1028 )
1029 elapsed = time.monotonic() - t0
1030
1031 assert elapsed < 2.0, f"delete_use took {elapsed:.2f}s — too slow"
1032 assert ok is True
1033 assert len(items) == N
1034
1035 def test_correctness_three_way_no_false_positives(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1036 """Regression: three-way diff must not produce false positives on unilateral changes."""
1037 N = 500
1038 # Ours changes even-indexed symbols, theirs changes odd-indexed.
1039 # Each symbol is changed by only ONE side → all should be no_conflict.
1040 base = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"BASE{i}") for i in range(N)}
1041 ours = {
1042 f"f.py::fn{i}": (
1043 _sym(f"fn{i}", content_id=f"OUR{i}") if i % 2 == 0
1044 else _sym(f"fn{i}", content_id=f"BASE{i}")
1045 )
1046 for i in range(N)
1047 }
1048 theirs = {
1049 f"f.py::fn{i}": (
1050 _sym(f"fn{i}", content_id=f"BASE{i}") if i % 2 == 0
1051 else _sym(f"fn{i}", content_id=f"THEIR{i}")
1052 )
1053 for i in range(N)
1054 }
1055
1056 _, out = _run_plan_merge(
1057 repo,
1058 mock_ours_syms=ours,
1059 mock_theirs_syms=theirs,
1060 mock_base_syms=base,
1061 )
1062 data = json.loads(out)
1063 assert data["conflicts"] == 0, (
1064 f"Three-way correctness: expected 0 conflicts, got {data['conflicts']}"
1065 )
1066
1067
1068 # ─────────────────────────────────────────────────────────────────────────────
1069 # Error shape tests
1070 # ─────────────────────────────────────────────────────────────────────────────
1071
1072
1073 class TestPlanMergeErrorShapes:
1074 """Verify error output shapes are consistent across text and JSON modes."""
1075
1076 def _run_with_no_ours(self, repo: pathlib.Path, fmt: str = "json") -> tuple[int, str, str]:
1077 """Run plan-merge where ours-ref resolves to None."""
1078 from muse.cli.commands.plan_merge import run as pm_run
1079 ns = argparse.Namespace(
1080 ours_ref="missing-ref", theirs_ref="main",
1081 base_ref=None, skip_call_graph=True, fmt=fmt,
1082 json_out=(fmt == "json"),
1083 )
1084 import io
1085 captured = io.StringIO()
1086 old_stdout = sys.stdout
1087 old_stderr = sys.stderr
1088 sys.stdout = captured
1089 err_captured = io.StringIO()
1090 sys.stderr = err_captured
1091 old = os.getcwd()
1092 os.chdir(repo)
1093 exit_code = 0
1094 try:
1095 with (
1096 patch("muse.cli.commands.plan_merge.require_repo", return_value=repo),
1097 patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"),
1098 patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"),
1099 patch("muse.cli.commands.plan_merge.resolve_commit_ref", return_value=None),
1100 ):
1101 with pytest.raises(SystemExit) as exc:
1102 pm_run(ns)
1103 exit_code = exc.value.code
1104 finally:
1105 sys.stdout = old_stdout
1106 sys.stderr = old_stderr
1107 os.chdir(old)
1108 return exit_code, captured.getvalue(), err_captured.getvalue()
1109
1110 def _run_with_no_theirs(self, repo: pathlib.Path, fmt: str = "json") -> tuple[int, str, str]:
1111 from muse.cli.commands.plan_merge import run as pm_run
1112 ours_c = _mock_commit("a" * 64)
1113 ns = argparse.Namespace(
1114 ours_ref="HEAD", theirs_ref="missing-branch",
1115 base_ref=None, skip_call_graph=True, fmt=fmt,
1116 json_out=(fmt == "json"),
1117 )
1118 import io
1119 captured = io.StringIO()
1120 err_captured = io.StringIO()
1121 old_stdout, old_stderr = sys.stdout, sys.stderr
1122 sys.stdout = captured
1123 sys.stderr = err_captured
1124 old = os.getcwd()
1125 os.chdir(repo)
1126 try:
1127 with (
1128 patch("muse.cli.commands.plan_merge.require_repo", return_value=repo),
1129 patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"),
1130 patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"),
1131 patch("muse.cli.commands.plan_merge.resolve_commit_ref",
1132 side_effect=lambda *a: ours_c if a[3] == "HEAD" else None),
1133 ):
1134 with pytest.raises(SystemExit) as exc:
1135 pm_run(ns)
1136 exit_code = exc.value.code
1137 finally:
1138 sys.stdout = old_stdout
1139 sys.stderr = old_stderr
1140 os.chdir(old)
1141 return exit_code, captured.getvalue(), err_captured.getvalue()
1142
1143 def _run_with_no_base(self, repo: pathlib.Path, fmt: str = "json") -> tuple[int, str, str]:
1144 from muse.cli.commands.plan_merge import run as pm_run
1145 ours_c = _mock_commit("a" * 64)
1146 theirs_c = _mock_commit("b" * 64)
1147 ns = argparse.Namespace(
1148 ours_ref="HEAD", theirs_ref="main",
1149 base_ref="missing-base", skip_call_graph=True, fmt=fmt,
1150 json_out=(fmt == "json"),
1151 )
1152 import io
1153 captured = io.StringIO()
1154 err_captured = io.StringIO()
1155 old_stdout, old_stderr = sys.stdout, sys.stderr
1156 sys.stdout = captured
1157 sys.stderr = err_captured
1158 old = os.getcwd()
1159 os.chdir(repo)
1160 try:
1161 with (
1162 patch("muse.cli.commands.plan_merge.require_repo", return_value=repo),
1163 patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"),
1164 patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"),
1165 patch("muse.cli.commands.plan_merge.resolve_commit_ref",
1166 side_effect=lambda *a: (
1167 ours_c if a[3] == "HEAD" else
1168 theirs_c if a[3] == "main" else
1169 None # base-ref not found
1170 )),
1171 ):
1172 with pytest.raises(SystemExit) as exc:
1173 pm_run(ns)
1174 exit_code = exc.value.code
1175 finally:
1176 sys.stdout = old_stdout
1177 sys.stderr = old_stderr
1178 os.chdir(old)
1179 return exit_code, captured.getvalue(), err_captured.getvalue()
1180
1181 # ── ours-ref not found ────────────────────────────────────────────────────
1182
1183 def test_ours_not_found_json_has_error_and_status(self, repo: pathlib.Path) -> None:
1184 code, out, _ = self._run_with_no_ours(repo, fmt="json")
1185 assert code == 1
1186 data = json.loads(out.strip())
1187 assert "error" in data
1188 assert data["status"] == "error"
1189
1190 def test_ours_not_found_json_error_mentions_ref(self, repo: pathlib.Path) -> None:
1191 code, out, _ = self._run_with_no_ours(repo, fmt="json")
1192 data = json.loads(out.strip())
1193 assert "missing-ref" in data["error"]
1194
1195 def test_ours_not_found_text_uses_tick_prefix(self, repo: pathlib.Path) -> None:
1196 code, _, err = self._run_with_no_ours(repo, fmt="text")
1197 assert code == 1
1198 assert "❌" in err
1199
1200 def test_ours_not_found_text_no_output_on_stdout(self, repo: pathlib.Path) -> None:
1201 code, out, _ = self._run_with_no_ours(repo, fmt="text")
1202 assert out == ""
1203
1204 # ── theirs-ref not found ──────────────────────────────────────────────────
1205
1206 def test_theirs_not_found_exits_1(self, repo: pathlib.Path) -> None:
1207 code, _, _ = self._run_with_no_theirs(repo)
1208 assert code == 1
1209
1210 def test_theirs_not_found_json_has_status(self, repo: pathlib.Path) -> None:
1211 code, out, _ = self._run_with_no_theirs(repo, fmt="json")
1212 data = json.loads(out.strip())
1213 assert data["status"] == "error"
1214
1215 def test_theirs_not_found_error_mentions_ref(self, repo: pathlib.Path) -> None:
1216 code, out, _ = self._run_with_no_theirs(repo, fmt="json")
1217 data = json.loads(out.strip())
1218 assert "missing-branch" in data["error"]
1219
1220 def test_theirs_not_found_text_uses_tick_prefix(self, repo: pathlib.Path) -> None:
1221 code, _, err = self._run_with_no_theirs(repo, fmt="text")
1222 assert "❌" in err
1223
1224 # ── base-ref not found ────────────────────────────────────────────────────
1225
1226 def test_base_not_found_exits_1(self, repo: pathlib.Path) -> None:
1227 code, _, _ = self._run_with_no_base(repo)
1228 assert code == 1
1229
1230 def test_base_not_found_json_has_status(self, repo: pathlib.Path) -> None:
1231 code, out, _ = self._run_with_no_base(repo, fmt="json")
1232 data = json.loads(out.strip())
1233 assert data["status"] == "error"
1234
1235 def test_base_not_found_error_mentions_ref(self, repo: pathlib.Path) -> None:
1236 code, out, _ = self._run_with_no_base(repo, fmt="json")
1237 data = json.loads(out.strip())
1238 assert "missing-base" in data["error"]
1239
1240 def test_base_not_found_text_uses_tick_prefix(self, repo: pathlib.Path) -> None:
1241 code, _, err = self._run_with_no_base(repo, fmt="text")
1242 assert "❌" in err
1243
1244
1245 # ─────────────────────────────────────────────────────────────────────────────
1246 # Compact JSON and conflicts_by_type
1247 # ─────────────────────────────────────────────────────────────────────────────
1248
1249
1250 class TestPlanMergeJsonOutput:
1251 """Verify JSON output shape, compactness, and new fields."""
1252
1253 def test_json_is_single_line(self, repo: pathlib.Path) -> None:
1254 """Output must be compact — no embedded newlines from indent=2."""
1255 _, out = _run_plan_merge(repo)
1256 lines = [ln for ln in out.splitlines() if ln.strip()]
1257 assert len(lines) == 1, f"JSON output must be a single line, got {len(lines)} lines"
1258
1259 def test_json_is_valid(self, repo: pathlib.Path) -> None:
1260 _, out = _run_plan_merge(repo)
1261 data = json.loads(out) # raises if invalid
1262 assert isinstance(data, dict)
1263
1264 def test_json_includes_conflicts_by_type_empty(self, repo: pathlib.Path) -> None:
1265 """When no conflicts, conflicts_by_type is an empty dict."""
1266 _, out = _run_plan_merge(repo)
1267 data = json.loads(out)
1268 assert "conflicts_by_type" in data
1269 assert data["conflicts_by_type"] == {}
1270
1271 def test_json_conflicts_by_type_single_type(self, repo: pathlib.Path) -> None:
1272 """Single conflict type → one entry in conflicts_by_type."""
1273 base = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")}
1274 ours = {"f.py::fn": _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")}
1275 theirs = {"f.py::fn": _sym("fn", content_id="Z", body_hash="BH2", signature_id="SIG")}
1276 _, out = _run_plan_merge(
1277 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
1278 )
1279 data = json.loads(out)
1280 assert data["conflicts_by_type"].get("symbol_edit_overlap", 0) == 1
1281
1282 def test_json_conflicts_by_type_count_matches_conflicts_field(self, repo: pathlib.Path) -> None:
1283 """Sum of conflicts_by_type values must equal the 'conflicts' field."""
1284 N = 5
1285 base = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"X{i}") for i in range(N)}
1286 ours = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"Y{i}") for i in range(N)}
1287 theirs = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"Z{i}") for i in range(N)}
1288 _, out = _run_plan_merge(
1289 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
1290 )
1291 data = json.loads(out)
1292 assert sum(data["conflicts_by_type"].values()) == data["conflicts"]
1293
1294 def test_json_schema_includes_all_required_fields(self, repo: pathlib.Path) -> None:
1295 """Backward-compat check: all documented schema fields are present."""
1296 _, out = _run_plan_merge(repo)
1297 data = json.loads(out)
1298 required = {
1299 "schema", "ours", "theirs", "base", "base_auto_computed",
1300 "call_graph_available", "call_graph_skipped", "warnings",
1301 "total_symbols", "conflicts", "clean", "conflicts_by_type",
1302 "items", "duration_ms",
1303 }
1304 missing = required - data.keys()
1305 assert not missing, f"Missing JSON fields: {missing}"
1306
1307 def test_json_items_contains_only_conflicts(self, repo: pathlib.Path) -> None:
1308 """items must contain only conflicting symbols, not clean ones."""
1309 base = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")}
1310 ours = {"f.py::fn": _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")}
1311 theirs = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")}
1312 _, out = _run_plan_merge(
1313 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
1314 )
1315 data = json.loads(out)
1316 assert data["conflicts"] == 0
1317 assert data["items"] == []
1318
1319 def test_json_error_shape_has_status(self, repo: pathlib.Path) -> None:
1320 """Error JSON must include 'status' = 'error' (agent-parseable error shape)."""
1321 from muse.cli.commands.plan_merge import run as pm_run
1322 ns = argparse.Namespace(
1323 ours_ref="bad", theirs_ref="main",
1324 base_ref=None, skip_call_graph=True, fmt="json",
1325 json_out=True,
1326 )
1327 import io
1328 captured = io.StringIO()
1329 old = os.getcwd()
1330 os.chdir(repo)
1331 old_stdout = sys.stdout
1332 sys.stdout = captured
1333 try:
1334 with (
1335 patch("muse.cli.commands.plan_merge.require_repo", return_value=repo),
1336 patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"),
1337 patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"),
1338 patch("muse.cli.commands.plan_merge.resolve_commit_ref", return_value=None),
1339 ):
1340 with pytest.raises(SystemExit):
1341 pm_run(ns)
1342 finally:
1343 sys.stdout = old_stdout
1344 os.chdir(old)
1345 data = json.loads(captured.getvalue().strip())
1346 assert data["status"] == "error"
1347 assert "error" in data
1348
1349
1350 # ─────────────────────────────────────────────────────────────────────────────
1351 # Stress tests — conflicts_by_type correctness at scale
1352 # ─────────────────────────────────────────────────────────────────────────────
1353
1354
1355 class TestPlanMergeConflictsByTypeStress:
1356 def test_conflicts_by_type_counts_correct_at_scale(self, repo: pathlib.Path) -> None:
1357 """100 symbol_edit_overlap conflicts → conflicts_by_type["symbol_edit_overlap"] == 100."""
1358 N = 100
1359 base = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"X{i}") for i in range(N)}
1360 ours = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"Y{i}") for i in range(N)}
1361 theirs = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"Z{i}") for i in range(N)}
1362 _, out = _run_plan_merge(
1363 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
1364 )
1365 data = json.loads(out)
1366 assert data["conflicts_by_type"]["symbol_edit_overlap"] == N
1367 assert sum(data["conflicts_by_type"].values()) == N
1368
1369 def test_json_is_still_compact_with_many_items(self, repo: pathlib.Path) -> None:
1370 """Even with 500 conflict items, JSON output is a single line."""
1371 N = 500
1372 base = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"X{i}") for i in range(N)}
1373 ours = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"Y{i}") for i in range(N)}
1374 theirs = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"Z{i}") for i in range(N)}
1375 _, out = _run_plan_merge(
1376 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
1377 )
1378 lines = [ln for ln in out.splitlines() if ln.strip()]
1379 assert len(lines) == 1
1380
1381 def test_total_symbols_counts_union_of_all_three(self, repo: pathlib.Path) -> None:
1382 """total_symbols is |ours ∪ theirs ∪ base| — not just conflicts."""
1383 # 5 shared (conflicts) + 5 ours-only (clean) + 5 theirs-only (clean)
1384 base = {f"f.py::shared{i}": _sym(f"shared{i}") for i in range(5)}
1385 ours_extra = {f"f.py::ours{i}": _sym(f"ours{i}") for i in range(5)}
1386 theirs_extra = {f"f.py::theirs{i}": _sym(f"theirs{i}") for i in range(5)}
1387 # Make shared symbols conflict
1388 ours_syms = {
1389 **{f"f.py::shared{i}": _sym(f"shared{i}", content_id=f"Y{i}") for i in range(5)},
1390 **ours_extra,
1391 }
1392 theirs_syms = {
1393 **{f"f.py::shared{i}": _sym(f"shared{i}", content_id=f"Z{i}") for i in range(5)},
1394 **theirs_extra,
1395 }
1396 _, out = _run_plan_merge(
1397 repo, mock_ours_syms=ours_syms, mock_theirs_syms=theirs_syms, mock_base_syms=base,
1398 )
1399 data = json.loads(out)
1400 assert data["total_symbols"] == 15
1401 assert data["conflicts"] == 5
1402 assert data["clean"] == 10
1403
1404
1405 # ─────────────────────────────────────────────────────────────────────────────
1406 # E2E tests — same-commit and multi-type plans
1407 # ─────────────────────────────────────────────────────────────────────────────
1408
1409
1410 class TestPlanMergeE2E:
1411 """E2E-style integration tests using mock commits that simulate real scenarios."""
1412
1413 def test_same_symbols_on_both_branches_zero_conflicts(self, repo: pathlib.Path) -> None:
1414 """Identical symbol trees → no conflicts, all clean."""
1415 syms = {f"f.py::fn{i}": _sym(f"fn{i}") for i in range(20)}
1416 _, out = _run_plan_merge(
1417 repo, mock_ours_syms=syms, mock_theirs_syms=syms, mock_base_syms=syms,
1418 )
1419 data = json.loads(out)
1420 assert data["conflicts"] == 0
1421 assert data["total_symbols"] == 20
1422 assert data["conflicts_by_type"] == {}
1423
1424 def test_disjoint_changes_no_conflicts(self, repo: pathlib.Path) -> None:
1425 """Ours and theirs each modify different symbols → all no_conflict."""
1426 base = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"BASE{i}") for i in range(10)}
1427 ours = dict(base)
1428 theirs = dict(base)
1429 # Ours modifies 0–4, theirs modifies 5–9
1430 for i in range(5):
1431 ours[f"f.py::fn{i}"] = _sym(f"fn{i}", content_id=f"OUR{i}")
1432 for i in range(5, 10):
1433 theirs[f"f.py::fn{i}"] = _sym(f"fn{i}", content_id=f"THEIR{i}")
1434 _, out = _run_plan_merge(
1435 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
1436 )
1437 data = json.loads(out)
1438 assert data["conflicts"] == 0
1439
1440 def test_one_new_symbol_on_each_branch_no_conflict(self, repo: pathlib.Path) -> None:
1441 """Each branch adds a different new symbol → no overlap → no conflict."""
1442 base: SymbolTree = {}
1443 ours = {"f.py::fn_ours": _sym("fn_ours")}
1444 theirs = {"f.py::fn_theirs": _sym("fn_theirs")}
1445 _, out = _run_plan_merge(
1446 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
1447 )
1448 data = json.loads(out)
1449 assert data["conflicts"] == 0
1450
1451 def test_both_add_same_symbol_different_content_conflict(self, repo: pathlib.Path) -> None:
1452 """Both branches add the same address with different content → conflict."""
1453 base: SymbolTree = {}
1454 ours = {"f.py::fn_new": _sym("fn_new", content_id="OUR", body_hash="BH1")}
1455 theirs = {"f.py::fn_new": _sym("fn_new", content_id="THEIR", body_hash="BH2")}
1456 _, out = _run_plan_merge(
1457 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
1458 )
1459 data = json.loads(out)
1460 assert data["conflicts"] == 1
1461
1462 def test_both_add_same_symbol_same_content_no_conflict(self, repo: pathlib.Path) -> None:
1463 """Both branches add the same symbol with identical content → no conflict."""
1464 base: SymbolTree = {}
1465 sym = _sym("fn_new", content_id="SAME")
1466 _, out = _run_plan_merge(
1467 repo, mock_ours_syms={"f.py::fn_new": sym},
1468 mock_theirs_syms={"f.py::fn_new": sym},
1469 mock_base_syms=base,
1470 )
1471 data = json.loads(out)
1472 assert data["conflicts"] == 0
1473
1474 def test_both_delete_same_symbol_no_conflict(self, repo: pathlib.Path) -> None:
1475 """Both branches delete the same symbol → no conflict."""
1476 base = {"f.py::fn": _sym("fn")}
1477 ours: SymbolTree = {}
1478 theirs: SymbolTree = {}
1479 _, out = _run_plan_merge(
1480 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
1481 )
1482 data = json.loads(out)
1483 assert data["conflicts"] == 0
1484
1485 def test_rename_edit_and_overlap_in_same_plan(self, repo: pathlib.Path) -> None:
1486 """A plan with both rename_edit and symbol_edit_overlap appears correctly in conflicts_by_type."""
1487 body = "UNIQUEBODY12345"
1488 # Symbol A: ours renames it, theirs modifies original → rename_edit
1489 base_a = {"f.py::fn_a_old": {**_sym("fn_a_old", body_hash=body), "name": "fn_a_old"}}
1490 ours_a = {"f.py::fn_a_new": {**_sym("fn_a_new", body_hash=body), "name": "fn_a_new"}}
1491 theirs_a = {"f.py::fn_a_old": {**_sym("fn_a_old", content_id="Z", body_hash="DIFF"), "name": "fn_a_old"}}
1492 # Symbol B: both change differently → symbol_edit_overlap
1493 base_b = {"f.py::fn_b": _sym("fn_b", content_id="X", body_hash="BH0", signature_id="SIG")}
1494 ours_b = {"f.py::fn_b": _sym("fn_b", content_id="Y", body_hash="BH1", signature_id="SIG")}
1495 theirs_b = {"f.py::fn_b": _sym("fn_b", content_id="Z2", body_hash="BH2", signature_id="SIG")}
1496
1497 base = {**base_a, **base_b}
1498 ours = {**ours_a, **ours_b}
1499 theirs = {**theirs_a, **theirs_b}
1500
1501 _, out = _run_plan_merge(
1502 repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base,
1503 )
1504 data = json.loads(out)
1505 cbt = data["conflicts_by_type"]
1506 assert cbt.get("rename_edit", 0) >= 1
1507 assert cbt.get("symbol_edit_overlap", 0) >= 1
1508 assert sum(cbt.values()) == data["conflicts"]
1509
1510
1511 class TestRegisterFlags:
1512 def _parse(self, *args: str) -> "argparse.Namespace":
1513 import argparse
1514 from muse.cli.commands.plan_merge import register
1515 p = argparse.ArgumentParser()
1516 subs = p.add_subparsers()
1517 register(subs)
1518 return p.parse_args(["plan-merge", "HEAD", "main", *args])
1519
1520 def test_json_short_flag(self) -> None:
1521 args = self._parse("-j")
1522 assert args.json_out is True
1523
1524 def test_json_long_flag(self) -> None:
1525 args = self._parse("--json")
1526 assert args.json_out is True
1527
1528 def test_default_no_json(self) -> None:
1529 args = self._parse()
1530 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