gabriel / muse public
test_phase3_weave_union_docs.py python
310 lines 12.8 KB
Raw
sha256:e6465e8a9b7fa8e6223ed4a3576e96c568c913ae2caeb9c31f15e7a81b250b40 docs: add | jq convention to --json section of agent-guide Sonnet 4.6 1 day 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 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
File History 1 commit
sha256:e6465e8a9b7fa8e6223ed4a3576e96c568c913ae2caeb9c31f15e7a81b250b40 docs: add | jq convention to --json section of agent-guide Sonnet 4.6 1 day ago