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