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