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