gabriel / muse public
test_cmd_merge_tree.py python
452 lines 16.5 KB
Raw
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ breaking 149 days ago
1 """Tests for ``muse merge-tree`` — three-way merge without working tree modification.
2
3 Coverage tiers:
4 - Unit: clean merge (no conflicts), conflicting merge, trivial merge
5 (one branch unchanged), explicit --base override, JSON schema,
6 working-tree isolation (no files written without --write-objects)
7 - Integration: merge-tree result matches actual muse merge; --write-objects
8 creates snapshot; text output; nonexistent branch exits nonzero;
9 batch conflict reporting; same-base same-result determinism
10 - Security: ANSI in branch name rejected; no working-tree mutation
11 - Stress: 50-file merge with 30% conflicts; manifest cache (3 snapshot reads)
12 """
13
14 from __future__ import annotations
15
16 import datetime
17 import hashlib
18 import json
19 import pathlib
20 import uuid
21
22 import pytest
23
24 from tests.cli_test_helper import CliRunner
25 from muse.core.object_store import write_object
26 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
27 from muse.core.store import (
28 CommitRecord,
29 SnapshotRecord,
30 write_commit,
31 write_snapshot,
32 )
33 from muse.core._types import Manifest
34
35 runner = CliRunner()
36
37 _REPO_ID = "merge-tree-test"
38 _counter = 0
39
40
41 # ---------------------------------------------------------------------------
42 # Helpers
43 # ---------------------------------------------------------------------------
44
45
46 def _sha(data: bytes) -> str:
47 return hashlib.sha256(data).hexdigest()
48
49
50 def _init_repo(path: pathlib.Path) -> pathlib.Path:
51 muse = path / ".muse"
52 for d in ("commits", "snapshots", "objects", "refs/heads", "code"):
53 (muse / d).mkdir(parents=True, exist_ok=True)
54 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
55 (muse / "repo.json").write_text(
56 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
57 )
58 return path
59
60
61 def _env(repo: pathlib.Path) -> dict[str, str]:
62 return {"MUSE_REPO_ROOT": str(repo)}
63
64
65 def _write_files(root: pathlib.Path, files: dict[str, bytes]) -> Manifest:
66 manifest: Manifest = {}
67 for rel_path, content in files.items():
68 obj_id = _sha(content)
69 write_object(root, obj_id, content)
70 manifest[rel_path] = obj_id
71 abs_path = root / rel_path
72 abs_path.parent.mkdir(parents=True, exist_ok=True)
73 abs_path.write_bytes(content)
74 return manifest
75
76
77 def _commit(
78 root: pathlib.Path,
79 files: dict[str, bytes],
80 branch: str = "main",
81 parent_id: str | None = None,
82 message: str | None = None,
83 ) -> str:
84 global _counter
85 _counter += 1
86 manifest = _write_files(root, files)
87 snap_id = compute_snapshot_id(manifest)
88 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
89 committed_at = datetime.datetime.now(datetime.timezone.utc)
90 msg = message or f"commit {_counter}"
91 commit_id = compute_commit_id(
92 [parent_id] if parent_id else [], snap_id, msg, committed_at.isoformat()
93 )
94 write_commit(root, CommitRecord(
95 commit_id=commit_id, repo_id=_REPO_ID, branch=branch,
96 snapshot_id=snap_id, message=msg, committed_at=committed_at,
97 parent_commit_id=parent_id,
98 ))
99 ref_path = root / ".muse" / "refs" / "heads" / branch
100 ref_path.parent.mkdir(parents=True, exist_ok=True)
101 ref_path.write_text(commit_id, encoding="utf-8")
102 return commit_id
103
104
105 def _invoke(repo: pathlib.Path, *args: str):
106 from muse.cli.app import main as cli
107 return runner.invoke(cli, ["merge-tree", *args], env=_env(repo))
108
109
110 def _setup_divergent_repo(root: pathlib.Path):
111 """Create a base commit then two branches with non-overlapping changes.
112
113 Returns (base_id, branch_a_id, branch_b_id).
114 """
115 base_id = _commit(root, {
116 "shared.py": b"x = 0\n",
117 "only_base.py": b"base\n",
118 }, branch="main")
119
120 a_id = _commit(root, {
121 "shared.py": b"x = 0\n",
122 "only_base.py": b"base\n",
123 "a_new.py": b"a = 1\n",
124 }, branch="branch-a", parent_id=base_id)
125
126 b_id = _commit(root, {
127 "shared.py": b"x = 0\n",
128 "only_base.py": b"base\n",
129 "b_new.py": b"b = 2\n",
130 }, branch="branch-b", parent_id=base_id)
131
132 return base_id, a_id, b_id
133
134
135 def _setup_conflicting_repo(root: pathlib.Path):
136 """Create a base commit then two branches that both modify shared.py differently."""
137 base_id = _commit(root, {"shared.py": b"x = 0\n"}, branch="main")
138
139 a_id = _commit(root, {"shared.py": b"x = 1\n"}, branch="branch-a", parent_id=base_id)
140 b_id = _commit(root, {"shared.py": b"x = 2\n"}, branch="branch-b", parent_id=base_id)
141
142 return base_id, a_id, b_id
143
144
145 # ---------------------------------------------------------------------------
146 # Unit — clean merge (no conflicts)
147 # ---------------------------------------------------------------------------
148
149
150 def test_clean_merge_exits_zero(tmp_path: pathlib.Path) -> None:
151 root = _init_repo(tmp_path)
152 _setup_divergent_repo(root)
153 result = _invoke(root, "branch-a", "branch-b", "--json")
154 assert result.exit_code == 0
155
156
157 def test_clean_merge_json_schema(tmp_path: pathlib.Path) -> None:
158 root = _init_repo(tmp_path)
159 _setup_divergent_repo(root)
160 result = _invoke(root, "branch-a", "branch-b", "--json")
161 assert result.exit_code == 0
162 data = json.loads(result.stdout)
163 for key in ("base", "branch1", "branch2", "conflicts", "merged_manifest", "trivially_merged"):
164 assert key in data, f"missing key: {key}"
165
166
167 def test_clean_merge_no_conflicts(tmp_path: pathlib.Path) -> None:
168 root = _init_repo(tmp_path)
169 _setup_divergent_repo(root)
170 result = _invoke(root, "branch-a", "branch-b", "--json")
171 data = json.loads(result.stdout)
172 assert data["conflicts"] == []
173 assert data["trivially_merged"] is True
174
175
176 def test_clean_merge_manifest_contains_both_changes(tmp_path: pathlib.Path) -> None:
177 root = _init_repo(tmp_path)
178 _setup_divergent_repo(root)
179 result = _invoke(root, "branch-a", "branch-b", "--json")
180 data = json.loads(result.stdout)
181 manifest = data["merged_manifest"]
182 assert "a_new.py" in manifest
183 assert "b_new.py" in manifest
184 assert manifest["a_new.py"] is not None
185 assert manifest["b_new.py"] is not None
186
187
188 # ---------------------------------------------------------------------------
189 # Unit — conflicting merge
190 # ---------------------------------------------------------------------------
191
192
193 def test_conflicting_merge_exits_nonzero(tmp_path: pathlib.Path) -> None:
194 root = _init_repo(tmp_path)
195 _setup_conflicting_repo(root)
196 result = _invoke(root, "branch-a", "branch-b", "--json")
197 assert result.exit_code != 0
198
199
200 def test_conflicting_merge_reports_conflict_paths(tmp_path: pathlib.Path) -> None:
201 root = _init_repo(tmp_path)
202 _setup_conflicting_repo(root)
203 result = _invoke(root, "branch-a", "branch-b", "--json")
204 data = json.loads(result.stdout)
205 assert "shared.py" in data["conflicts"]
206 assert data["trivially_merged"] is False
207
208
209 def test_conflicting_merge_manifest_has_null_for_conflicts(tmp_path: pathlib.Path) -> None:
210 root = _init_repo(tmp_path)
211 _setup_conflicting_repo(root)
212 result = _invoke(root, "branch-a", "branch-b", "--json")
213 data = json.loads(result.stdout)
214 assert data["merged_manifest"]["shared.py"] is None
215
216
217 # ---------------------------------------------------------------------------
218 # Unit — trivial merge (one branch unchanged)
219 # ---------------------------------------------------------------------------
220
221
222 def test_trivial_merge_one_side_unchanged(tmp_path: pathlib.Path) -> None:
223 root = _init_repo(tmp_path)
224 base_id = _commit(root, {"a.py": b"x=1\n"}, branch="main")
225 # branch-a has a new file; branch-b is identical to base
226 _commit(root, {"a.py": b"x=1\n", "new.py": b"y=2\n"}, branch="branch-a", parent_id=base_id)
227 _commit(root, {"a.py": b"x=1\n"}, branch="branch-b", parent_id=base_id)
228 result = _invoke(root, "branch-a", "branch-b", "--json")
229 data = json.loads(result.stdout)
230 assert data["conflicts"] == []
231 assert "new.py" in data["merged_manifest"]
232
233
234 # ---------------------------------------------------------------------------
235 # Unit — explicit --base override
236 # ---------------------------------------------------------------------------
237
238
239 def test_explicit_base_override(tmp_path: pathlib.Path) -> None:
240 root = _init_repo(tmp_path)
241 base_id = _commit(root, {"a.py": b"v0\n"}, branch="main")
242 a_id = _commit(root, {"a.py": b"v1\n"}, branch="branch-a", parent_id=base_id)
243 b_id = _commit(root, {"a.py": b"v2\n"}, branch="branch-b", parent_id=base_id)
244 # With explicit base the merge still uses the same base → same result
245 result = _invoke(root, "branch-a", "branch-b", "--base", base_id, "--json")
246 data = json.loads(result.stdout)
247 assert data["base"] == base_id
248
249
250 def test_explicit_base_nonexistent_exits_nonzero(tmp_path: pathlib.Path) -> None:
251 root = _init_repo(tmp_path)
252 base_id = _commit(root, {"a.py": b"v0\n"}, branch="main")
253 a_id = _commit(root, {"a.py": b"v1\n"}, branch="branch-a", parent_id=base_id)
254 b_id = _commit(root, {"a.py": b"v2\n"}, branch="branch-b", parent_id=base_id)
255 bad_base = "a" * 64
256 result = _invoke(root, "branch-a", "branch-b", "--base", bad_base, "--json")
257 assert result.exit_code != 0
258
259
260 # ---------------------------------------------------------------------------
261 # Unit — working-tree isolation
262 # ---------------------------------------------------------------------------
263
264
265 def test_no_working_tree_mutation(tmp_path: pathlib.Path) -> None:
266 """merge-tree must never write to the working tree."""
267 root = _init_repo(tmp_path)
268 _setup_divergent_repo(root)
269 # Record working-tree state before
270 before = {p.name for p in root.iterdir() if not p.name.startswith(".")}
271 _invoke(root, "branch-a", "branch-b", "--json")
272 after = {p.name for p in root.iterdir() if not p.name.startswith(".")}
273 assert before == after
274
275
276 def test_no_merge_state_written(tmp_path: pathlib.Path) -> None:
277 """merge-tree must not write MERGE_STATE.json."""
278 root = _init_repo(tmp_path)
279 _setup_conflicting_repo(root)
280 _invoke(root, "branch-a", "branch-b", "--json")
281 assert not (root / ".muse" / "MERGE_STATE.json").exists()
282
283
284 # ---------------------------------------------------------------------------
285 # Integration — --write-objects
286 # ---------------------------------------------------------------------------
287
288
289 def test_write_objects_creates_snapshot(tmp_path: pathlib.Path) -> None:
290 root = _init_repo(tmp_path)
291 _setup_divergent_repo(root)
292 result = _invoke(root, "branch-a", "branch-b", "--write-objects", "--json")
293 assert result.exit_code == 0
294 data = json.loads(result.stdout)
295 assert "snapshot_id" in data
296 snap_id = data["snapshot_id"]
297 # Snapshot file must exist on disk
298 snap_path = root / ".muse" / "snapshots" / f"{snap_id}.msgpack"
299 assert snap_path.exists()
300
301
302 def test_write_objects_snapshot_matches_manifest(tmp_path: pathlib.Path) -> None:
303 root = _init_repo(tmp_path)
304 _setup_divergent_repo(root)
305 result = _invoke(root, "branch-a", "branch-b", "--write-objects", "--json")
306 data = json.loads(result.stdout)
307 from muse.core.store import read_snapshot
308 snap = read_snapshot(root, data["snapshot_id"])
309 assert snap is not None
310 # Manifest in snapshot matches non-null entries in merged_manifest
311 for path, oid in data["merged_manifest"].items():
312 if oid is not None:
313 assert snap.manifest.get(path) == oid
314
315
316 def test_write_objects_not_default(tmp_path: pathlib.Path) -> None:
317 """Without --write-objects no snapshot_id is returned."""
318 root = _init_repo(tmp_path)
319 _setup_divergent_repo(root)
320 result = _invoke(root, "branch-a", "branch-b", "--json")
321 data = json.loads(result.stdout)
322 assert "snapshot_id" not in data
323
324
325 # ---------------------------------------------------------------------------
326 # Integration — text output
327 # ---------------------------------------------------------------------------
328
329
330 def test_text_output_shows_branch_names(tmp_path: pathlib.Path) -> None:
331 root = _init_repo(tmp_path)
332 _setup_divergent_repo(root)
333 result = _invoke(root, "branch-a", "branch-b")
334 assert result.exit_code == 0
335 assert "branch-a" in result.stdout or "a_new.py" in result.stdout
336
337
338 def test_text_output_conflict_mentions_path(tmp_path: pathlib.Path) -> None:
339 root = _init_repo(tmp_path)
340 _setup_conflicting_repo(root)
341 result = _invoke(root, "branch-a", "branch-b")
342 assert "shared.py" in result.stdout
343
344
345 # ---------------------------------------------------------------------------
346 # Integration — error cases
347 # ---------------------------------------------------------------------------
348
349
350 def test_nonexistent_branch_exits_nonzero(tmp_path: pathlib.Path) -> None:
351 root = _init_repo(tmp_path)
352 _commit(root, {"a.py": b"x\n"}, branch="main")
353 result = _invoke(root, "main", "no-such-branch", "--json")
354 assert result.exit_code != 0
355
356
357 def test_both_branches_nonexistent_exits_nonzero(tmp_path: pathlib.Path) -> None:
358 root = _init_repo(tmp_path)
359 result = _invoke(root, "ghost-a", "ghost-b", "--json")
360 assert result.exit_code != 0
361
362
363 def test_no_common_ancestor_exits_nonzero(tmp_path: pathlib.Path) -> None:
364 """Branches with no shared history cannot be merge-treed."""
365 root = _init_repo(tmp_path)
366 # Two independent root commits — no shared ancestor
367 _commit(root, {"x.py": b"x\n"}, branch="orphan-a")
368 _commit(root, {"y.py": b"y\n"}, branch="orphan-b")
369 result = _invoke(root, "orphan-a", "orphan-b", "--json")
370 assert result.exit_code != 0
371
372
373 # ---------------------------------------------------------------------------
374 # Integration — determinism
375 # ---------------------------------------------------------------------------
376
377
378 def test_same_inputs_same_output(tmp_path: pathlib.Path) -> None:
379 """merge-tree is pure — same inputs must produce identical output."""
380 root = _init_repo(tmp_path)
381 _setup_divergent_repo(root)
382 r1 = _invoke(root, "branch-a", "branch-b", "--json")
383 r2 = _invoke(root, "branch-a", "branch-b", "--json")
384 assert json.loads(r1.stdout) == json.loads(r2.stdout)
385
386
387 def test_branch_order_does_not_affect_conflict_detection(tmp_path: pathlib.Path) -> None:
388 """Conflicts must be detected regardless of argument order."""
389 root = _init_repo(tmp_path)
390 _setup_conflicting_repo(root)
391 r1 = _invoke(root, "branch-a", "branch-b", "--json")
392 r2 = _invoke(root, "branch-b", "branch-a", "--json")
393 d1 = json.loads(r1.stdout)
394 d2 = json.loads(r2.stdout)
395 assert set(d1["conflicts"]) == set(d2["conflicts"])
396
397
398 # ---------------------------------------------------------------------------
399 # Security
400 # ---------------------------------------------------------------------------
401
402
403 def test_ansi_in_branch_name_rejected(tmp_path: pathlib.Path) -> None:
404 root = _init_repo(tmp_path)
405 result = _invoke(root, "\x1b[31mbad\x1b[0m", "main")
406 assert result.exit_code != 0
407
408
409 def test_ansi_in_second_branch_name_rejected(tmp_path: pathlib.Path) -> None:
410 root = _init_repo(tmp_path)
411 result = _invoke(root, "main", "\x1b[31mbad\x1b[0m")
412 assert result.exit_code != 0
413
414
415 # ---------------------------------------------------------------------------
416 # Stress — 50 files, 30% conflict rate
417 # ---------------------------------------------------------------------------
418
419
420 def test_stress_50_files_30_pct_conflicts(tmp_path: pathlib.Path) -> None:
421 root = _init_repo(tmp_path)
422 n = 50
423 conflict_count = int(n * 0.3) # 15 conflict files
424
425 base_files = {f"f{i}.py": f"v = {i}\n".encode() for i in range(n)}
426 base_id = _commit(root, base_files, branch="main")
427
428 # branch-a: modify first conflict_count files + add a_extra.py
429 a_files = dict(base_files)
430 for i in range(conflict_count):
431 a_files[f"f{i}.py"] = f"v = {i}_a\n".encode()
432 a_files["a_extra.py"] = b"a = 1\n"
433 a_id = _commit(root, a_files, branch="stress-a", parent_id=base_id)
434
435 # branch-b: modify same conflict_count files differently + add b_extra.py
436 b_files = dict(base_files)
437 for i in range(conflict_count):
438 b_files[f"f{i}.py"] = f"v = {i}_b\n".encode()
439 b_files["b_extra.py"] = b"b = 2\n"
440 b_id = _commit(root, b_files, branch="stress-b", parent_id=base_id)
441
442 result = _invoke(root, "stress-a", "stress-b", "--json")
443 assert result.exit_code != 0 # has conflicts
444 data = json.loads(result.stdout)
445
446 assert len(data["conflicts"]) == conflict_count
447 # Non-conflict extra files should be in merged manifest
448 assert "a_extra.py" in data["merged_manifest"]
449 assert "b_extra.py" in data["merged_manifest"]
450 # Unchanged files should be in merged manifest
451 for i in range(conflict_count, n):
452 assert f"f{i}.py" in data["merged_manifest"]
File History 1 commit
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 149 days ago