test_phase3_weave_union_docs.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Phase 3 — Weave-based union strategy for docs and markdown. |
| 2 | |
| 3 | The 'union' strategy in .museattributes currently silently discards theirs' |
| 4 | content (takes ours blob, ignores theirs). The correct behavior uses |
| 5 | three_way_merge_lines to interleave both sides' additions — the same |
| 6 | line-level union-resolve logic from Phase 2's _independence_merge_blob. |
| 7 | |
| 8 | This applies to any path matched by a 'union' strategy rule, primarily: |
| 9 | - docs/** (documentation additions from both branches always welcome) |
| 10 | - *.md (markdown prose additions from both branches) |
| 11 | |
| 12 | After Phase 3, both sides' additions appear in the merged blob with no |
| 13 | conflict markers and no data loss. |
| 14 | |
| 15 | Test categories |
| 16 | --------------- |
| 17 | TestUnionStrategyCorrectness — union strategy merges both sides' content |
| 18 | TestUnionStrategyNoDuplication — stable lines appear exactly once |
| 19 | TestUnionStrategyEdgeCases — ours-only, theirs-only, identical content |
| 20 | TestUnionStrategyFallback — no repo_root → graceful fallback to ours |
| 21 | """ |
| 22 | |
| 23 | from __future__ import annotations |
| 24 | from collections.abc import Mapping |
| 25 | |
| 26 | from typing import TYPE_CHECKING |
| 27 | import pathlib |
| 28 | |
| 29 | import pytest |
| 30 | |
| 31 | from muse.plugins.code.plugin import CodePlugin |
| 32 | from muse.core.types import blob_id, long_id |
| 33 | |
| 34 | if TYPE_CHECKING: |
| 35 | from muse.domain import MergeResult |
| 36 | |
| 37 | |
| 38 | # --------------------------------------------------------------------------- |
| 39 | # Helpers |
| 40 | # --------------------------------------------------------------------------- |
| 41 | |
| 42 | def _oid(content: bytes) -> str: |
| 43 | return blob_id(content) |
| 44 | |
| 45 | |
| 46 | def _write_blob(root: pathlib.Path, content: bytes) -> str: |
| 47 | from muse.core.object_store import write_object |
| 48 | oid = _oid(content) |
| 49 | write_object(root, oid, content) |
| 50 | return oid |
| 51 | |
| 52 | |
| 53 | def _snap(root: pathlib.Path, files: Mapping[str, bytes]) -> Mapping[str, object]: |
| 54 | return { |
| 55 | "files": {path: _write_blob(root, content) for path, content in files.items()}, |
| 56 | "domain": "code", |
| 57 | "directories": [], |
| 58 | } |
| 59 | |
| 60 | |
| 61 | def _read_blob(root: pathlib.Path, result: "MergeResult", path: str) -> str: |
| 62 | from muse.core.object_store import read_object |
| 63 | oid = result.merged["files"][path] |
| 64 | raw = read_object(root, oid) |
| 65 | assert raw is not None, f"merged blob for {path} not in object store" |
| 66 | return raw.decode("utf-8") |
| 67 | |
| 68 | |
| 69 | def _attrs(tmp_path: pathlib.Path, rules: list[dict]) -> None: |
| 70 | """Write a .museattributes file with the given rules.""" |
| 71 | lines = ['[meta]\ndomain = "code"\n\n'] |
| 72 | for rule in rules: |
| 73 | lines.append("[[rules]]\n") |
| 74 | for k, v in rule.items(): |
| 75 | if isinstance(v, str): |
| 76 | lines.append(f'{k} = "{v}"\n') |
| 77 | else: |
| 78 | lines.append(f"{k} = {v}\n") |
| 79 | lines.append("\n") |
| 80 | (tmp_path / ".museattributes").write_text("".join(lines)) |
| 81 | |
| 82 | |
| 83 | _DOCS_BASE = ( |
| 84 | "# Project Guide\n" |
| 85 | "\n" |
| 86 | "## Overview\n" |
| 87 | "The quick brown fox.\n" |
| 88 | ) |
| 89 | |
| 90 | _DOCS_OURS = ( |
| 91 | "# Project Guide\n" |
| 92 | "\n" |
| 93 | "## Overview\n" |
| 94 | "The quick brown fox.\n" |
| 95 | "\n" |
| 96 | "## Installation\n" |
| 97 | "Run `pip install muse`.\n" |
| 98 | ) |
| 99 | |
| 100 | _DOCS_THEIRS = ( |
| 101 | "# Project Guide\n" |
| 102 | "\n" |
| 103 | "## Overview\n" |
| 104 | "The quick brown fox.\n" |
| 105 | "\n" |
| 106 | "## Usage\n" |
| 107 | "Run `muse status`.\n" |
| 108 | ) |
| 109 | |
| 110 | |
| 111 | # --------------------------------------------------------------------------- |
| 112 | # TestUnionStrategyCorrectness |
| 113 | # --------------------------------------------------------------------------- |
| 114 | |
| 115 | class TestUnionStrategyCorrectness: |
| 116 | """Union strategy must produce a merged blob containing both sides' additions.""" |
| 117 | |
| 118 | def test_union_merges_both_sides_additions_no_conflict( |
| 119 | self, tmp_path: pathlib.Path |
| 120 | ) -> None: |
| 121 | """docs/README.md with additions on each side → clean merge, no conflict.""" |
| 122 | _attrs(tmp_path, [{"path": "docs/**", "dimension": "*", "strategy": "union", "priority": 50}]) |
| 123 | plugin = CodePlugin() |
| 124 | |
| 125 | base = _snap(tmp_path, {"docs/README.md": _DOCS_BASE.encode()}) |
| 126 | ours = _snap(tmp_path, {"docs/README.md": _DOCS_OURS.encode()}) |
| 127 | theirs = _snap(tmp_path, {"docs/README.md": _DOCS_THEIRS.encode()}) |
| 128 | |
| 129 | result = plugin.merge(base, ours, theirs, repo_root=tmp_path) |
| 130 | |
| 131 | assert "docs/README.md" not in result.conflicts, ( |
| 132 | "Union strategy must not produce a conflict for docs additions" |
| 133 | ) |
| 134 | |
| 135 | def test_union_ours_additions_present_in_merged(self, tmp_path: pathlib.Path) -> None: |
| 136 | """Ours' added section appears in the merged blob.""" |
| 137 | _attrs(tmp_path, [{"path": "docs/**", "dimension": "*", "strategy": "union", "priority": 50}]) |
| 138 | plugin = CodePlugin() |
| 139 | |
| 140 | base = _snap(tmp_path, {"docs/README.md": _DOCS_BASE.encode()}) |
| 141 | ours = _snap(tmp_path, {"docs/README.md": _DOCS_OURS.encode()}) |
| 142 | theirs = _snap(tmp_path, {"docs/README.md": _DOCS_THEIRS.encode()}) |
| 143 | |
| 144 | result = plugin.merge(base, ours, theirs, repo_root=tmp_path) |
| 145 | |
| 146 | merged = _read_blob(tmp_path, result, "docs/README.md") |
| 147 | assert "## Installation" in merged, "ours' Installation section must be in merged blob" |
| 148 | assert "pip install muse" in merged |
| 149 | |
| 150 | def test_union_theirs_additions_present_in_merged(self, tmp_path: pathlib.Path) -> None: |
| 151 | """Theirs' added section appears in the merged blob.""" |
| 152 | _attrs(tmp_path, [{"path": "docs/**", "dimension": "*", "strategy": "union", "priority": 50}]) |
| 153 | plugin = CodePlugin() |
| 154 | |
| 155 | base = _snap(tmp_path, {"docs/README.md": _DOCS_BASE.encode()}) |
| 156 | ours = _snap(tmp_path, {"docs/README.md": _DOCS_OURS.encode()}) |
| 157 | theirs = _snap(tmp_path, {"docs/README.md": _DOCS_THEIRS.encode()}) |
| 158 | |
| 159 | result = plugin.merge(base, ours, theirs, repo_root=tmp_path) |
| 160 | |
| 161 | merged = _read_blob(tmp_path, result, "docs/README.md") |
| 162 | assert "## Usage" in merged, "theirs' Usage section must be in merged blob" |
| 163 | assert "muse status" in merged |
| 164 | |
| 165 | def test_union_md_glob_rule_works(self, tmp_path: pathlib.Path) -> None: |
| 166 | """*.md rule at root level also triggers weave union.""" |
| 167 | _attrs(tmp_path, [{"path": "*.md", "dimension": "*", "strategy": "union", "priority": 10}]) |
| 168 | plugin = CodePlugin() |
| 169 | |
| 170 | base = _snap(tmp_path, {"CHANGELOG.md": b"# v1.0\n- initial\n"}) |
| 171 | ours = _snap(tmp_path, {"CHANGELOG.md": b"# v1.0\n- initial\n\n# v1.1\n- new feature\n"}) |
| 172 | theirs = _snap(tmp_path, {"CHANGELOG.md": b"# v1.0\n- initial\n\n# v1.2\n- hotfix\n"}) |
| 173 | |
| 174 | result = plugin.merge(base, ours, theirs, repo_root=tmp_path) |
| 175 | |
| 176 | assert "CHANGELOG.md" not in result.conflicts |
| 177 | merged = _read_blob(tmp_path, result, "CHANGELOG.md") |
| 178 | assert "v1.1" in merged, "ours' changelog entry must appear" |
| 179 | assert "v1.2" in merged, "theirs' changelog entry must appear" |
| 180 | |
| 181 | def test_union_merged_blob_has_no_conflict_markers(self, tmp_path: pathlib.Path) -> None: |
| 182 | """Union-merged blob must not contain <<<<<<< conflict markers.""" |
| 183 | _attrs(tmp_path, [{"path": "docs/**", "dimension": "*", "strategy": "union", "priority": 50}]) |
| 184 | plugin = CodePlugin() |
| 185 | |
| 186 | base = _snap(tmp_path, {"docs/guide.md": _DOCS_BASE.encode()}) |
| 187 | ours = _snap(tmp_path, {"docs/guide.md": _DOCS_OURS.encode()}) |
| 188 | theirs = _snap(tmp_path, {"docs/guide.md": _DOCS_THEIRS.encode()}) |
| 189 | |
| 190 | result = plugin.merge(base, ours, theirs, repo_root=tmp_path) |
| 191 | |
| 192 | merged = _read_blob(tmp_path, result, "docs/guide.md") |
| 193 | assert "<<<<<<<" not in merged |
| 194 | assert "=======" not in merged |
| 195 | assert ">>>>>>>" not in merged |
| 196 | |
| 197 | |
| 198 | # --------------------------------------------------------------------------- |
| 199 | # TestUnionStrategyNoDuplication |
| 200 | # --------------------------------------------------------------------------- |
| 201 | |
| 202 | class TestUnionStrategyNoDuplication: |
| 203 | """Stable (unchanged) lines must appear exactly once in the merged blob.""" |
| 204 | |
| 205 | def test_base_content_not_duplicated(self, tmp_path: pathlib.Path) -> None: |
| 206 | """The Overview section from base appears only once in the union merge.""" |
| 207 | _attrs(tmp_path, [{"path": "docs/**", "dimension": "*", "strategy": "union", "priority": 50}]) |
| 208 | plugin = CodePlugin() |
| 209 | |
| 210 | base = _snap(tmp_path, {"docs/README.md": _DOCS_BASE.encode()}) |
| 211 | ours = _snap(tmp_path, {"docs/README.md": _DOCS_OURS.encode()}) |
| 212 | theirs = _snap(tmp_path, {"docs/README.md": _DOCS_THEIRS.encode()}) |
| 213 | |
| 214 | result = plugin.merge(base, ours, theirs, repo_root=tmp_path) |
| 215 | |
| 216 | merged = _read_blob(tmp_path, result, "docs/README.md") |
| 217 | assert merged.count("# Project Guide") == 1, "header must appear exactly once" |
| 218 | assert merged.count("## Overview") == 1, "stable section must appear exactly once" |
| 219 | assert merged.count("The quick brown fox.") == 1, "stable content must not duplicate" |
| 220 | |
| 221 | def test_same_addition_on_both_sides_deduplicated(self, tmp_path: pathlib.Path) -> None: |
| 222 | """Both sides adding the same line → appears once in merged output.""" |
| 223 | _attrs(tmp_path, [{"path": "*.md", "dimension": "*", "strategy": "union", "priority": 10}]) |
| 224 | plugin = CodePlugin() |
| 225 | |
| 226 | base = _snap(tmp_path, {"NOTES.md": b"# Notes\n"}) |
| 227 | both = b"# Notes\n\n## Common\nAdded by both.\n" |
| 228 | ours = _snap(tmp_path, {"NOTES.md": both}) |
| 229 | theirs = _snap(tmp_path, {"NOTES.md": both}) |
| 230 | |
| 231 | result = plugin.merge(base, ours, theirs, repo_root=tmp_path) |
| 232 | |
| 233 | merged = _read_blob(tmp_path, result, "NOTES.md") |
| 234 | assert merged.count("## Common") == 1, "consensus addition must appear once" |
| 235 | |
| 236 | |
| 237 | # --------------------------------------------------------------------------- |
| 238 | # TestUnionStrategyEdgeCases |
| 239 | # --------------------------------------------------------------------------- |
| 240 | |
| 241 | class TestUnionStrategyEdgeCases: |
| 242 | """Edge cases: ours-only, theirs-only, identical content, empty base.""" |
| 243 | |
| 244 | def test_ours_only_change_preserved(self, tmp_path: pathlib.Path) -> None: |
| 245 | """When only ours changed (b == r), ours wins — no duplication.""" |
| 246 | _attrs(tmp_path, [{"path": "docs/**", "dimension": "*", "strategy": "union", "priority": 50}]) |
| 247 | plugin = CodePlugin() |
| 248 | |
| 249 | base = _snap(tmp_path, {"docs/api.md": b"# API\n"}) |
| 250 | ours = _snap(tmp_path, {"docs/api.md": b"# API\n\n## Methods\n"}) |
| 251 | theirs = _snap(tmp_path, {"docs/api.md": b"# API\n"}) # unchanged |
| 252 | |
| 253 | result = plugin.merge(base, ours, theirs, repo_root=tmp_path) |
| 254 | |
| 255 | # b == r → takes ours via the non-union path, no conflict |
| 256 | assert "docs/api.md" not in result.conflicts |
| 257 | merged = _read_blob(tmp_path, result, "docs/api.md") |
| 258 | assert "## Methods" in merged |
| 259 | |
| 260 | def test_theirs_only_change_preserved(self, tmp_path: pathlib.Path) -> None: |
| 261 | """When only theirs changed (b == l), theirs wins — no conflict.""" |
| 262 | _attrs(tmp_path, [{"path": "docs/**", "dimension": "*", "strategy": "union", "priority": 50}]) |
| 263 | plugin = CodePlugin() |
| 264 | |
| 265 | base = _snap(tmp_path, {"docs/api.md": b"# API\n"}) |
| 266 | ours = _snap(tmp_path, {"docs/api.md": b"# API\n"}) # unchanged |
| 267 | theirs = _snap(tmp_path, {"docs/api.md": b"# API\n\n## Examples\n"}) |
| 268 | |
| 269 | result = plugin.merge(base, ours, theirs, repo_root=tmp_path) |
| 270 | |
| 271 | assert "docs/api.md" not in result.conflicts |
| 272 | merged = _read_blob(tmp_path, result, "docs/api.md") |
| 273 | assert "## Examples" in merged |
| 274 | |
| 275 | def test_identical_both_sides_no_conflict(self, tmp_path: pathlib.Path) -> None: |
| 276 | """Both sides made identical changes → consensus, no duplication.""" |
| 277 | _attrs(tmp_path, [{"path": "*.md", "dimension": "*", "strategy": "union", "priority": 10}]) |
| 278 | plugin = CodePlugin() |
| 279 | |
| 280 | base = _snap(tmp_path, {"README.md": b"# Project\n"}) |
| 281 | both = b"# Project\n\nBrief description.\n" |
| 282 | ours = _snap(tmp_path, {"README.md": both}) |
| 283 | theirs = _snap(tmp_path, {"README.md": both}) |
| 284 | |
| 285 | result = plugin.merge(base, ours, theirs, repo_root=tmp_path) |
| 286 | |
| 287 | assert "README.md" not in result.conflicts |
| 288 | merged = _read_blob(tmp_path, result, "README.md") |
| 289 | assert merged.count("Brief description.") == 1 |
| 290 | |
| 291 | |
| 292 | # --------------------------------------------------------------------------- |
| 293 | # TestUnionStrategyFallback |
| 294 | # --------------------------------------------------------------------------- |
| 295 | |
| 296 | class TestUnionStrategyFallback: |
| 297 | """Without repo_root, union must not crash — graceful fallback.""" |
| 298 | |
| 299 | def test_no_repo_root_does_not_crash(self) -> None: |
| 300 | """merge() called without repo_root still returns a MergeResult.""" |
| 301 | plugin = CodePlugin() |
| 302 | # No repo_root, so object store is unavailable. The attrs won't load |
| 303 | # (load_attributes requires a path), so the union path isn't reached — |
| 304 | # this just confirms we don't regress on the no-root fast path. |
| 305 | base = {"files": {"a.md": long_id("a" * 64)}, "domain": "code", "directories": []} |
| 306 | ours = {"files": {"a.md": long_id("b" * 64)}, "domain": "code", "directories": []} |
| 307 | theirs = {"files": {"a.md": long_id("c" * 64)}, "domain": "code", "directories": []} |
| 308 | |
| 309 | result = plugin.merge(base, ours, theirs) # no repo_root |
| 310 | assert result is not None |