gabriel / muse public
test_core_merge_engine.py python
352 lines 13.4 KB
Raw
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 151 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
12 import pytest
13
14 from muse.core.merge_engine import (
15 MergeState,
16 apply_merge,
17 clear_merge_state,
18 detect_conflicts,
19 diff_snapshots,
20 find_merge_base,
21 read_merge_state,
22 write_merge_state,
23 )
24 from muse.core.op_transform import MergeOpsResult, merge_op_lists, merge_structured
25 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
26 from muse.core.store import CommitRecord, write_commit
27 from muse.domain import (
28 DeleteOp,
29 DomainOp,
30 InsertOp,
31 ReplaceOp,
32 SnapshotManifest,
33 StructuredDelta,
34 StructuredMergePlugin,
35 )
36 from muse.plugins.midi.plugin import MidiPlugin
37
38
39 @pytest.fixture
40 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
41 muse_dir = tmp_path / ".muse"
42 (muse_dir / "commits").mkdir(parents=True)
43 (muse_dir / "refs" / "heads").mkdir(parents=True)
44 return tmp_path
45
46
47 def _commit(root: pathlib.Path, cid: str, parent: str | None = None, parent2: str | None = None) -> str:
48 """Write a commit with a valid content-hash commit_id. Returns the actual commit_id."""
49 snap_id = compute_snapshot_id({})
50 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
51 parent_ids = [p for p in [parent, parent2] if p is not None]
52 commit_id = compute_commit_id(parent_ids, snap_id, cid, committed_at.isoformat())
53 write_commit(root, CommitRecord(
54 commit_id=commit_id,
55 repo_id="r",
56 branch="main",
57 snapshot_id=snap_id,
58 message=cid,
59 committed_at=committed_at,
60 parent_commit_id=parent,
61 parent2_commit_id=parent2,
62 ))
63 return commit_id
64
65
66 class TestDiffSnapshots:
67 def test_no_change(self) -> None:
68 m = {"a.mid": "h1", "b.mid": "h2"}
69 assert diff_snapshots(m, m) == set()
70
71 def test_added(self) -> None:
72 assert diff_snapshots({}, {"a.mid": "h1"}) == {"a.mid"}
73
74 def test_removed(self) -> None:
75 assert diff_snapshots({"a.mid": "h1"}, {}) == {"a.mid"}
76
77 def test_modified(self) -> None:
78 assert diff_snapshots({"a.mid": "old"}, {"a.mid": "new"}) == {"a.mid"}
79
80
81 class TestDetectConflicts:
82 def test_no_conflict_disjoint(self) -> None:
83 ours_m = {"a.mid": "h_a"}
84 theirs_m = {"b.mid": "h_b"}
85 assert detect_conflicts({"a.mid"}, {"b.mid"}, ours_m, theirs_m) == set()
86
87 def test_conflict_divergent_content(self) -> None:
88 ours_m = {"a.mid": "h_a", "b.mid": "h_b_ours"}
89 theirs_m = {"b.mid": "h_b_theirs", "c.mid": "h_c"}
90 assert detect_conflicts({"a.mid", "b.mid"}, {"b.mid", "c.mid"}, ours_m, theirs_m) == {"b.mid"}
91
92 def test_both_empty(self) -> None:
93 assert detect_conflicts(set(), set(), {}, {}) == set()
94
95 def test_convergent_both_delete(self) -> None:
96 """Both branches deleted the same file — convergent, NOT a conflict."""
97 ours_m: Manifest = {} # a.py deleted
98 theirs_m: Manifest = {} # a.py deleted
99 assert detect_conflicts({"a.py"}, {"a.py"}, ours_m, theirs_m) == set()
100
101 def test_convergent_same_add(self) -> None:
102 """Both branches independently added the same file with identical content."""
103 ours_m = {"new.py": "hash_n"}
104 theirs_m = {"new.py": "hash_n"}
105 assert detect_conflicts({"new.py"}, {"new.py"}, ours_m, theirs_m) == set()
106
107 def test_delete_vs_modify_is_conflict(self) -> None:
108 """One side deleted, other modified — genuinely divergent."""
109 ours_m: Manifest = {} # deleted a.py
110 theirs_m = {"a.py": "hash_new"} # modified a.py
111 assert detect_conflicts({"a.py"}, {"a.py"}, ours_m, theirs_m) == {"a.py"}
112
113
114 class TestApplyMerge:
115 def test_clean_merge(self) -> None:
116 base = {"a.mid": "h0", "b.mid": "h0"}
117 ours = {"a.mid": "h_ours", "b.mid": "h0"}
118 theirs = {"a.mid": "h0", "b.mid": "h_theirs"}
119 ours_changed = {"a.mid"}
120 theirs_changed = {"b.mid"}
121 result = apply_merge(base, ours, theirs, ours_changed, theirs_changed, set())
122 assert result == {"a.mid": "h_ours", "b.mid": "h_theirs"}
123
124 def test_conflict_paths_excluded(self) -> None:
125 base = {"a.mid": "h0"}
126 ours = {"a.mid": "h_ours"}
127 theirs = {"a.mid": "h_theirs"}
128 ours_changed = theirs_changed = {"a.mid"}
129 result = apply_merge(base, ours, theirs, ours_changed, theirs_changed, {"a.mid"})
130 assert result == {"a.mid": "h0"} # Falls back to base
131
132 def test_ours_deletion_applied(self) -> None:
133 base = {"a.mid": "h0", "b.mid": "h0"}
134 ours = {"b.mid": "h0"} # a.mid deleted on ours
135 theirs = {"a.mid": "h0", "b.mid": "h0"}
136 result = apply_merge(base, ours, theirs, {"a.mid"}, set(), set())
137 assert "a.mid" not in result
138
139
140 class TestMergeStateIO:
141 def test_write_and_read(self, repo: pathlib.Path) -> None:
142 base_id = "b" * 64
143 ours_id = "1" * 64
144 theirs_id = "2" * 64
145 write_merge_state(
146 repo,
147 base_commit=base_id,
148 ours_commit=ours_id,
149 theirs_commit=theirs_id,
150 conflict_paths=["a.mid", "b.mid"],
151 other_branch="feature/x",
152 )
153 state = read_merge_state(repo)
154 assert state is not None
155 assert state.base_commit == base_id
156 assert state.conflict_paths == ["a.mid", "b.mid"]
157 assert state.other_branch == "feature/x"
158
159 def test_read_no_state(self, repo: pathlib.Path) -> None:
160 assert read_merge_state(repo) is None
161
162 def test_clear(self, repo: pathlib.Path) -> None:
163 write_merge_state(repo, base_commit="b" * 64, ours_commit="c" * 64, theirs_commit="d" * 64, conflict_paths=[])
164 clear_merge_state(repo)
165 assert read_merge_state(repo) is None
166
167
168 class TestFindMergeBase:
169 def test_direct_parent(self, repo: pathlib.Path) -> None:
170 root_id = _commit(repo, "root")
171 a_id = _commit(repo, "a", parent=root_id)
172 b_id = _commit(repo, "b", parent=root_id)
173 base = find_merge_base(repo, a_id, b_id)
174 assert base == root_id
175
176 def test_same_commit(self, repo: pathlib.Path) -> None:
177 _commit(repo, "root")
178 base = find_merge_base(repo, "root", "root")
179 assert base == "root"
180
181 def test_linear_history(self, repo: pathlib.Path) -> None:
182 a_id = _commit(repo, "a")
183 b_id = _commit(repo, "b", parent=a_id)
184 c_id = _commit(repo, "c", parent=b_id)
185 base = find_merge_base(repo, c_id, b_id)
186 assert base == b_id
187
188 def test_no_common_ancestor(self, repo: pathlib.Path) -> None:
189 _commit(repo, "x")
190 _commit(repo, "y")
191 assert find_merge_base(repo, "x", "y") is None
192
193
194 # ===========================================================================
195 # Structured merge engine integration tests
196 # ===========================================================================
197
198
199 def _ins(addr: str, pos: int | None, cid: str) -> InsertOp:
200 return InsertOp(op="insert", address=addr, position=pos, content_id=cid, content_summary=cid)
201
202
203 def _del(addr: str, pos: int | None, cid: str) -> DeleteOp:
204 return DeleteOp(op="delete", address=addr, position=pos, content_id=cid, content_summary=cid)
205
206
207 def _rep(addr: str, old: str, new: str) -> ReplaceOp:
208 return ReplaceOp(
209 op="replace",
210 address=addr,
211 position=None,
212 old_content_id=old,
213 new_content_id=new,
214 old_summary="old",
215 new_summary="new",
216 )
217
218
219 def _delta(ops: list[DomainOp]) -> StructuredDelta:
220 return StructuredDelta(domain="midi", ops=ops, summary="test")
221
222
223 class TestMergeStructuredIntegration:
224 """Verify merge_structured delegates correctly to merge_op_lists."""
225
226 def test_clean_non_overlapping_file_ops(self) -> None:
227 ours = _delta([_ins("a.mid", pos=0, cid="a-hash")])
228 theirs = _delta([_ins("b.mid", pos=0, cid="b-hash")])
229 result = merge_structured(_delta([]), ours, theirs)
230 assert result.is_clean is True
231 assert len(result.merged_ops) == 2
232
233 def test_conflicting_same_address_replaces_detected(self) -> None:
234 ours = _delta([_rep("shared.mid", "old", "v-ours")])
235 theirs = _delta([_rep("shared.mid", "old", "v-theirs")])
236 result = merge_structured(_delta([]), ours, theirs)
237 assert result.is_clean is False
238 assert len(result.conflict_ops) == 1
239
240 def test_base_ops_kept_by_both_sides_preserved(self) -> None:
241 shared = _ins("base.mid", pos=0, cid="base-cid")
242 result = merge_structured(
243 _delta([shared]),
244 _delta([shared]),
245 _delta([shared]),
246 )
247 assert result.is_clean is True
248 assert any(_op_key_tuple(op) == _op_key_tuple(shared) for op in result.merged_ops)
249
250 def test_position_adjustment_in_structured_merge(self) -> None:
251 """Non-conflicting note inserts get position-adjusted in structured merge."""
252 ours = _delta([_ins("lead.mid", pos=3, cid="note-A")])
253 theirs = _delta([_ins("lead.mid", pos=7, cid="note-B")])
254 result = merge_structured(_delta([]), ours, theirs)
255 assert result.is_clean is True
256 pos_by_cid = {
257 op["content_id"]: op["position"]
258 for op in result.merged_ops
259 if op["op"] == "insert"
260 }
261 # note-A(3): no theirs ≤ 3 → stays 3
262 assert pos_by_cid["note-A"] == 3
263 # note-B(7): ours A(3) ≤ 7 → 7+1 = 8
264 assert pos_by_cid["note-B"] == 8
265
266
267 def _op_key_tuple(op: DomainOp) -> tuple[str, ...]:
268 """Re-implementation of _op_key for test assertions."""
269 if op["op"] == "insert":
270 return ("insert", op["address"], str(op["position"]), op["content_id"])
271 if op["op"] == "delete":
272 return ("delete", op["address"], str(op["position"]), op["content_id"])
273 if op["op"] == "replace":
274 return ("replace", op["address"], str(op["position"]), op["old_content_id"], op["new_content_id"])
275 return (op["op"], op["address"])
276
277
278 class TestStructuredMergePluginProtocol:
279 """Verify MidiPlugin satisfies the StructuredMergePlugin protocol."""
280
281 def test_midi_plugin_isinstance_structured_merge_plugin(self) -> None:
282 plugin = MidiPlugin()
283 assert isinstance(plugin, StructuredMergePlugin)
284
285 def test_merge_ops_non_conflicting_files_is_clean(self) -> None:
286 plugin = MidiPlugin()
287 base = SnapshotManifest(files={}, domain="midi")
288 ours_snap = SnapshotManifest(files={"a.mid": "hash-a"}, domain="midi")
289 theirs_snap = SnapshotManifest(files={"b.mid": "hash-b"}, domain="midi")
290 ours_ops: list[DomainOp] = [_ins("a.mid", pos=None, cid="hash-a")]
291 theirs_ops: list[DomainOp] = [_ins("b.mid", pos=None, cid="hash-b")]
292
293 result = plugin.merge_ops(
294 base, ours_snap, theirs_snap, ours_ops, theirs_ops
295 )
296 assert result.is_clean is True
297 assert "a.mid" in result.merged["files"]
298 assert "b.mid" in result.merged["files"]
299
300 def test_merge_ops_conflicting_same_file_replace_not_clean(self) -> None:
301 plugin = MidiPlugin()
302 base = SnapshotManifest(files={"f.mid": "base-hash"}, domain="midi")
303 ours_snap = SnapshotManifest(files={"f.mid": "ours-hash"}, domain="midi")
304 theirs_snap = SnapshotManifest(files={"f.mid": "theirs-hash"}, domain="midi")
305 ours_ops: list[DomainOp] = [_rep("f.mid", "base-hash", "ours-hash")]
306 theirs_ops: list[DomainOp] = [_rep("f.mid", "base-hash", "theirs-hash")]
307
308 result = plugin.merge_ops(
309 base, ours_snap, theirs_snap, ours_ops, theirs_ops
310 )
311 assert not result.is_clean
312 assert "f.mid" in result.conflicts
313
314 def test_merge_ops_ours_strategy_resolves_conflict(self) -> None:
315 plugin = MidiPlugin()
316 base = SnapshotManifest(files={"f.mid": "base"}, domain="midi")
317 ours_snap = SnapshotManifest(files={"f.mid": "ours-v"}, domain="midi")
318 theirs_snap = SnapshotManifest(files={"f.mid": "theirs-v"}, domain="midi")
319 ours_ops: list[DomainOp] = [_rep("f.mid", "base", "ours-v")]
320 theirs_ops: list[DomainOp] = [_rep("f.mid", "base", "theirs-v")]
321
322 result = plugin.merge_ops(
323 base,
324 ours_snap,
325 theirs_snap,
326 ours_ops,
327 theirs_ops,
328 )
329 # Without .museattributes the conflict stands — verify conflict is reported.
330 assert not result.is_clean
331
332 def test_merge_ops_delete_on_only_one_side_is_clean(self) -> None:
333 plugin = MidiPlugin()
334 base = SnapshotManifest(files={"keep.mid": "k", "remove.mid": "r"}, domain="midi")
335 ours_snap = SnapshotManifest(files={"keep.mid": "k"}, domain="midi")
336 theirs_snap = SnapshotManifest(files={"keep.mid": "k", "remove.mid": "r"}, domain="midi")
337 ours_ops: list[DomainOp] = [_del("remove.mid", pos=None, cid="r")]
338 theirs_ops: list[DomainOp] = []
339
340 result = plugin.merge_ops(
341 base, ours_snap, theirs_snap, ours_ops, theirs_ops
342 )
343 assert result.is_clean is True
344 assert "keep.mid" in result.merged["files"]
345 assert "remove.mid" not in result.merged["files"]
346
347 def test_merge_ops_empty_changes_returns_base(self) -> None:
348 plugin = MidiPlugin()
349 base = SnapshotManifest(files={"f.mid": "h"}, domain="midi")
350 result = plugin.merge_ops(base, base, base, [], [])
351 assert result.is_clean is True
352 assert result.merged["files"] == {"f.mid": "h"}
File History 1 commit
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 151 days ago