"""TDD tests for conflict marker writing during muse merge. When a three-way merge produces a conflict, the conflicting file must contain Cohen Transform markers (<<<<<<< / ||||||| / ======= / >>>>>>>) in the working tree so the user (or agent) can inspect both versions and resolve manually. Before the fix, merge.py left conflicting files at their ours content without any markers, making it impossible to see what the conflict was. """ from __future__ import annotations import datetime import json import pathlib from collections.abc import Mapping import pytest from tests.cli_test_helper import CliRunner from muse.core.types import blob_id from muse.core.object_store import write_object from muse.core.paths import heads_dir, muse_dir, ref_path type Manifest = dict[str, str] runner = CliRunner() cli = None # --------------------------------------------------------------------------- # Shared helpers (same pattern as test_cmd_merge.py) # --------------------------------------------------------------------------- def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]: from muse.core.types import fake_id dot_muse = muse_dir(tmp_path) dot_muse.mkdir() repo_id = fake_id("repo") (dot_muse / "repo.json").write_text(json.dumps({ "repo_id": repo_id, "domain": "code", "default_branch": "main", "created_at": "2025-01-01T00:00:00+00:00", }), encoding="utf-8") (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8") (dot_muse / "refs" / "heads").mkdir(parents=True) (dot_muse / "snapshots").mkdir() (dot_muse / "commits").mkdir() (dot_muse / "objects").mkdir() return tmp_path, repo_id def _write_blob(root: pathlib.Path, content: bytes) -> str: oid = blob_id(content) write_object(root, oid, content) return oid def _make_commit( root: pathlib.Path, repo_id: str, branch: str, files: dict[str, bytes], parent_id: str | None = None, message: str = "commit", ) -> str: from muse.core.commits import CommitRecord, write_commit from muse.core.snapshots import SnapshotRecord, write_snapshot from muse.core.ids import hash_snapshot, hash_commit manifest: Manifest = {} for rel, content in files.items(): oid = _write_blob(root, content) manifest[rel] = oid dest = root / rel dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(content) snap_id = hash_snapshot(manifest) committed_at = datetime.datetime.now(datetime.timezone.utc) commit_id = hash_commit( parent_ids=[parent_id] if parent_id else [], snapshot_id=snap_id, message=message, committed_at_iso=committed_at.isoformat(), ) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) write_commit(root, CommitRecord( commit_id=commit_id, branch=branch, snapshot_id=snap_id, message=message, committed_at=committed_at, parent_commit_id=parent_id, )) rf = ref_path(root, branch) rf.parent.mkdir(parents=True, exist_ok=True) rf.write_text(commit_id, encoding="utf-8") return commit_id def _env(root: pathlib.Path) -> Mapping[str, str]: return {"MUSE_REPO_ROOT": str(root)} # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- class TestConflictMarkersWrittenToWorkingTree: """Conflict markers must appear in the working tree file after muse merge.""" def test_conflict_file_contains_ours_marker(self, tmp_path: pathlib.Path) -> None: """<<<<<< ours marker must be present in the conflicting file.""" root, repo_id = _init_repo(tmp_path) base_id = _make_commit(root, repo_id, "main", {"hello.txt": b"Version Base\n"}, message="base") (heads_dir(root) / "feat").write_text(base_id) _make_commit(root, repo_id, "feat", {"hello.txt": b"Version B\n"}, parent_id=base_id, message="feat") _make_commit(root, repo_id, "main", {"hello.txt": b"Version A\n"}, parent_id=base_id, message="main") runner.invoke(cli, ["merge", "feat"], env=_env(root)) content = (root / "hello.txt").read_text(encoding="utf-8") assert "<<<<<<<" in content, f"Expected conflict marker in file, got:\n{content}" def test_conflict_file_contains_theirs_separator(self, tmp_path: pathlib.Path) -> None: """======= separator must be present in the conflicting file.""" root, repo_id = _init_repo(tmp_path) base_id = _make_commit(root, repo_id, "main", {"hello.txt": b"Version Base\n"}, message="base") (heads_dir(root) / "feat").write_text(base_id) _make_commit(root, repo_id, "feat", {"hello.txt": b"Version B\n"}, parent_id=base_id, message="feat") _make_commit(root, repo_id, "main", {"hello.txt": b"Version A\n"}, parent_id=base_id, message="main") runner.invoke(cli, ["merge", "feat"], env=_env(root)) content = (root / "hello.txt").read_text(encoding="utf-8") assert "=======" in content, f"Expected ======= separator, got:\n{content}" def test_conflict_file_contains_end_marker(self, tmp_path: pathlib.Path) -> None: """>>>>>>> end marker must be present in the conflicting file.""" root, repo_id = _init_repo(tmp_path) base_id = _make_commit(root, repo_id, "main", {"hello.txt": b"Version Base\n"}, message="base") (heads_dir(root) / "feat").write_text(base_id) _make_commit(root, repo_id, "feat", {"hello.txt": b"Version B\n"}, parent_id=base_id, message="feat") _make_commit(root, repo_id, "main", {"hello.txt": b"Version A\n"}, parent_id=base_id, message="main") runner.invoke(cli, ["merge", "feat"], env=_env(root)) content = (root / "hello.txt").read_text(encoding="utf-8") assert ">>>>>>>" in content, f"Expected >>>>>>> end marker, got:\n{content}" def test_conflict_file_contains_ours_content(self, tmp_path: pathlib.Path) -> None: """The ours side content must appear in the conflict block.""" root, repo_id = _init_repo(tmp_path) base_id = _make_commit(root, repo_id, "main", {"hello.txt": b"Version Base\n"}, message="base") (heads_dir(root) / "feat").write_text(base_id) _make_commit(root, repo_id, "feat", {"hello.txt": b"Version B\n"}, parent_id=base_id, message="feat") _make_commit(root, repo_id, "main", {"hello.txt": b"Version A\n"}, parent_id=base_id, message="main") runner.invoke(cli, ["merge", "feat"], env=_env(root)) content = (root / "hello.txt").read_text(encoding="utf-8") assert "Version A" in content, f"Expected ours content in markers, got:\n{content}" def test_conflict_file_contains_theirs_content(self, tmp_path: pathlib.Path) -> None: """The theirs side content must appear in the conflict block.""" root, repo_id = _init_repo(tmp_path) base_id = _make_commit(root, repo_id, "main", {"hello.txt": b"Version Base\n"}, message="base") (heads_dir(root) / "feat").write_text(base_id) _make_commit(root, repo_id, "feat", {"hello.txt": b"Version B\n"}, parent_id=base_id, message="feat") _make_commit(root, repo_id, "main", {"hello.txt": b"Version A\n"}, parent_id=base_id, message="main") runner.invoke(cli, ["merge", "feat"], env=_env(root)) content = (root / "hello.txt").read_text(encoding="utf-8") assert "Version B" in content, f"Expected theirs content in markers, got:\n{content}" def test_conflict_file_contains_base_content(self, tmp_path: pathlib.Path) -> None: """The base content must appear in the ||||||| block (diff3 style).""" root, repo_id = _init_repo(tmp_path) base_id = _make_commit(root, repo_id, "main", {"hello.txt": b"Version Base\n"}, message="base") (heads_dir(root) / "feat").write_text(base_id) _make_commit(root, repo_id, "feat", {"hello.txt": b"Version B\n"}, parent_id=base_id, message="feat") _make_commit(root, repo_id, "main", {"hello.txt": b"Version A\n"}, parent_id=base_id, message="main") runner.invoke(cli, ["merge", "feat"], env=_env(root)) content = (root / "hello.txt").read_text(encoding="utf-8") assert "Version Base" in content, f"Expected base content in ||||||| block, got:\n{content}" def test_non_conflicting_file_has_no_markers(self, tmp_path: pathlib.Path) -> None: """A file only changed on one side must not receive conflict markers.""" root, repo_id = _init_repo(tmp_path) base_id = _make_commit(root, repo_id, "main", { "shared.txt": b"conflict line\n", "theirs_only.txt": b"stable\n", }, message="base") (heads_dir(root) / "feat").write_text(base_id) _make_commit(root, repo_id, "feat", { "shared.txt": b"Version B\n", "theirs_only.txt": b"theirs change\n", }, parent_id=base_id, message="feat") _make_commit(root, repo_id, "main", { "shared.txt": b"Version A\n", "theirs_only.txt": b"stable\n", }, parent_id=base_id, message="main") runner.invoke(cli, ["merge", "feat"], env=_env(root)) content = (root / "theirs_only.txt").read_text(encoding="utf-8") assert "<<<<<<<" not in content, ( f"No markers expected in theirs-only file, got:\n{content}" ) def test_binary_conflict_file_not_garbled(self, tmp_path: pathlib.Path) -> None: """Binary files in conflict must not have text markers written into them.""" root, repo_id = _init_repo(tmp_path) # Create a fake binary blob (null bytes make it binary) base_bytes = b"\x00\x01\x02\x03 base binary" ours_bytes = b"\x00\x01\x02\x03 ours binary" theirs_bytes = b"\x00\x01\x02\x03 theirs binary" base_id = _make_commit(root, repo_id, "main", {"data.bin": base_bytes}, message="base") (heads_dir(root) / "feat").write_text(base_id) _make_commit(root, repo_id, "feat", {"data.bin": theirs_bytes}, parent_id=base_id, message="feat") _make_commit(root, repo_id, "main", {"data.bin": ours_bytes}, parent_id=base_id, message="main") runner.invoke(cli, ["merge", "feat"], env=_env(root)) # File should exist and not contain the text marker sequence if (root / "data.bin").exists(): content = (root / "data.bin").read_bytes() assert b"<<<<<<<"[:3] not in content or b"<<<<<<< " not in content, ( "Binary conflict file must not have text markers injected" ) def test_merge_state_still_records_conflict_path(self, tmp_path: pathlib.Path) -> None: """MERGE_STATE.json must still record hello.txt as conflicted even after markers are written.""" root, repo_id = _init_repo(tmp_path) base_id = _make_commit(root, repo_id, "main", {"hello.txt": b"base\n"}, message="base") (heads_dir(root) / "feat").write_text(base_id) _make_commit(root, repo_id, "feat", {"hello.txt": b"Version B\n"}, parent_id=base_id, message="feat") _make_commit(root, repo_id, "main", {"hello.txt": b"Version A\n"}, parent_id=base_id, message="main") runner.invoke(cli, ["merge", "feat"], env=_env(root)) merge_state = json.loads( (muse_dir(root) / "MERGE_STATE.json").read_text() ) assert "hello.txt" in merge_state["conflict_paths"], ( "MERGE_STATE.json must still list hello.txt as a conflict path" ) def test_multi_file_conflict_all_files_get_markers(self, tmp_path: pathlib.Path) -> None: """Every conflicting file must receive markers, not just the first one.""" root, repo_id = _init_repo(tmp_path) base_id = _make_commit(root, repo_id, "main", { "alpha.txt": b"base alpha\n", "beta.txt": b"base beta\n", }, message="base") (heads_dir(root) / "feat").write_text(base_id) _make_commit(root, repo_id, "feat", { "alpha.txt": b"theirs alpha\n", "beta.txt": b"theirs beta\n", }, parent_id=base_id, message="feat") _make_commit(root, repo_id, "main", { "alpha.txt": b"ours alpha\n", "beta.txt": b"ours beta\n", }, parent_id=base_id, message="main") runner.invoke(cli, ["merge", "feat"], env=_env(root)) for fname in ("alpha.txt", "beta.txt"): content = (root / fname).read_text(encoding="utf-8") assert "<<<<<<<" in content, ( f"Expected conflict markers in {fname}, got:\n{content}" )