gabriel / muse public
test_core_merge_engine.py python
606 lines 24.7 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 143 days ago
1 """Tests for muse.core.merge_engine — three-way merge logic.
2
3 Extended to cover the structured (operation-level) merge path via
4 :func:`~muse.core.op_transform.merge_structured` and the
5 :class:`~muse.domain.StructuredMergePlugin` integration.
6 """
7
8 import datetime
9 import json
10 import pathlib
11 import unittest.mock
12
13 import pytest
14
15 from muse.core._types import blob_id, long_id
16 from muse.core.merge_engine import (
17 MergeState,
18 apply_merge,
19 clear_merge_state,
20 detect_conflicts,
21 diff_snapshots,
22 find_merge_base,
23 read_merge_state,
24 write_merge_state,
25 )
26 from muse.core.op_transform import MergeOpsResult, merge_op_lists, merge_structured
27 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
28 from muse.core.store import CommitRecord, write_commit
29 from muse.domain import (
30 DeleteOp,
31 DomainOp,
32 InsertOp,
33 ReplaceOp,
34 SnapshotManifest,
35 StructuredDelta,
36 StructuredMergePlugin,
37 )
38 from muse.core.attributes import AttributeRule
39 from muse.plugins.code.plugin import CodePlugin
40 from muse.plugins.midi.plugin import MidiPlugin
41
42 _OID_AAA = blob_id(b"aaa")
43 _OID_OLD = blob_id(b"old")
44 _OID_NEW = blob_id(b"new")
45 _OID_BASE = blob_id(b"base")
46 _OID_OURS = blob_id(b"ours")
47 _OID_THEIRS = blob_id(b"theirs")
48 _OID_K = blob_id(b"k")
49 _OID_FIXED = blob_id(b"fixed")
50
51
52 @pytest.fixture
53 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
54 muse_dir = tmp_path / ".muse"
55 (muse_dir / "commits").mkdir(parents=True)
56 (muse_dir / "refs" / "heads").mkdir(parents=True)
57 return tmp_path
58
59
60 def _commit(root: pathlib.Path, cid: str, parent: str | None = None, parent2: str | None = None) -> str:
61 """Write a commit with a valid content-hash commit_id. Returns the actual commit_id."""
62 snap_id = compute_snapshot_id({})
63 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
64 parent_ids = [p for p in [parent, parent2] if p is not None]
65 commit_id = compute_commit_id(parent_ids, snap_id, cid, committed_at.isoformat())
66 write_commit(root, CommitRecord(
67 commit_id=commit_id,
68 repo_id="r",
69 branch="main",
70 snapshot_id=snap_id,
71 message=cid,
72 committed_at=committed_at,
73 parent_commit_id=parent,
74 parent2_commit_id=parent2,
75 ))
76 return commit_id
77
78
79 class TestDiffSnapshots:
80 def test_no_change(self) -> None:
81 m = {"a.mid": "h1", "b.mid": "h2"}
82 assert diff_snapshots(m, m) == set()
83
84 def test_added(self) -> None:
85 assert diff_snapshots({}, {"a.mid": "h1"}) == {"a.mid"}
86
87 def test_removed(self) -> None:
88 assert diff_snapshots({"a.mid": "h1"}, {}) == {"a.mid"}
89
90 def test_modified(self) -> None:
91 assert diff_snapshots({"a.mid": "old"}, {"a.mid": "new"}) == {"a.mid"}
92
93
94 class TestDetectConflicts:
95 def test_no_conflict_disjoint(self) -> None:
96 ours_m = {"a.mid": "h_a"}
97 theirs_m = {"b.mid": "h_b"}
98 assert detect_conflicts({"a.mid"}, {"b.mid"}, ours_m, theirs_m) == set()
99
100 def test_conflict_divergent_content(self) -> None:
101 ours_m = {"a.mid": "h_a", "b.mid": "h_b_ours"}
102 theirs_m = {"b.mid": "h_b_theirs", "c.mid": "h_c"}
103 assert detect_conflicts({"a.mid", "b.mid"}, {"b.mid", "c.mid"}, ours_m, theirs_m) == {"b.mid"}
104
105 def test_both_empty(self) -> None:
106 assert detect_conflicts(set(), set(), {}, {}) == set()
107
108 def test_convergent_both_delete(self) -> None:
109 """Both branches deleted the same file — convergent, NOT a conflict."""
110 ours_m: Manifest = {} # a.py deleted
111 theirs_m: Manifest = {} # a.py deleted
112 assert detect_conflicts({"a.py"}, {"a.py"}, ours_m, theirs_m) == set()
113
114 def test_convergent_same_add(self) -> None:
115 """Both branches independently added the same file with identical content."""
116 ours_m = {"new.py": "hash_n"}
117 theirs_m = {"new.py": "hash_n"}
118 assert detect_conflicts({"new.py"}, {"new.py"}, ours_m, theirs_m) == set()
119
120 def test_delete_vs_modify_is_conflict(self) -> None:
121 """One side deleted, other modified — genuinely divergent."""
122 ours_m: Manifest = {} # deleted a.py
123 theirs_m = {"a.py": "hash_new"} # modified a.py
124 assert detect_conflicts({"a.py"}, {"a.py"}, ours_m, theirs_m) == {"a.py"}
125
126
127 class TestApplyMerge:
128 def test_clean_merge(self) -> None:
129 base = {"a.mid": "h0", "b.mid": "h0"}
130 ours = {"a.mid": "h_ours", "b.mid": "h0"}
131 theirs = {"a.mid": "h0", "b.mid": "h_theirs"}
132 ours_changed = {"a.mid"}
133 theirs_changed = {"b.mid"}
134 result = apply_merge(base, ours, theirs, ours_changed, theirs_changed, set())
135 assert result == {"a.mid": "h_ours", "b.mid": "h_theirs"}
136
137 def test_conflict_paths_excluded(self) -> None:
138 base = {"a.mid": "h0"}
139 ours = {"a.mid": "h_ours"}
140 theirs = {"a.mid": "h_theirs"}
141 ours_changed = theirs_changed = {"a.mid"}
142 result = apply_merge(base, ours, theirs, ours_changed, theirs_changed, {"a.mid"})
143 assert result == {"a.mid": "h0"} # Falls back to base
144
145 def test_ours_deletion_applied(self) -> None:
146 base = {"a.mid": "h0", "b.mid": "h0"}
147 ours = {"b.mid": "h0"} # a.mid deleted on ours
148 theirs = {"a.mid": "h0", "b.mid": "h0"}
149 result = apply_merge(base, ours, theirs, {"a.mid"}, set(), set())
150 assert "a.mid" not in result
151
152
153 class TestMergeStateIO:
154 def test_write_and_read(self, repo: pathlib.Path) -> None:
155 base_id = long_id("b" * 64)
156 ours_id = long_id("1" * 64)
157 theirs_id = long_id("2" * 64)
158 write_merge_state(
159 repo,
160 base_commit=base_id,
161 ours_commit=ours_id,
162 theirs_commit=theirs_id,
163 conflict_paths=["a.mid", "b.mid"],
164 other_branch="feature/x",
165 )
166 state = read_merge_state(repo)
167 assert state is not None
168 assert state.base_commit == base_id
169 assert state.conflict_paths == ["a.mid", "b.mid"]
170 assert state.other_branch == "feature/x"
171
172 def test_read_no_state(self, repo: pathlib.Path) -> None:
173 assert read_merge_state(repo) is None
174
175 def test_clear(self, repo: pathlib.Path) -> None:
176 write_merge_state(repo, base_commit=long_id("b" * 64), ours_commit=long_id("c" * 64), theirs_commit=long_id("d" * 64), conflict_paths=[])
177 clear_merge_state(repo)
178 assert read_merge_state(repo) is None
179
180
181 class TestFindMergeBase:
182 def test_direct_parent(self, repo: pathlib.Path) -> None:
183 root_id = _commit(repo, "root")
184 a_id = _commit(repo, "a", parent=root_id)
185 b_id = _commit(repo, "b", parent=root_id)
186 base = find_merge_base(repo, a_id, b_id)
187 assert base == root_id
188
189 def test_same_commit(self, repo: pathlib.Path) -> None:
190 _commit(repo, "root")
191 base = find_merge_base(repo, "root", "root")
192 assert base == "root"
193
194 def test_linear_history(self, repo: pathlib.Path) -> None:
195 a_id = _commit(repo, "a")
196 b_id = _commit(repo, "b", parent=a_id)
197 c_id = _commit(repo, "c", parent=b_id)
198 base = find_merge_base(repo, c_id, b_id)
199 assert base == b_id
200
201 def test_no_common_ancestor(self, repo: pathlib.Path) -> None:
202 _commit(repo, "x")
203 _commit(repo, "y")
204 assert find_merge_base(repo, "x", "y") is None
205
206 def test_bidirectional_terminates_early(self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
207 """Bidirectional BFS reads O(distance_to_LCA) commits, not O(total_history)."""
208 import muse.core.store as store_mod
209
210 # 100-commit chain: root → c0 → ... → c97 → head_a
211 # ↘ feat (branches from c97)
212 # LCA = c97, one hop from each tip
213 root = _commit(repo, "root")
214 tip = root
215 for i in range(97):
216 tip = _commit(repo, f"c{i}", parent=tip)
217 lca = tip
218 head_a = _commit(repo, "head_a", parent=lca)
219 feat = _commit(repo, "feat", parent=lca)
220
221 call_count = 0
222 original = store_mod.read_commit
223
224 def counting_read(rr: pathlib.Path, cid: str) -> object:
225 nonlocal call_count
226 call_count += 1
227 return original(rr, cid)
228
229 monkeypatch.setattr(store_mod, "read_commit", counting_read)
230
231 base = find_merge_base(repo, head_a, feat)
232 assert base == lca
233 # Bidirectional BFS finds LCA in ~2 reads (one per tip).
234 # Old two-phase BFS read all 99 A ancestors before touching B.
235 assert call_count <= 10
236
237 def test_deep_chain_diamond(self, repo: pathlib.Path) -> None:
238 """LCA correct for a long chain with diverging feature branches."""
239 root = _commit(repo, "root")
240 tip = root
241 for i in range(50):
242 tip = _commit(repo, f"m{i}", parent=tip)
243 lca = tip
244 branch_a = _commit(repo, "a0", parent=lca)
245 branch_b = _commit(repo, "b0", parent=lca)
246 for i in range(1, 5):
247 branch_a = _commit(repo, f"a{i}", parent=branch_a)
248 branch_b = _commit(repo, f"b{i}", parent=branch_b)
249 base = find_merge_base(repo, branch_a, branch_b)
250 assert base == lca
251
252
253 # ===========================================================================
254 # Structured merge engine integration tests
255 # ===========================================================================
256
257
258 def _ins(addr: str, pos: int | None, cid: str) -> InsertOp:
259 return InsertOp(op="insert", address=addr, position=pos, content_id=cid, content_summary=cid)
260
261
262 def _del(addr: str, pos: int | None, cid: str) -> DeleteOp:
263 return DeleteOp(op="delete", address=addr, position=pos, content_id=cid, content_summary=cid)
264
265
266 def _rep(addr: str, old: str, new: str) -> ReplaceOp:
267 return ReplaceOp(
268 op="replace",
269 address=addr,
270 position=None,
271 old_content_id=old,
272 new_content_id=new,
273 old_summary="old",
274 new_summary="new",
275 )
276
277
278 def _delta(ops: list[DomainOp]) -> StructuredDelta:
279 return StructuredDelta(domain="midi", ops=ops, summary="test")
280
281
282 class TestMergeStructuredIntegration:
283 """Verify merge_structured delegates correctly to merge_op_lists."""
284
285 def test_clean_non_overlapping_file_ops(self) -> None:
286 ours = _delta([_ins("a.mid", pos=0, cid="a-hash")])
287 theirs = _delta([_ins("b.mid", pos=0, cid="b-hash")])
288 result = merge_structured(_delta([]), ours, theirs)
289 assert result.is_clean is True
290 assert len(result.merged_ops) == 2
291
292 def test_conflicting_same_address_replaces_detected(self) -> None:
293 ours = _delta([_rep("shared.mid", "old", "v-ours")])
294 theirs = _delta([_rep("shared.mid", "old", "v-theirs")])
295 result = merge_structured(_delta([]), ours, theirs)
296 assert result.is_clean is False
297 assert len(result.conflict_ops) == 1
298
299 def test_base_ops_kept_by_both_sides_preserved(self) -> None:
300 shared = _ins("base.mid", pos=0, cid="base-cid")
301 result = merge_structured(
302 _delta([shared]),
303 _delta([shared]),
304 _delta([shared]),
305 )
306 assert result.is_clean is True
307 assert any(_op_key_tuple(op) == _op_key_tuple(shared) for op in result.merged_ops)
308
309 def test_position_adjustment_in_structured_merge(self) -> None:
310 """Non-conflicting note inserts get position-adjusted in structured merge."""
311 ours = _delta([_ins("lead.mid", pos=3, cid="note-A")])
312 theirs = _delta([_ins("lead.mid", pos=7, cid="note-B")])
313 result = merge_structured(_delta([]), ours, theirs)
314 assert result.is_clean is True
315 pos_by_cid = {
316 op["content_id"]: op["position"]
317 for op in result.merged_ops
318 if op["op"] == "insert"
319 }
320 # note-A(3): no theirs ≤ 3 → stays 3
321 assert pos_by_cid["note-A"] == 3
322 # note-B(7): ours A(3) ≤ 7 → 7+1 = 8
323 assert pos_by_cid["note-B"] == 8
324
325
326 def _op_key_tuple(op: DomainOp) -> tuple[str, ...]:
327 """Re-implementation of _op_key for test assertions."""
328 if op["op"] == "insert":
329 return ("insert", op["address"], str(op["position"]), op["content_id"])
330 if op["op"] == "delete":
331 return ("delete", op["address"], str(op["position"]), op["content_id"])
332 if op["op"] == "replace":
333 return ("replace", op["address"], str(op["position"]), op["old_content_id"], op["new_content_id"])
334 return (op["op"], op["address"])
335
336
337 class TestStructuredMergePluginProtocol:
338 """Verify MidiPlugin satisfies the StructuredMergePlugin protocol."""
339
340 def test_midi_plugin_isinstance_structured_merge_plugin(self) -> None:
341 plugin = MidiPlugin()
342 assert isinstance(plugin, StructuredMergePlugin)
343
344 def test_merge_ops_non_conflicting_files_is_clean(self) -> None:
345 plugin = MidiPlugin()
346 base = SnapshotManifest(files={}, domain="midi")
347 ours_snap = SnapshotManifest(files={"a.mid": "hash-a"}, domain="midi")
348 theirs_snap = SnapshotManifest(files={"b.mid": "hash-b"}, domain="midi")
349 ours_ops: list[DomainOp] = [_ins("a.mid", pos=None, cid="hash-a")]
350 theirs_ops: list[DomainOp] = [_ins("b.mid", pos=None, cid="hash-b")]
351
352 result = plugin.merge_ops(
353 base, ours_snap, theirs_snap, ours_ops, theirs_ops
354 )
355 assert result.is_clean is True
356 assert "a.mid" in result.merged["files"]
357 assert "b.mid" in result.merged["files"]
358
359 def test_merge_ops_conflicting_same_file_replace_not_clean(self) -> None:
360 plugin = MidiPlugin()
361 base = SnapshotManifest(files={"f.mid": "base-hash"}, domain="midi")
362 ours_snap = SnapshotManifest(files={"f.mid": "ours-hash"}, domain="midi")
363 theirs_snap = SnapshotManifest(files={"f.mid": "theirs-hash"}, domain="midi")
364 ours_ops: list[DomainOp] = [_rep("f.mid", "base-hash", "ours-hash")]
365 theirs_ops: list[DomainOp] = [_rep("f.mid", "base-hash", "theirs-hash")]
366
367 result = plugin.merge_ops(
368 base, ours_snap, theirs_snap, ours_ops, theirs_ops
369 )
370 assert not result.is_clean
371 assert "f.mid" in result.conflicts
372
373 def test_merge_ops_ours_strategy_resolves_conflict(self) -> None:
374 plugin = MidiPlugin()
375 base = SnapshotManifest(files={"f.mid": "base"}, domain="midi")
376 ours_snap = SnapshotManifest(files={"f.mid": "ours-v"}, domain="midi")
377 theirs_snap = SnapshotManifest(files={"f.mid": "theirs-v"}, domain="midi")
378 ours_ops: list[DomainOp] = [_rep("f.mid", "base", "ours-v")]
379 theirs_ops: list[DomainOp] = [_rep("f.mid", "base", "theirs-v")]
380
381 result = plugin.merge_ops(
382 base,
383 ours_snap,
384 theirs_snap,
385 ours_ops,
386 theirs_ops,
387 )
388 # Without .museattributes the conflict stands — verify conflict is reported.
389 assert not result.is_clean
390
391 def test_merge_ops_delete_on_only_one_side_is_clean(self) -> None:
392 plugin = MidiPlugin()
393 base = SnapshotManifest(files={"keep.mid": "k", "remove.mid": "r"}, domain="midi")
394 ours_snap = SnapshotManifest(files={"keep.mid": "k"}, domain="midi")
395 theirs_snap = SnapshotManifest(files={"keep.mid": "k", "remove.mid": "r"}, domain="midi")
396 ours_ops: list[DomainOp] = [_del("remove.mid", pos=None, cid="r")]
397 theirs_ops: list[DomainOp] = []
398
399 result = plugin.merge_ops(
400 base, ours_snap, theirs_snap, ours_ops, theirs_ops
401 )
402 assert result.is_clean is True
403 assert "keep.mid" in result.merged["files"]
404 assert "remove.mid" not in result.merged["files"]
405
406 def test_merge_ops_empty_changes_returns_base(self) -> None:
407 plugin = MidiPlugin()
408 base = SnapshotManifest(files={"f.mid": "h"}, domain="midi")
409 result = plugin.merge_ops(base, base, base, [], [])
410 assert result.is_clean is True
411 assert result.merged["files"] == {"f.mid": "h"}
412
413
414 # ---------------------------------------------------------------------------
415 # Bug: "manual" attribute strategy on l==r paths causes false conflicts
416 #
417 # The `manual` strategy fires "even when the engine would auto-resolve" a
418 # DIVERGENT change (one side changed, the engine would take it automatically).
419 # It must NOT fire when both sides agree (l == r) — whether nothing changed
420 # (b == l == r) or both made the same convergent edit (b != l == r).
421 #
422 # Regression for: muse merge task/core-cat → 73 false conflicts in muse/core/**
423 # caused by [[rules]] path="muse/core/**" strategy="manual" in .museattributes.
424 # ---------------------------------------------------------------------------
425
426 _DUMMY_ROOT = pathlib.Path("/nonexistent-repo-for-testing")
427
428
429 def _code_plugin() -> CodePlugin:
430 return CodePlugin()
431
432
433 def _snap(files: dict[str, str]) -> SnapshotManifest:
434 return SnapshotManifest(files=files, domain="code")
435
436
437 def _manual_attrs() -> list:
438 return [AttributeRule(path_pattern="core/**", dimension="*", strategy="manual", priority=100)]
439
440
441 def _merge_with_manual(
442 base: SnapshotManifest,
443 ours: SnapshotManifest,
444 theirs: SnapshotManifest,
445 ) -> MergeResult:
446 plugin = _code_plugin()
447 with unittest.mock.patch(
448 "muse.plugins.code.plugin.load_attributes",
449 return_value=_manual_attrs(),
450 ):
451 return plugin.merge(base, ours, theirs, repo_root=_DUMMY_ROOT)
452
453
454 class TestManualStrategyUnchangedFiles:
455 """manual fires for single-branch changes but NEVER for l == r paths."""
456
457 def test_unchanged_no_conflict(self) -> None:
458 """b == l == r: neither branch touched the file → no conflict."""
459 base = _snap({"core/store.py": _OID_AAA})
460 result = _merge_with_manual(base, base, base)
461 assert result.is_clean
462 assert "core/store.py" not in result.conflicts
463
464 def test_convergent_same_change_no_conflict(self) -> None:
465 """b != l == r: both independently made the same edit → convergent, no conflict."""
466 base = _snap({"core/store.py": _OID_OLD})
467 same = _snap({"core/store.py": _OID_NEW})
468 result = _merge_with_manual(base, same, same)
469 assert result.is_clean
470 assert "core/store.py" not in result.conflicts
471
472 def test_only_ours_changed_manual_forces_conflict(self) -> None:
473 """b == theirs != ours: one side changed → manual forces human review."""
474 base = _snap({"core/store.py": _OID_OLD})
475 ours = _snap({"core/store.py": _OID_NEW})
476 theirs = _snap({"core/store.py": _OID_OLD})
477 result = _merge_with_manual(base, ours, theirs)
478 assert "core/store.py" in result.conflicts
479
480 def test_only_theirs_changed_manual_forces_conflict(self) -> None:
481 """b == ours != theirs: one side changed → manual forces human review."""
482 base = _snap({"core/store.py": _OID_OLD})
483 ours = _snap({"core/store.py": _OID_OLD})
484 theirs = _snap({"core/store.py": _OID_NEW})
485 result = _merge_with_manual(base, ours, theirs)
486 assert "core/store.py" in result.conflicts
487
488 def test_divergent_changes_conflict(self) -> None:
489 """Both changed differently → conflict regardless."""
490 base = _snap({"core/store.py": _OID_BASE})
491 ours = _snap({"core/store.py": _OID_OURS})
492 theirs = _snap({"core/store.py": _OID_THEIRS})
493 result = _merge_with_manual(base, ours, theirs)
494 assert "core/store.py" in result.conflicts
495
496 def test_73_unchanged_plus_one_real_conflict(self) -> None:
497 """Regression: 73 unchanged core files + 1 real conflict → only 1 conflict."""
498 unchanged = {f"core/file_{i}.py": blob_id(f"file_{i}".encode()) for i in range(73)}
499 base_files = {**unchanged, "core/store.py": _OID_BASE}
500 ours_files = {**unchanged, "core/store.py": _OID_OURS}
501 theirs_files = {**unchanged, "core/store.py": _OID_THEIRS}
502 result = _merge_with_manual(
503 _snap(base_files), _snap(ours_files), _snap(theirs_files)
504 )
505 false_conflicts = [p for p in result.conflicts if p != "core/store.py"]
506 assert false_conflicts == [], f"{len(false_conflicts)} false conflicts: {false_conflicts[:5]}"
507 assert "core/store.py" in result.conflicts
508
509
510 # ---------------------------------------------------------------------------
511 # Bug: one-sided changes must NEVER produce false conflicts
512 #
513 # When only one branch changes a file (b == ours or b == theirs), the merge
514 # must take the changed side cleanly — no conflict, no manual review needed.
515 # This covers both the file-level merge() path and the operation-level
516 # merge_ops() path used by the code plugin (StructuredMergePlugin).
517 # ---------------------------------------------------------------------------
518
519
520 class TestOneSidedChangeNeverConflicts:
521 """One side changes a file, other side doesn't → always clean."""
522
523 def test_only_theirs_changed_no_conflict(self) -> None:
524 base = _snap({"pyproject.toml": _OID_OLD, "describe.py": _OID_OLD})
525 ours = _snap({"pyproject.toml": _OID_OLD, "describe.py": _OID_OLD})
526 theirs = _snap({"pyproject.toml": _OID_NEW, "describe.py": _OID_FIXED})
527 result = _code_plugin().merge(base, ours, theirs, repo_root=None)
528 assert result.is_clean
529 assert result.conflicts == []
530 assert result.merged["files"]["pyproject.toml"] == _OID_NEW
531 assert result.merged["files"]["describe.py"] == _OID_FIXED
532
533 def test_only_ours_changed_no_conflict(self) -> None:
534 base = _snap({"a.py": _OID_OLD})
535 ours = _snap({"a.py": _OID_OURS})
536 theirs = _snap({"a.py": _OID_OLD})
537 result = _code_plugin().merge(base, ours, theirs, repo_root=None)
538 assert result.is_clean
539 assert result.conflicts == []
540 assert result.merged["files"]["a.py"] == _OID_OURS
541
542 def test_theirs_deleted_file_ours_untouched(self) -> None:
543 base = _snap({"gone.py": _OID_OLD, "keep.py": _OID_K})
544 ours = _snap({"gone.py": _OID_OLD, "keep.py": _OID_K})
545 theirs = _snap({"keep.py": _OID_K})
546 result = _code_plugin().merge(base, ours, theirs, repo_root=None)
547 assert result.is_clean
548 assert "gone.py" not in result.merged["files"]
549
550 def test_merge_ops_one_sided_no_conflict(self) -> None:
551 """merge_ops() must not flag one-sided changes as conflicts."""
552 plugin = _code_plugin()
553 base = _snap({"pyproject.toml": _OID_OLD})
554 ours = _snap({"pyproject.toml": _OID_OLD})
555 theirs = _snap({"pyproject.toml": _OID_NEW})
556 ours_delta: StructuredDelta = {"ops": [], "summary": "", "domain": "code"}
557 theirs_delta: StructuredDelta = {
558 "ops": [{"op": "patch", "address": "pyproject.toml", "child_ops": [],
559 "file_change": "modified", "content_summary": ""}],
560 "summary": "1 change",
561 "domain": "code",
562 }
563 result = plugin.merge_ops(base, ours, theirs, ours_delta["ops"], theirs_delta["ops"])
564 assert result.conflicts == [], f"False conflicts: {result.conflicts}"
565
566
567 # ---------------------------------------------------------------------------
568 # Bug: merge commit with two parents must pass write_commit hash verification
569 #
570 # compute_commit_id and _verify_commit_id must be symmetric for merge commits
571 # (two parents). If they disagree, write_commit raises ValueError and the
572 # merge cannot be completed — data is permanently stuck.
573 # ---------------------------------------------------------------------------
574
575
576 class TestMergeCommitHashVerification:
577 """write_commit must accept merge commits (two parents) without raising."""
578
579 def test_two_parent_commit_passes_verification(self, repo: pathlib.Path) -> None:
580 parent1 = _commit(repo, "ours")
581 parent2 = _commit(repo, "theirs")
582 # A merge commit with both parents must write cleanly.
583 _commit(repo, "merge", parent=parent1, parent2=parent2)
584
585 def test_merge_commit_id_is_deterministic(self, repo: pathlib.Path) -> None:
586 """Same inputs → same commit_id regardless of parent order in the list."""
587 snap_id = compute_snapshot_id({})
588 committed_at = datetime.datetime(2026, 3, 1, tzinfo=datetime.timezone.utc)
589 p1 = _commit(repo, "p1")
590 p2 = _commit(repo, "p2")
591 id_ab = compute_commit_id([p1, p2], snap_id, "merge", committed_at.isoformat())
592 id_ba = compute_commit_id([p2, p1], snap_id, "merge", committed_at.isoformat())
593 assert id_ab == id_ba, "Merge commit ID must be order-independent"
594
595 def test_verify_sees_same_id_as_compute(self, repo: pathlib.Path) -> None:
596 """_verify_commit_id (called inside write_commit) must agree with compute_commit_id."""
597 from muse.core.store import read_commit
598 parent1 = _commit(repo, "ours")
599 parent2 = _commit(repo, "theirs")
600 merge_id = _commit(repo, "merge", parent=parent1, parent2=parent2)
601 # If write_commit succeeded, read_commit must return the record intact.
602 rec = read_commit(repo, merge_id)
603 assert rec is not None
604 assert rec.commit_id == merge_id
605 assert rec.parent_commit_id == parent1
606 assert rec.parent2_commit_id == parent2
File History 2 commits
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 143 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 146 days ago