test_core_merge_engine.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 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 | def test_bidirectional_terminates_early(self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: |
| 194 | """Bidirectional BFS reads O(distance_to_LCA) commits, not O(total_history).""" |
| 195 | import muse.core.store as store_mod |
| 196 | |
| 197 | # 100-commit chain: root → c0 → ... → c97 → head_a |
| 198 | # ↘ feat (branches from c97) |
| 199 | # LCA = c97, one hop from each tip |
| 200 | root = _commit(repo, "root") |
| 201 | tip = root |
| 202 | for i in range(97): |
| 203 | tip = _commit(repo, f"c{i}", parent=tip) |
| 204 | lca = tip |
| 205 | head_a = _commit(repo, "head_a", parent=lca) |
| 206 | feat = _commit(repo, "feat", parent=lca) |
| 207 | |
| 208 | call_count = 0 |
| 209 | original = store_mod.read_commit |
| 210 | |
| 211 | def counting_read(rr: pathlib.Path, cid: str) -> object: |
| 212 | nonlocal call_count |
| 213 | call_count += 1 |
| 214 | return original(rr, cid) |
| 215 | |
| 216 | monkeypatch.setattr(store_mod, "read_commit", counting_read) |
| 217 | |
| 218 | base = find_merge_base(repo, head_a, feat) |
| 219 | assert base == lca |
| 220 | # Bidirectional BFS finds LCA in ~2 reads (one per tip). |
| 221 | # Old two-phase BFS read all 99 A ancestors before touching B. |
| 222 | assert call_count <= 10 |
| 223 | |
| 224 | def test_deep_chain_diamond(self, repo: pathlib.Path) -> None: |
| 225 | """LCA correct for a long chain with diverging feature branches.""" |
| 226 | root = _commit(repo, "root") |
| 227 | tip = root |
| 228 | for i in range(50): |
| 229 | tip = _commit(repo, f"m{i}", parent=tip) |
| 230 | lca = tip |
| 231 | branch_a = _commit(repo, "a0", parent=lca) |
| 232 | branch_b = _commit(repo, "b0", parent=lca) |
| 233 | for i in range(1, 5): |
| 234 | branch_a = _commit(repo, f"a{i}", parent=branch_a) |
| 235 | branch_b = _commit(repo, f"b{i}", parent=branch_b) |
| 236 | base = find_merge_base(repo, branch_a, branch_b) |
| 237 | assert base == lca |
| 238 | |
| 239 | |
| 240 | # =========================================================================== |
| 241 | # Structured merge engine integration tests |
| 242 | # =========================================================================== |
| 243 | |
| 244 | |
| 245 | def _ins(addr: str, pos: int | None, cid: str) -> InsertOp: |
| 246 | return InsertOp(op="insert", address=addr, position=pos, content_id=cid, content_summary=cid) |
| 247 | |
| 248 | |
| 249 | def _del(addr: str, pos: int | None, cid: str) -> DeleteOp: |
| 250 | return DeleteOp(op="delete", address=addr, position=pos, content_id=cid, content_summary=cid) |
| 251 | |
| 252 | |
| 253 | def _rep(addr: str, old: str, new: str) -> ReplaceOp: |
| 254 | return ReplaceOp( |
| 255 | op="replace", |
| 256 | address=addr, |
| 257 | position=None, |
| 258 | old_content_id=old, |
| 259 | new_content_id=new, |
| 260 | old_summary="old", |
| 261 | new_summary="new", |
| 262 | ) |
| 263 | |
| 264 | |
| 265 | def _delta(ops: list[DomainOp]) -> StructuredDelta: |
| 266 | return StructuredDelta(domain="midi", ops=ops, summary="test") |
| 267 | |
| 268 | |
| 269 | class TestMergeStructuredIntegration: |
| 270 | """Verify merge_structured delegates correctly to merge_op_lists.""" |
| 271 | |
| 272 | def test_clean_non_overlapping_file_ops(self) -> None: |
| 273 | ours = _delta([_ins("a.mid", pos=0, cid="a-hash")]) |
| 274 | theirs = _delta([_ins("b.mid", pos=0, cid="b-hash")]) |
| 275 | result = merge_structured(_delta([]), ours, theirs) |
| 276 | assert result.is_clean is True |
| 277 | assert len(result.merged_ops) == 2 |
| 278 | |
| 279 | def test_conflicting_same_address_replaces_detected(self) -> None: |
| 280 | ours = _delta([_rep("shared.mid", "old", "v-ours")]) |
| 281 | theirs = _delta([_rep("shared.mid", "old", "v-theirs")]) |
| 282 | result = merge_structured(_delta([]), ours, theirs) |
| 283 | assert result.is_clean is False |
| 284 | assert len(result.conflict_ops) == 1 |
| 285 | |
| 286 | def test_base_ops_kept_by_both_sides_preserved(self) -> None: |
| 287 | shared = _ins("base.mid", pos=0, cid="base-cid") |
| 288 | result = merge_structured( |
| 289 | _delta([shared]), |
| 290 | _delta([shared]), |
| 291 | _delta([shared]), |
| 292 | ) |
| 293 | assert result.is_clean is True |
| 294 | assert any(_op_key_tuple(op) == _op_key_tuple(shared) for op in result.merged_ops) |
| 295 | |
| 296 | def test_position_adjustment_in_structured_merge(self) -> None: |
| 297 | """Non-conflicting note inserts get position-adjusted in structured merge.""" |
| 298 | ours = _delta([_ins("lead.mid", pos=3, cid="note-A")]) |
| 299 | theirs = _delta([_ins("lead.mid", pos=7, cid="note-B")]) |
| 300 | result = merge_structured(_delta([]), ours, theirs) |
| 301 | assert result.is_clean is True |
| 302 | pos_by_cid = { |
| 303 | op["content_id"]: op["position"] |
| 304 | for op in result.merged_ops |
| 305 | if op["op"] == "insert" |
| 306 | } |
| 307 | # note-A(3): no theirs ≤ 3 → stays 3 |
| 308 | assert pos_by_cid["note-A"] == 3 |
| 309 | # note-B(7): ours A(3) ≤ 7 → 7+1 = 8 |
| 310 | assert pos_by_cid["note-B"] == 8 |
| 311 | |
| 312 | |
| 313 | def _op_key_tuple(op: DomainOp) -> tuple[str, ...]: |
| 314 | """Re-implementation of _op_key for test assertions.""" |
| 315 | if op["op"] == "insert": |
| 316 | return ("insert", op["address"], str(op["position"]), op["content_id"]) |
| 317 | if op["op"] == "delete": |
| 318 | return ("delete", op["address"], str(op["position"]), op["content_id"]) |
| 319 | if op["op"] == "replace": |
| 320 | return ("replace", op["address"], str(op["position"]), op["old_content_id"], op["new_content_id"]) |
| 321 | return (op["op"], op["address"]) |
| 322 | |
| 323 | |
| 324 | class TestStructuredMergePluginProtocol: |
| 325 | """Verify MidiPlugin satisfies the StructuredMergePlugin protocol.""" |
| 326 | |
| 327 | def test_midi_plugin_isinstance_structured_merge_plugin(self) -> None: |
| 328 | plugin = MidiPlugin() |
| 329 | assert isinstance(plugin, StructuredMergePlugin) |
| 330 | |
| 331 | def test_merge_ops_non_conflicting_files_is_clean(self) -> None: |
| 332 | plugin = MidiPlugin() |
| 333 | base = SnapshotManifest(files={}, domain="midi") |
| 334 | ours_snap = SnapshotManifest(files={"a.mid": "hash-a"}, domain="midi") |
| 335 | theirs_snap = SnapshotManifest(files={"b.mid": "hash-b"}, domain="midi") |
| 336 | ours_ops: list[DomainOp] = [_ins("a.mid", pos=None, cid="hash-a")] |
| 337 | theirs_ops: list[DomainOp] = [_ins("b.mid", pos=None, cid="hash-b")] |
| 338 | |
| 339 | result = plugin.merge_ops( |
| 340 | base, ours_snap, theirs_snap, ours_ops, theirs_ops |
| 341 | ) |
| 342 | assert result.is_clean is True |
| 343 | assert "a.mid" in result.merged["files"] |
| 344 | assert "b.mid" in result.merged["files"] |
| 345 | |
| 346 | def test_merge_ops_conflicting_same_file_replace_not_clean(self) -> None: |
| 347 | plugin = MidiPlugin() |
| 348 | base = SnapshotManifest(files={"f.mid": "base-hash"}, domain="midi") |
| 349 | ours_snap = SnapshotManifest(files={"f.mid": "ours-hash"}, domain="midi") |
| 350 | theirs_snap = SnapshotManifest(files={"f.mid": "theirs-hash"}, domain="midi") |
| 351 | ours_ops: list[DomainOp] = [_rep("f.mid", "base-hash", "ours-hash")] |
| 352 | theirs_ops: list[DomainOp] = [_rep("f.mid", "base-hash", "theirs-hash")] |
| 353 | |
| 354 | result = plugin.merge_ops( |
| 355 | base, ours_snap, theirs_snap, ours_ops, theirs_ops |
| 356 | ) |
| 357 | assert not result.is_clean |
| 358 | assert "f.mid" in result.conflicts |
| 359 | |
| 360 | def test_merge_ops_ours_strategy_resolves_conflict(self) -> None: |
| 361 | plugin = MidiPlugin() |
| 362 | base = SnapshotManifest(files={"f.mid": "base"}, domain="midi") |
| 363 | ours_snap = SnapshotManifest(files={"f.mid": "ours-v"}, domain="midi") |
| 364 | theirs_snap = SnapshotManifest(files={"f.mid": "theirs-v"}, domain="midi") |
| 365 | ours_ops: list[DomainOp] = [_rep("f.mid", "base", "ours-v")] |
| 366 | theirs_ops: list[DomainOp] = [_rep("f.mid", "base", "theirs-v")] |
| 367 | |
| 368 | result = plugin.merge_ops( |
| 369 | base, |
| 370 | ours_snap, |
| 371 | theirs_snap, |
| 372 | ours_ops, |
| 373 | theirs_ops, |
| 374 | ) |
| 375 | # Without .museattributes the conflict stands — verify conflict is reported. |
| 376 | assert not result.is_clean |
| 377 | |
| 378 | def test_merge_ops_delete_on_only_one_side_is_clean(self) -> None: |
| 379 | plugin = MidiPlugin() |
| 380 | base = SnapshotManifest(files={"keep.mid": "k", "remove.mid": "r"}, domain="midi") |
| 381 | ours_snap = SnapshotManifest(files={"keep.mid": "k"}, domain="midi") |
| 382 | theirs_snap = SnapshotManifest(files={"keep.mid": "k", "remove.mid": "r"}, domain="midi") |
| 383 | ours_ops: list[DomainOp] = [_del("remove.mid", pos=None, cid="r")] |
| 384 | theirs_ops: list[DomainOp] = [] |
| 385 | |
| 386 | result = plugin.merge_ops( |
| 387 | base, ours_snap, theirs_snap, ours_ops, theirs_ops |
| 388 | ) |
| 389 | assert result.is_clean is True |
| 390 | assert "keep.mid" in result.merged["files"] |
| 391 | assert "remove.mid" not in result.merged["files"] |
| 392 | |
| 393 | def test_merge_ops_empty_changes_returns_base(self) -> None: |
| 394 | plugin = MidiPlugin() |
| 395 | base = SnapshotManifest(files={"f.mid": "h"}, domain="midi") |
| 396 | result = plugin.merge_ops(base, base, base, [], []) |
| 397 | assert result.is_clean is True |
| 398 | assert result.merged["files"] == {"f.mid": "h"} |
File History
5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62
baseline: rc14 re-baseline after rc3 store corruption recovery
Human
patch
26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e
Merge branch 'dev' into main
Human
29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce
Merge branch 'dev' into main
Human
48 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa
feat: Muse — version control for the agent era
Human
100 days ago