"""Phase 1 — Cohen action labels in merge op stream JSON. Every conflict record now carries ``ours_action`` and ``theirs_action`` fields telling agents *what each side did* rather than just *that they conflicted*. Values are the Cohen-transform labels: "inserted", "deleted", "modified". Test categories --------------- TestConflictRecordSchema — ConflictRecord and ConflictDict have the new fields TestCodePluginMergeLabels — plugin.merge() populates action labels on conflict_records TestMergeJsonOutput — muse merge --json includes action labels in conflict output """ from __future__ import annotations from collections.abc import Mapping import pathlib import pytest from muse.domain import ConflictRecord from muse.plugins.code.plugin import CodePlugin from muse.core.types import blob_id # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _oid(content: bytes) -> str: return blob_id(content) def _write_blob(root: pathlib.Path, content: bytes) -> str: from muse.core.object_store import write_object oid = _oid(content) write_object(root, oid, content) return oid def _snap(root: pathlib.Path, files: Mapping[str, bytes]) -> Mapping[str, object]: return { "files": {path: _write_blob(root, content) for path, content in files.items()}, "domain": "code", "directories": [], } _BASE_PY = b"def existing(): pass\n" _OURS_PY = b"def existing(): pass\ndef added_by_ours(): pass\n" _THEIRS_PY = b"def existing(): pass\ndef added_by_theirs(): pass\n" _OURS_MODIFIED = b"def existing(): return 1\n" _THEIRS_MODIFIED = b"def existing(): return 2\n" # --------------------------------------------------------------------------- # TestConflictRecordSchema # --------------------------------------------------------------------------- class TestConflictRecordSchema: """ConflictRecord and its dict form must carry the new action fields.""" def test_conflict_record_has_ours_action(self) -> None: rec = ConflictRecord(path="src/a.py") assert hasattr(rec, "ours_action"), "ConflictRecord missing ours_action field" def test_conflict_record_has_theirs_action(self) -> None: rec = ConflictRecord(path="src/a.py") assert hasattr(rec, "theirs_action"), "ConflictRecord missing theirs_action field" def test_conflict_record_defaults_empty_string(self) -> None: rec = ConflictRecord(path="src/a.py") assert rec.ours_action == "" assert rec.theirs_action == "" def test_conflict_record_explicit_actions(self) -> None: rec = ConflictRecord(path="src/a.py", ours_action="deleted", theirs_action="modified") assert rec.ours_action == "deleted" assert rec.theirs_action == "modified" def test_to_dict_includes_ours_action(self) -> None: rec = ConflictRecord(path="src/a.py", ours_action="inserted", theirs_action="modified") d = rec.to_dict() assert "ours_action" in d, "to_dict() missing ours_action" assert d["ours_action"] == "inserted" def test_to_dict_includes_theirs_action(self) -> None: rec = ConflictRecord(path="src/a.py", ours_action="modified", theirs_action="deleted") d = rec.to_dict() assert "theirs_action" in d, "to_dict() missing theirs_action" assert d["theirs_action"] == "deleted" def test_all_action_values_accepted(self) -> None: for action in ("inserted", "deleted", "modified"): rec = ConflictRecord(path="a.py", ours_action=action, theirs_action=action) assert rec.ours_action == action assert rec.theirs_action == action # --------------------------------------------------------------------------- # TestCodePluginMergeLabels # --------------------------------------------------------------------------- class TestCodePluginMergeLabels: """CodePlugin.merge() must populate conflict_records with action labels.""" def test_both_modified_produces_modified_labels(self, tmp_path: pathlib.Path) -> None: """Both sides changed the same file from the same base → both 'modified'.""" plugin = CodePlugin() base = _snap(tmp_path, {"src/a.py": _BASE_PY}) ours = _snap(tmp_path, {"src/a.py": _OURS_MODIFIED}) theirs = _snap(tmp_path, {"src/a.py": _THEIRS_MODIFIED}) result = plugin.merge(base, ours, theirs) assert "src/a.py" in result.conflicts recs = {r.path: r for r in result.conflict_records} assert "src/a.py" in recs, "conflict_records must include the conflicting path" assert recs["src/a.py"].ours_action == "modified" assert recs["src/a.py"].theirs_action == "modified" def test_ours_deleted_theirs_modified_labels(self, tmp_path: pathlib.Path) -> None: """Ours deleted, theirs modified → ours='deleted', theirs='modified'.""" plugin = CodePlugin() base = _snap(tmp_path, {"src/a.py": _BASE_PY}) ours = _snap(tmp_path, {}) # ours deleted the file theirs = _snap(tmp_path, {"src/a.py": _THEIRS_MODIFIED}) result = plugin.merge(base, ours, theirs) assert "src/a.py" in result.conflicts recs = {r.path: r for r in result.conflict_records} assert recs["src/a.py"].ours_action == "deleted" assert recs["src/a.py"].theirs_action == "modified" def test_ours_modified_theirs_deleted_labels(self, tmp_path: pathlib.Path) -> None: """Ours modified, theirs deleted → ours='modified', theirs='deleted'.""" plugin = CodePlugin() base = _snap(tmp_path, {"src/a.py": _BASE_PY}) ours = _snap(tmp_path, {"src/a.py": _OURS_MODIFIED}) theirs = _snap(tmp_path, {}) # theirs deleted the file result = plugin.merge(base, ours, theirs) assert "src/a.py" in result.conflicts recs = {r.path: r for r in result.conflict_records} assert recs["src/a.py"].ours_action == "modified" assert recs["src/a.py"].theirs_action == "deleted" def test_both_inserted_different_content_labels(self, tmp_path: pathlib.Path) -> None: """Both sides created the same path from nothing → both 'inserted'.""" plugin = CodePlugin() base = _snap(tmp_path, {}) # file doesn't exist at base ours = _snap(tmp_path, {"src/new.py": _OURS_MODIFIED}) theirs = _snap(tmp_path, {"src/new.py": _THEIRS_MODIFIED}) result = plugin.merge(base, ours, theirs) assert "src/new.py" in result.conflicts recs = {r.path: r for r in result.conflict_records} assert recs["src/new.py"].ours_action == "inserted" assert recs["src/new.py"].theirs_action == "inserted" def test_no_conflict_no_conflict_records(self, tmp_path: pathlib.Path) -> None: """Clean merge must not produce spurious conflict_records.""" plugin = CodePlugin() base = _snap(tmp_path, {"src/a.py": _BASE_PY}) ours = _snap(tmp_path, {"src/a.py": _OURS_MODIFIED}) theirs = _snap(tmp_path, {"src/a.py": _BASE_PY}) # theirs unchanged result = plugin.merge(base, ours, theirs) assert result.conflicts == [] assert result.conflict_records == [] def test_multiple_conflicts_each_gets_record(self, tmp_path: pathlib.Path) -> None: """Each conflicting path gets its own ConflictRecord.""" plugin = CodePlugin() base = _snap(tmp_path, {"a.py": _BASE_PY, "b.py": _BASE_PY}) ours = _snap(tmp_path, {"a.py": _OURS_MODIFIED, "b.py": _OURS_MODIFIED}) theirs = _snap(tmp_path, {"a.py": _THEIRS_MODIFIED, "b.py": _THEIRS_MODIFIED}) result = plugin.merge(base, ours, theirs) assert len(result.conflicts) == 2 assert len(result.conflict_records) == 2 paths = {r.path for r in result.conflict_records} assert "a.py" in paths assert "b.py" in paths def test_manual_strategy_conflict_record_has_labels(self, tmp_path: pathlib.Path) -> None: """Manual-strategy forced conflicts also get action labels.""" attrs_path = tmp_path / ".museattributes" attrs_path.write_text( '[meta]\ndomain = "code"\n\n' '[[rules]]\npath = "locked/**"\ndimension = "*"\nstrategy = "manual"\npriority = 10\n' ) plugin = CodePlugin() base = _snap(tmp_path, {"locked/cfg.py": _BASE_PY}) ours = _snap(tmp_path, {"locked/cfg.py": _BASE_PY}) # ours unchanged (b==l) theirs = _snap(tmp_path, {"locked/cfg.py": _THEIRS_MODIFIED}) # theirs changed result = plugin.merge(base, ours, theirs, repo_root=tmp_path) assert "locked/cfg.py" in result.conflicts recs = {r.path: r for r in result.conflict_records} # theirs_action: theirs changed from base → "modified" # ours_action: ours == base → effectively "unmodified" (no ours change) # Convention: ours_action for b==l case = "" or explicit label for the forced-manual case assert recs["locked/cfg.py"].theirs_action == "modified"