gabriel / muse public
test_phase5_phantom_guard.py python
478 lines 20.0 KB
Raw
sha256:03f9550f64e06b3df5753fbb3dfde079ce726f56375ebd4662178f3e33d33d96 test(phase5): phantom conflict guard unit tests for all fou… Sonnet 4.6 21 hours ago
1 """TDD tests for Phase 5 — Phantom conflict guard (universal).
2
3 Audit finding: all four guard locations already have equivalent logic.
4 This file provides direct unit-level proof of each guard so regressions
5 are caught immediately rather than through multi-layer integration failures.
6
7 Issue #86 Phase 5 deliverables:
8 PG_01: detect_conflicts() — same object ID on both sides → never conflicts
9 PG_02: CodePlugin.merge() — l == r at file level → no conflict produced
10 PG_03: ops_commute() — ReplaceOp convergent (same new_content_id) → True
11 PG_04: CodePlugin.merge_ops() Step 1 — same file content → no symbol conflict
12 PG_05: CodePlugin.merge_ops() Step 1.5 — ours_id == theirs_id excludes path
13 PG_06: musehub merge_overlay — convergent edit → no ConflictEntry
14 PG_07: musehub merge_weave — convergent edit → no ConflictEntry
15 PG_08: musehub merge_replay — convergent edit → no ConflictEntry
16
17 Background
18 ----------
19 A phantom conflict occurs when both branches end up with the same object ID
20 for a file but the merge engine incorrectly reports a conflict. The guard
21 ``if ours_object_id == theirs_object_id: skip`` is logically equivalent to
22 ``if l == r:`` (CodePlugin.merge), ``ours_manifest.get(path) != theirs_manifest.get(path)``
23 (detect_conflicts), and ``frm_id != to_id`` (musehub weave). All four
24 locations must independently enforce this invariant — a single layer failing
25 is enough to corrupt the merged result.
26 """
27 from __future__ import annotations
28
29 import datetime
30 import json
31 import pathlib
32 import pytest
33
34 from muse.core.types import blob_id, fake_id
35 from muse.core.object_store import write_object
36 from muse.core.paths import heads_dir, muse_dir, ref_path
37
38
39 # ---------------------------------------------------------------------------
40 # PG_01 — detect_conflicts unit guard
41 # ---------------------------------------------------------------------------
42
43 class TestDetectConflictsGuard:
44 """detect_conflicts() must never flag a path where both sides agree."""
45
46 def test_PG_01_convergent_same_id_not_in_conflicts(self) -> None:
47 """PG_01 — detect_conflicts: same object ID on both sides → not in conflict set.
48
49 This is the most direct test of the core guard:
50 ours_manifest.get(path) != theirs_manifest.get(path)
51 When both manifests have the same object ID the condition is False
52 and the path is excluded from the returned set.
53 """
54 from muse.core.merge_engine import detect_conflicts
55
56 shared_id = "sha256:" + "a" * 64
57 other_id = "sha256:" + "b" * 64
58
59 # Both branches changed 'shared.py' but arrived at the same content.
60 # Only 'real.py' genuinely diverged.
61 ours_changed = {"shared.py", "real.py"}
62 theirs_changed = {"shared.py", "real.py"}
63 ours_manifest = {"shared.py": shared_id, "real.py": "sha256:" + "c" * 64}
64 theirs_manifest = {"shared.py": shared_id, "real.py": "sha256:" + "d" * 64}
65
66 result = detect_conflicts(ours_changed, theirs_changed, ours_manifest, theirs_manifest)
67
68 assert "shared.py" not in result, (
69 "PG_01: detect_conflicts must NOT flag shared.py — both sides have the same object ID"
70 )
71 assert "real.py" in result, (
72 "PG_01: detect_conflicts must still flag real.py — genuinely divergent"
73 )
74
75 def test_PG_01b_all_convergent_returns_empty(self) -> None:
76 """PG_01b — when ALL changed paths converged to the same ID, result is empty."""
77 from muse.core.merge_engine import detect_conflicts
78
79 shared_id = "sha256:" + "e" * 64
80 ours_changed = theirs_changed = {"a.py", "b.py"}
81 manifest = {"a.py": shared_id, "b.py": shared_id}
82
83 assert detect_conflicts(ours_changed, theirs_changed, manifest, dict(manifest)) == set(), (
84 "PG_01b: fully convergent change set must produce empty conflict set"
85 )
86
87
88 # ---------------------------------------------------------------------------
89 # PG_02 — CodePlugin.merge() file-level guard
90 # ---------------------------------------------------------------------------
91
92 class TestCodePluginMergeGuard:
93 """CodePlugin.merge() l == r at line 845 must not produce conflicts."""
94
95 def test_PG_02_same_object_id_not_conflicted(self, tmp_path: pathlib.Path) -> None:
96 """PG_02 — CodePlugin.merge(): when l == r (both sides agree), no conflict is produced.
97
98 This directly exercises the ``if l == r:`` branch in plugin.merge().
99 Even when the file differs from base (both branches independently made
100 the same change), the result must be clean.
101 """
102 from muse.plugins.code.plugin import CodePlugin
103
104 base_id = "sha256:" + "0" * 64
105 conv_id = "sha256:" + "1" * 64
106
107 base_snap = {"files": {"config.py": base_id}}
108 left_snap = {"files": {"config.py": conv_id}} # left changed it
109 right_snap = {"files": {"config.py": conv_id}} # right changed it to the same
110
111 plugin = CodePlugin()
112 result = plugin.merge(base_snap, left_snap, right_snap, repo_root=None)
113
114 assert result.conflicts == [], (
115 "PG_02: same object ID on both sides must produce no conflicts in CodePlugin.merge()"
116 )
117 assert result.merged["files"].get("config.py") == conv_id, (
118 "PG_02: merged snapshot must contain the convergent object ID"
119 )
120
121 def test_PG_02b_real_divergence_still_conflicts(self, tmp_path: pathlib.Path) -> None:
122 """PG_02b — verify the guard does NOT suppress genuine divergence."""
123 from muse.plugins.code.plugin import CodePlugin
124
125 base_id = "sha256:" + "0" * 64
126 left_id = "sha256:" + "1" * 64
127 right_id = "sha256:" + "2" * 64
128
129 base_snap = {"files": {"config.py": base_id}}
130 left_snap = {"files": {"config.py": left_id}}
131 right_snap = {"files": {"config.py": right_id}}
132
133 plugin = CodePlugin()
134 result = plugin.merge(base_snap, left_snap, right_snap, repo_root=None)
135
136 assert "config.py" in result.conflicts, (
137 "PG_02b: genuine divergence must still be detected"
138 )
139
140
141 # ---------------------------------------------------------------------------
142 # PG_03 — ops_commute convergent ReplaceOp guard
143 # ---------------------------------------------------------------------------
144
145 class TestOpsCommuteConvergentGuard:
146 """ops_commute must return True for convergent ReplaceOps at the same address."""
147
148 def test_PG_03_replace_same_new_content_id_commutes(self) -> None:
149 """PG_03 — ops_commute: ReplaceOp at same address with same new_content_id → True.
150
151 This is the Step 1 implicit guard: when both branches produce a ReplaceOp
152 for the same symbol to the same content, ops_commute returns True and the
153 symbol is NOT added to conflict_addresses.
154 """
155 from muse.core.op_merge import ops_commute
156
157 conv_id = "sha256:" + "a" * 64
158 op_a: dict = {"op": "replace", "address": "config.py::MAX_CONN", "new_content_id": conv_id}
159 op_b: dict = {"op": "replace", "address": "config.py::MAX_CONN", "new_content_id": conv_id}
160
161 assert ops_commute(op_a, op_b) is True, (
162 "PG_03: ReplaceOp at same address with same new_content_id must commute"
163 )
164
165 def test_PG_03b_replace_different_new_content_id_conflicts(self) -> None:
166 """PG_03b — genuine symbol divergence: different new_content_id → False (conflict)."""
167 from muse.core.op_merge import ops_commute
168
169 op_a: dict = {"op": "replace", "address": "config.py::MAX_CONN",
170 "new_content_id": "sha256:" + "a" * 64}
171 op_b: dict = {"op": "replace", "address": "config.py::MAX_CONN",
172 "new_content_id": "sha256:" + "b" * 64}
173
174 assert ops_commute(op_a, op_b) is False, (
175 "PG_03b: different new_content_id at same address must not commute"
176 )
177
178 def test_PG_03c_replace_different_address_commutes(self) -> None:
179 """PG_03c — ReplaceOps at different addresses always commute (independent symbols)."""
180 from muse.core.op_merge import ops_commute
181
182 op_a: dict = {"op": "replace", "address": "config.py::ALPHA",
183 "new_content_id": "sha256:" + "a" * 64}
184 op_b: dict = {"op": "replace", "address": "config.py::BETA",
185 "new_content_id": "sha256:" + "b" * 64}
186
187 assert ops_commute(op_a, op_b) is True, (
188 "PG_03c: ReplaceOps at different addresses must always commute"
189 )
190
191
192 # ---------------------------------------------------------------------------
193 # PG_04 — CodePlugin.merge_ops() Step 1 — no symbol conflict for convergent content
194 # ---------------------------------------------------------------------------
195
196 class TestMergeOpsStep1Guard:
197 """CodePlugin.merge_ops() Step 1 must not produce symbol conflicts for convergent content."""
198
199 def _init_repo(self, tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
200 dot_muse = muse_dir(tmp_path)
201 dot_muse.mkdir()
202 repo_id = fake_id("repo")
203 (dot_muse / "repo.json").write_text(json.dumps({
204 "repo_id": repo_id,
205 "domain": "code",
206 "default_branch": "main",
207 "created_at": "2025-01-01T00:00:00+00:00",
208 }), encoding="utf-8")
209 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
210 (dot_muse / "refs" / "heads").mkdir(parents=True)
211 (dot_muse / "snapshots").mkdir()
212 (dot_muse / "commits").mkdir()
213 (dot_muse / "objects").mkdir()
214 return tmp_path, repo_id
215
216 def _write_obj(self, root: pathlib.Path, content: bytes) -> str:
217 oid = blob_id(content)
218 write_object(root, oid, content)
219 return oid
220
221 def test_PG_04_merge_ops_convergent_file_no_symbol_conflict(
222 self, tmp_path: pathlib.Path
223 ) -> None:
224 """PG_04 — merge_ops(): when ours and theirs have the same file object ID,
225 Step 1 produces no symbol conflicts and Step 1.5 skips the path.
226
227 Both branches independently changed config.py to the same content (convergent).
228 merge_ops must not return any conflict addressing config.py.
229 """
230 from muse.plugins.code.plugin import CodePlugin
231 from muse.core.types import blob_id as make_id
232
233 root, _ = self._init_repo(tmp_path)
234
235 # valid Python for all three: base, convergent result
236 base_content = b"x = 1\n"
237 conv_content = b"x = 99\n"
238
239 base_id = self._write_obj(root, base_content)
240 conv_id = self._write_obj(root, conv_content)
241
242 base_snap = {"files": {"config.py": base_id}}
243 ours_snap = {"files": {"config.py": conv_id}}
244 theirs_snap = {"files": {"config.py": conv_id}} # same as ours — convergent
245
246 plugin = CodePlugin()
247 result = plugin.merge_ops(
248 base_snap, ours_snap, theirs_snap,
249 ours_ops=[], theirs_ops=[],
250 repo_root=root,
251 )
252
253 conflict_files = {
254 c.split("::")[0] if "::" in c else c
255 for c in result.conflicts
256 }
257 assert "config.py" not in conflict_files, (
258 "PG_04: convergent file (same object ID on both sides) must not produce "
259 "any conflict in merge_ops — either Step 1 or Step 1.5 must guard it"
260 )
261
262
263 # ---------------------------------------------------------------------------
264 # PG_05 — CodePlugin.merge_ops() Step 1.5 explicit exclusion
265 # ---------------------------------------------------------------------------
266
267 class TestMergeOpsStep15ExplicitGuard:
268 """Step 1.5 candidate filter must exclude paths where ours_id == theirs_id."""
269
270 def test_PG_05_step15_excludes_convergent_path(
271 self, tmp_path: pathlib.Path
272 ) -> None:
273 """PG_05 — merge_ops Step 1.5: when ours_snap[path] == theirs_snap[path],
274 the path is NOT a candidate for independence merge.
275
276 The guard at line 1260:
277 ours_snap["files"].get(p) != theirs_snap["files"].get(p)
278 excludes convergent paths from candidate_paths. We verify indirectly:
279 if the path were incorrectly processed, it would either appear in conflicts
280 or in the merged snapshot with unexpected content. The clean assertion
281 is that no conflict is produced and the correct content is in the snapshot.
282 """
283 import json
284
285 dot_muse = muse_dir(tmp_path)
286 dot_muse.mkdir()
287 repo_id = fake_id("repo")
288 (dot_muse / "repo.json").write_text(json.dumps({
289 "repo_id": repo_id,
290 "domain": "code",
291 "default_branch": "main",
292 "created_at": "2025-01-01T00:00:00+00:00",
293 }), encoding="utf-8")
294 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
295 (dot_muse / "refs" / "heads").mkdir(parents=True)
296 (dot_muse / "snapshots").mkdir()
297 (dot_muse / "commits").mkdir()
298 (dot_muse / "objects").mkdir()
299
300 # Both ours and theirs changed the same file to the same content.
301 base_content = b"value = 1\n"
302 conv_content = b"value = 99\n" # valid Python — both sides arrived here
303 base_id = blob_id(base_content)
304 conv_id = blob_id(conv_content)
305 write_object(tmp_path, base_id, base_content)
306 write_object(tmp_path, conv_id, conv_content)
307
308 base_snap = {"files": {"cfg.py": base_id}}
309 ours_snap = {"files": {"cfg.py": conv_id}}
310 theirs_snap = {"files": {"cfg.py": conv_id}} # convergent
311
312 from muse.plugins.code.plugin import CodePlugin
313 plugin = CodePlugin()
314 result = plugin.merge_ops(
315 base_snap, ours_snap, theirs_snap,
316 ours_ops=[], theirs_ops=[],
317 repo_root=tmp_path,
318 )
319
320 assert result.conflicts == [], (
321 "PG_05: Step 1.5 convergent-file guard must prevent any conflict "
322 "when ours_snap[path] == theirs_snap[path]"
323 )
324 assert result.merged["files"].get("cfg.py") == conv_id, (
325 "PG_05: merged snapshot must contain the convergent content ID"
326 )
327
328
329 # ---------------------------------------------------------------------------
330 # PG_06 — musehub merge_overlay phantom guard
331 # ---------------------------------------------------------------------------
332
333 class TestMusehubOverlayGuard:
334 """musehub merge_overlay must not create ConflictEntry for convergent edits."""
335
336 def test_PG_06_overlay_convergent_no_conflict_entry(self) -> None:
337 """PG_06 — merge_overlay: both branches changed a file to the same ID → no conflict.
338
339 Guard in merge_overlay (lines 206-213):
340 if to_manifest.get(path) != from_manifest.get(path):
341 conflicts.append(...)
342 When both sides have the same object ID the condition is False.
343 """
344 from musehub.services.proposal_merge_strategies import merge_overlay
345
346 anc_id = "sha256:" + "0" * 64
347 conv_id = "sha256:" + "1" * 64
348 other_id = "sha256:" + "2" * 64
349
350 ancestor = {"cfg.py": anc_id, "util.py": anc_id}
351 to_m = {"cfg.py": conv_id, "util.py": other_id} # both changed cfg.py convergently; util differs
352 from_m = {"cfg.py": conv_id, "util.py": anc_id} # from branch only changed cfg.py
353
354 result = merge_overlay(to_m, from_m, ancestor_manifest=ancestor)
355
356 conflict_paths = {c.path for c in result.conflicts}
357 assert "cfg.py" not in conflict_paths, (
358 "PG_06: merge_overlay must not flag cfg.py — both sides have the same object ID"
359 )
360
361 def test_PG_06b_overlay_divergent_creates_entry(self) -> None:
362 """PG_06b — genuine divergence still produces a ConflictEntry."""
363 from musehub.services.proposal_merge_strategies import merge_overlay
364
365 anc_id = "sha256:" + "0" * 64
366 to_id = "sha256:" + "1" * 64
367 frm_id = "sha256:" + "2" * 64
368
369 ancestor = {"cfg.py": anc_id}
370 to_m = {"cfg.py": to_id}
371 from_m = {"cfg.py": frm_id}
372
373 result = merge_overlay(to_m, from_m, ancestor_manifest=ancestor)
374
375 conflict_paths = {c.path for c in result.conflicts}
376 assert "cfg.py" in conflict_paths, (
377 "PG_06b: genuine divergence must produce a ConflictEntry in overlay"
378 )
379
380
381 # ---------------------------------------------------------------------------
382 # PG_07 — musehub merge_weave phantom guard
383 # ---------------------------------------------------------------------------
384
385 class TestMusehubWeaveGuard:
386 """musehub merge_weave must not create ConflictEntry for convergent edits."""
387
388 def test_PG_07_weave_convergent_no_conflict_entry(self) -> None:
389 """PG_07 — merge_weave: frm_id != to_id guard excludes convergent paths.
390
391 Guard in merge_weave (lines 282-290):
392 if frm_changed and to_changed and frm_id != to_id:
393 conflicts.append(...)
394 When both sides arrive at the same ID, frm_id == to_id → no conflict.
395 """
396 from musehub.services.proposal_merge_strategies import merge_weave
397
398 anc_id = "sha256:" + "0" * 64
399 conv_id = "sha256:" + "1" * 64
400
401 ancestor = {"cfg.py": anc_id}
402 to_m = {"cfg.py": conv_id}
403 from_m = {"cfg.py": conv_id} # convergent — same as to_m
404
405 result = merge_weave(to_m, from_m, ancestor_manifest=ancestor)
406
407 assert result.conflicts == [], (
408 "PG_07: merge_weave must not flag cfg.py — both sides converged to same ID"
409 )
410 assert result.manifest.get("cfg.py") == conv_id, (
411 "PG_07: merged manifest must contain the convergent content"
412 )
413
414 def test_PG_07b_weave_divergent_creates_entry(self) -> None:
415 """PG_07b — genuine divergence still produces a ConflictEntry."""
416 from musehub.services.proposal_merge_strategies import merge_weave
417
418 anc_id = "sha256:" + "0" * 64
419 to_id = "sha256:" + "1" * 64
420 frm_id = "sha256:" + "2" * 64
421
422 ancestor = {"cfg.py": anc_id}
423 to_m = {"cfg.py": to_id}
424 from_m = {"cfg.py": frm_id}
425
426 result = merge_weave(to_m, from_m, ancestor_manifest=ancestor)
427 assert any(c.path == "cfg.py" for c in result.conflicts), (
428 "PG_07b: genuine divergence must produce a ConflictEntry in weave"
429 )
430
431
432 # ---------------------------------------------------------------------------
433 # PG_08 — musehub merge_replay phantom guard
434 # ---------------------------------------------------------------------------
435
436 class TestMusehubReplayGuard:
437 """musehub merge_replay must not create ConflictEntry for convergent edits."""
438
439 def test_PG_08_replay_convergent_no_conflict_entry(self) -> None:
440 """PG_08 — merge_replay: to_manifest.get(path) != from_manifest[path] guard.
441
442 Guard in merge_replay (lines 349-357):
443 if to_manifest.get(path) != from_manifest[path]:
444 conflicts.append(...)
445 When both branches independently arrive at the same content, no entry is created.
446 """
447 from musehub.services.proposal_merge_strategies import merge_replay
448
449 anc_id = "sha256:" + "0" * 64
450 conv_id = "sha256:" + "1" * 64
451
452 # Both branches changed cfg.py to the same content (convergent).
453 ancestor = {"cfg.py": anc_id}
454 to_m = {"cfg.py": conv_id}
455 from_m = {"cfg.py": conv_id}
456
457 result = merge_replay(to_m, from_m, ancestor_manifest=ancestor)
458
459 assert result.conflicts == [], (
460 "PG_08: merge_replay must not flag cfg.py — both sides have the same object ID"
461 )
462
463 def test_PG_08b_replay_divergent_creates_entry(self) -> None:
464 """PG_08b — genuine divergence still produces a ConflictEntry."""
465 from musehub.services.proposal_merge_strategies import merge_replay
466
467 anc_id = "sha256:" + "0" * 64
468 to_id = "sha256:" + "1" * 64
469 frm_id = "sha256:" + "2" * 64
470
471 ancestor = {"cfg.py": anc_id}
472 to_m = {"cfg.py": to_id}
473 from_m = {"cfg.py": frm_id}
474
475 result = merge_replay(to_m, from_m, ancestor_manifest=ancestor)
476 assert any(c.path == "cfg.py" for c in result.conflicts), (
477 "PG_08b: genuine divergence must produce a ConflictEntry in replay"
478 )
File History 1 commit
sha256:03f9550f64e06b3df5753fbb3dfde079ce726f56375ebd4662178f3e33d33d96 test(phase5): phantom conflict guard unit tests for all fou… Sonnet 4.6 21 hours ago