gabriel / muse public
test_cmd_merge_base_and_snapshot_diff.py python
503 lines 19.1 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 125 days ago
1 """Comprehensive tests for ``muse merge-base`` and ``snapshot-diff``.
2
3 Coverage tiers
4 --------------
5 - Integration: linear ancestor, diverged branches, no common ancestor,
6 branch name resolution, HEAD resolution, JSON/text format
7 - Security: ANSI in paths stripped in text mode, errors to stderr
8 - Stress: 10-commit chain merge-base, 50-path manifest diff
9 """
10 from __future__ import annotations
11
12 import datetime
13 import json
14 import pathlib
15
16 from muse.core.errors import ExitCode
17 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
18 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
19 from muse.core.types import Manifest
20 from muse.core.paths import head_path, muse_dir, ref_path
21 from tests.cli_test_helper import CliRunner, InvokeResult
22
23 runner = CliRunner()
24
25 _DT = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
26
27
28 # ---------------------------------------------------------------------------
29 # Helpers
30 # ---------------------------------------------------------------------------
31
32 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
33 repo = tmp_path / "repo"
34 dot_muse = muse_dir(repo)
35 for sub in ("objects", "commits", "snapshots", "refs/heads"):
36 (dot_muse / sub).mkdir(parents=True)
37 (dot_muse / "HEAD").write_text("ref: refs/heads/main")
38 (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"}))
39 return repo
40
41
42 def _snap(
43 repo: pathlib.Path,
44 *,
45 manifest: Manifest | None = None,
46 ) -> str:
47 """Write a snapshot with a real content-addressed ID; return the ID."""
48 m = manifest if manifest is not None else {}
49 sid = compute_snapshot_id(m)
50 write_snapshot(repo, SnapshotRecord(
51 snapshot_id=sid,
52 manifest=m,
53 created_at=_DT,
54 ))
55 return sid
56
57
58 def _commit(
59 repo: pathlib.Path,
60 snap_id: str,
61 *,
62 message: str = "test",
63 parent: str | None = None,
64 parent2: str | None = None,
65 branch: str = "main",
66 ) -> str:
67 """Write a commit with a real content-addressed ID; return the ID."""
68 parent_ids: list[str] = [p for p in [parent, parent2] if p is not None]
69 cid = compute_commit_id( parent_ids=parent_ids,
70 snapshot_id=snap_id,
71 message=message,
72 committed_at_iso=_DT.isoformat(),
73 )
74 write_commit(repo, CommitRecord(
75 commit_id=cid,
76 repo_id="test-repo",
77 branch=branch,
78 snapshot_id=snap_id,
79 message=message,
80 committed_at=_DT,
81 parent_commit_id=parent,
82 parent2_commit_id=parent2,
83 ))
84 return cid
85
86
87 def _set_head(repo: pathlib.Path, branch: str, commit_id: str) -> None:
88 ref = ref_path(repo, branch)
89 ref.parent.mkdir(parents=True, exist_ok=True)
90 ref.write_text(commit_id)
91 (head_path(repo)).write_text(f"ref: refs/heads/{branch}")
92
93
94 def _mb(repo: pathlib.Path, *args: str) -> InvokeResult:
95 from muse.cli.app import main as cli
96 return runner.invoke(
97 cli,
98 ["merge-base", *args],
99 env={"MUSE_REPO_ROOT": str(repo)},
100 )
101
102
103 def _sd(repo: pathlib.Path, *args: str) -> InvokeResult:
104 from muse.cli.app import main as cli
105 return runner.invoke(
106 cli,
107 ["snapshot-diff", *args],
108 env={"MUSE_REPO_ROOT": str(repo)},
109 )
110
111
112 def _fake_oid(n: int) -> str:
113 return format(n, "064x")
114
115
116 # ===========================================================================
117 # merge-base tests
118 # ===========================================================================
119
120
121 class TestMergeBase:
122 def test_same_commit_is_its_own_base(self, tmp_path: pathlib.Path) -> None:
123 repo = _make_repo(tmp_path)
124 sid = _snap(repo)
125 cid = _commit(repo, sid, message="solo")
126 result = _mb(repo, "--json", cid, cid)
127 assert result.exit_code == 0
128 data = json.loads(result.output)
129 assert data["merge_base"] == cid
130
131 def test_linear_chain_base_is_parent(self, tmp_path: pathlib.Path) -> None:
132 repo = _make_repo(tmp_path)
133 sid = _snap(repo)
134 c1 = _commit(repo, sid, message="c1")
135 c2 = _commit(repo, sid, message="c2", parent=c1)
136 result = _mb(repo, "--json", c1, c2)
137 assert result.exit_code == 0
138 data = json.loads(result.output)
139 assert data["merge_base"] == c1
140
141 def test_diverged_branches_find_common_ancestor(self, tmp_path: pathlib.Path) -> None:
142 """
143 base → left
144 → right
145 merge-base(left, right) == base
146 """
147 repo = _make_repo(tmp_path)
148 sid = _snap(repo)
149 base = _commit(repo, sid, message="base")
150 left = _commit(repo, sid, message="left", parent=base)
151 right = _commit(repo, sid, message="right", parent=base)
152 result = _mb(repo, "--json", left, right)
153 assert result.exit_code == 0
154 data = json.loads(result.output)
155 assert data["merge_base"] == base
156
157 def test_unrelated_commits_no_common_ancestor(self, tmp_path: pathlib.Path) -> None:
158 repo = _make_repo(tmp_path)
159 sid = _snap(repo)
160 c1 = _commit(repo, sid, message="unrelated-c1")
161 c2 = _commit(repo, sid, message="unrelated-c2")
162 result = _mb(repo, "--json", c1, c2)
163 assert result.exit_code == 0
164 data = json.loads(result.output)
165 assert data["merge_base"] is None
166 assert "error" in data
167
168 def test_branch_name_resolution(self, tmp_path: pathlib.Path) -> None:
169 repo = _make_repo(tmp_path)
170 sid = _snap(repo)
171 cid = _commit(repo, sid, message="branch-res")
172 _set_head(repo, "main", cid)
173 result = _mb(repo, "--json", "main", cid)
174 assert result.exit_code == 0
175 data = json.loads(result.output)
176 assert data["merge_base"] == cid
177
178 def test_head_resolution(self, tmp_path: pathlib.Path) -> None:
179 repo = _make_repo(tmp_path)
180 sid = _snap(repo)
181 cid = _commit(repo, sid, message="head-res")
182 _set_head(repo, "main", cid)
183 result = _mb(repo, "--json", "HEAD", cid)
184 assert result.exit_code == 0
185 data = json.loads(result.output)
186 assert data["merge_base"] == cid
187
188 def test_text_format_prints_bare_id(self, tmp_path: pathlib.Path) -> None:
189 repo = _make_repo(tmp_path)
190 sid = _snap(repo)
191 cid = _commit(repo, sid, message="text-bare")
192 result = _mb(repo, cid, cid)
193 assert result.exit_code == 0
194 assert cid in result.output
195
196 def test_text_format_no_ancestor(self, tmp_path: pathlib.Path) -> None:
197 repo = _make_repo(tmp_path)
198 sid = _snap(repo)
199 c1 = _commit(repo, sid, message="no-anc-c1")
200 c2 = _commit(repo, sid, message="no-anc-c2")
201 result = _mb(repo, c1, c2)
202 assert result.exit_code == 0
203 assert "no common ancestor" in result.output
204
205 def test_invalid_ref_errors(self, tmp_path: pathlib.Path) -> None:
206 repo = _make_repo(tmp_path)
207 sid = _snap(repo)
208 cid = _commit(repo, sid, message="inv-ref")
209 result = _mb(repo, cid, "nonexistent-branch")
210 assert result.exit_code == ExitCode.USER_ERROR
211
212 def test_no_traceback_on_bad_ref(self, tmp_path: pathlib.Path) -> None:
213 repo = _make_repo(tmp_path)
214 result = _mb(repo, "bad", "refs")
215 assert "Traceback" not in result.output
216
217 def test_10_commit_chain(self, tmp_path: pathlib.Path) -> None:
218 repo = _make_repo(tmp_path)
219 sid = _snap(repo)
220 ids: list[str] = []
221 for i in range(10):
222 parent = ids[i - 1] if i > 0 else None
223 cid = _commit(repo, sid, message=f"chain-{i}", parent=parent)
224 ids.append(cid)
225 result = _mb(repo, "--json", ids[-1], ids[5])
226 assert result.exit_code == 0
227 data = json.loads(result.output)
228 assert data["merge_base"] == ids[5]
229
230
231 # ===========================================================================
232 # snapshot-diff tests
233 # ===========================================================================
234
235
236 class TestSnapshotDiff:
237 def test_identical_snapshots_zero_changes(self, tmp_path: pathlib.Path) -> None:
238 repo = _make_repo(tmp_path)
239 sid = _snap(repo, manifest={"a.py": _fake_oid(1)})
240 result = _sd(repo, "--json", sid, sid)
241 assert result.exit_code == 0
242 data = json.loads(result.output)
243 assert data["total_changes"] == 0
244 assert data["added"] == []
245 assert data["modified"] == []
246 assert data["deleted"] == []
247
248 def test_added_files(self, tmp_path: pathlib.Path) -> None:
249 repo = _make_repo(tmp_path)
250 sa = _snap(repo, manifest={})
251 sb = _snap(repo, manifest={"new.py": _fake_oid(1)})
252 data = json.loads(_sd(repo, "--json", sa, sb).output)
253 assert len(data["added"]) == 1
254 assert data["added"][0]["path"] == "new.py"
255
256 def test_deleted_files(self, tmp_path: pathlib.Path) -> None:
257 repo = _make_repo(tmp_path)
258 sa = _snap(repo, manifest={"old.py": _fake_oid(1)})
259 sb = _snap(repo, manifest={})
260 data = json.loads(_sd(repo, "--json", sa, sb).output)
261 assert len(data["deleted"]) == 1
262 assert data["deleted"][0]["path"] == "old.py"
263
264 def test_modified_files(self, tmp_path: pathlib.Path) -> None:
265 repo = _make_repo(tmp_path)
266 sa = _snap(repo, manifest={"main.py": _fake_oid(1)})
267 sb = _snap(repo, manifest={"main.py": _fake_oid(2)})
268 data = json.loads(_sd(repo, "--json", sa, sb).output)
269 assert len(data["modified"]) == 1
270 assert data["modified"][0]["path"] == "main.py"
271
272 def test_text_format_prefixes(self, tmp_path: pathlib.Path) -> None:
273 repo = _make_repo(tmp_path)
274 sa = _snap(repo, manifest={"old.py": _fake_oid(1)})
275 sb = _snap(repo, manifest={"new.py": _fake_oid(2)})
276 result = _sd(repo, sa, sb)
277 assert result.exit_code == 0
278 assert "A new.py" in result.output
279 assert "D old.py" in result.output
280
281 def test_stat_flag_appends_summary(self, tmp_path: pathlib.Path) -> None:
282 repo = _make_repo(tmp_path)
283 sa = _snap(repo, manifest={})
284 sb = _snap(repo, manifest={"x.py": _fake_oid(1), "y.py": _fake_oid(2)})
285 result = _sd(repo, "--stat", sa, sb)
286 assert "2 added" in result.output
287
288 def test_commit_id_resolution(self, tmp_path: pathlib.Path) -> None:
289 """snapshot-diff should accept a commit ID and resolve its snapshot."""
290 repo = _make_repo(tmp_path)
291 sid = _snap(repo, manifest={"x.py": _fake_oid(1)})
292 cid = _commit(repo, sid, message="cid-res")
293 data = json.loads(_sd(repo, "--json", sid, cid).output)
294 assert data["total_changes"] == 0
295
296 def test_invalid_ref_errors(self, tmp_path: pathlib.Path) -> None:
297 repo = _make_repo(tmp_path)
298 result = _sd(repo, "notexist", "also-not")
299 assert result.exit_code == ExitCode.USER_ERROR
300
301 def test_no_traceback_on_bad_ref(self, tmp_path: pathlib.Path) -> None:
302 repo = _make_repo(tmp_path)
303 result = _sd(repo, "bad", "also-bad")
304 assert "Traceback" not in result.output
305
306 def test_ansi_in_paths_stripped_text_mode(self, tmp_path: pathlib.Path) -> None:
307 repo = _make_repo(tmp_path)
308 malicious_path = "\x1b[31mmalicious.py\x1b[0m"
309 sa = _snap(repo, manifest={})
310 sb = _snap(repo, manifest={malicious_path: _fake_oid(3)})
311 result = _sd(repo, sa, sb)
312 assert result.exit_code == 0
313 assert "\x1b" not in result.output
314
315 def test_50_path_manifest_diff(self, tmp_path: pathlib.Path) -> None:
316 repo = _make_repo(tmp_path)
317 manifest_a = {f"src/file{i:03d}.py": _fake_oid(i) for i in range(50)}
318 manifest_b = {f"src/file{i:03d}.py": _fake_oid(i + 100) for i in range(50)}
319 sa = _snap(repo, manifest=manifest_a)
320 sb = _snap(repo, manifest=manifest_b)
321 data = json.loads(_sd(repo, "--json", sa, sb).output)
322 assert data["total_changes"] == 50
323 assert len(data["modified"]) == 50
324
325
326 # ===========================================================================
327 # Unit tests for private helpers
328 # ===========================================================================
329
330
331 class TestMergeBaseUnit:
332 def test_resolve_ref_branch_name(self, tmp_path: pathlib.Path) -> None:
333 from muse.cli.commands.merge_base import _resolve_ref
334 repo = _make_repo(tmp_path)
335 sid = _snap(repo)
336 cid = _commit(repo, sid, message="branch-resolve", branch="main")
337 _set_head(repo, "main", cid)
338 result = _resolve_ref(repo, "main")
339 assert result == cid
340
341 def test_resolve_ref_head(self, tmp_path: pathlib.Path) -> None:
342 from muse.cli.commands.merge_base import _resolve_ref
343 repo = _make_repo(tmp_path)
344 sid = _snap(repo)
345 cid = _commit(repo, sid, message="head-resolve", branch="main")
346 _set_head(repo, "main", cid)
347 assert _resolve_ref(repo, "HEAD") == cid
348
349 def test_resolve_ref_commit_id(self, tmp_path: pathlib.Path) -> None:
350 from muse.cli.commands.merge_base import _resolve_ref
351 repo = _make_repo(tmp_path)
352 sid = _snap(repo)
353 cid = _commit(repo, sid, message="cid-resolve", branch="main")
354 assert _resolve_ref(repo, cid) == cid
355
356 def test_resolve_ref_nonexistent_returns_none(self, tmp_path: pathlib.Path) -> None:
357 from muse.cli.commands.merge_base import _resolve_ref
358 repo = _make_repo(tmp_path)
359 assert _resolve_ref(repo, f"deadbeef{'0' * 56}") is None
360
361 def test_resolve_ref_invalid_hex_returns_none(self, tmp_path: pathlib.Path) -> None:
362 from muse.cli.commands.merge_base import _resolve_ref
363 repo = _make_repo(tmp_path)
364 assert _resolve_ref(repo, "not-valid") is None
365
366
367 class TestSnapshotDiffUnit:
368 def test_added_entry_fields(self) -> None:
369 from muse.cli.commands.snapshot_diff import _AddedEntry
370 fields = set(_AddedEntry.__annotations__.keys())
371 assert "path" in fields
372 assert "object_id" in fields
373
374 def test_modified_entry_fields(self) -> None:
375 from muse.cli.commands.snapshot_diff import _ModifiedEntry
376 fields = set(_ModifiedEntry.__annotations__.keys())
377 assert "path" in fields
378 assert "object_id_a" in fields
379 assert "object_id_b" in fields
380
381 def test_deleted_entry_fields(self) -> None:
382 from muse.cli.commands.snapshot_diff import _DeletedEntry
383 fields = set(_DeletedEntry.__annotations__.keys())
384 assert "path" in fields
385 assert "object_id" in fields
386
387 def test_diff_result_fields(self) -> None:
388 from muse.cli.commands.snapshot_diff import _DiffResult
389 fields = set(_DiffResult.__annotations__.keys())
390 assert "snapshot_a" in fields
391 assert "snapshot_b" in fields
392 assert "added" in fields
393 assert "modified" in fields
394 assert "deleted" in fields
395 assert "total_changes" in fields
396
397 def test_resolve_to_snapshot_id_branch(self, tmp_path: pathlib.Path) -> None:
398 from muse.cli.commands.snapshot_diff import _resolve_to_snapshot_id
399 repo = _make_repo(tmp_path)
400 sid = _snap(repo)
401 cid = _commit(repo, sid, message="branch-snap-res", branch="main")
402 _set_head(repo, "main", cid)
403 result = _resolve_to_snapshot_id(repo, "main")
404 assert result == sid
405
406 def test_resolve_to_snapshot_id_head(self, tmp_path: pathlib.Path) -> None:
407 from muse.cli.commands.snapshot_diff import _resolve_to_snapshot_id
408 repo = _make_repo(tmp_path)
409 sid = _snap(repo)
410 cid = _commit(repo, sid, message="head-snap-res", branch="main")
411 _set_head(repo, "main", cid)
412 result = _resolve_to_snapshot_id(repo, "HEAD")
413 assert result == sid
414
415 def test_resolve_to_snapshot_id_direct(self, tmp_path: pathlib.Path) -> None:
416 from muse.cli.commands.snapshot_diff import _resolve_to_snapshot_id
417 repo = _make_repo(tmp_path)
418 sid = _snap(repo)
419 result = _resolve_to_snapshot_id(repo, sid)
420 assert result == sid
421
422 def test_resolve_to_snapshot_id_invalid_returns_none(self, tmp_path: pathlib.Path) -> None:
423 from muse.cli.commands.snapshot_diff import _resolve_to_snapshot_id
424 repo = _make_repo(tmp_path)
425 assert _resolve_to_snapshot_id(repo, "not-valid") is None
426
427 def test_resolve_to_snapshot_id_missing_returns_none(self, tmp_path: pathlib.Path) -> None:
428 from muse.cli.commands.snapshot_diff import _resolve_to_snapshot_id
429 repo = _make_repo(tmp_path)
430 assert _resolve_to_snapshot_id(repo, f"ab{'0' * 62}") is None
431
432
433 # ===========================================================================
434 # Additional security & format tests
435 # ===========================================================================
436
437
438 class TestMergeBaseSecurity:
439 def test_format_error_to_stderr(self, tmp_path: pathlib.Path) -> None:
440 repo = _make_repo(tmp_path)
441 r = _mb(repo, "--format", "xml", "main", "dev")
442 assert r.exit_code != 0
443 assert r.stdout_bytes == b""
444 assert r.stderr.strip() # some message emitted to stderr
445
446 def test_no_traceback_on_bad_format(self, tmp_path: pathlib.Path) -> None:
447 repo = _make_repo(tmp_path)
448 r = _mb(repo, "--format", "bad", "main", "dev")
449 assert "Traceback" not in r.output
450
451 def test_json_shorthand(self, tmp_path: pathlib.Path) -> None:
452 repo = _make_repo(tmp_path)
453 sid = _snap(repo)
454 cid = _commit(repo, sid, message="json-sh")
455 _set_head(repo, "main", cid)
456 r = _mb(repo, "--json", cid, cid)
457 assert r.exit_code == 0
458 d = json.loads(r.output)
459 assert d["merge_base"] == cid
460
461 def test_200_sequential_merge_base_calls(self, tmp_path: pathlib.Path) -> None:
462 repo = _make_repo(tmp_path)
463 sid = _snap(repo)
464 c1 = _commit(repo, sid, message="seq-mb-c1")
465 c2 = _commit(repo, sid, message="seq-mb-c2", parent=c1)
466 _set_head(repo, "main", c2)
467 for i in range(200):
468 r = _mb(repo, c1, c2)
469 assert r.exit_code == 0, f"failed at {i}"
470
471
472 class TestSnapshotDiffSecurity:
473 def test_format_error_to_stderr(self, tmp_path: pathlib.Path) -> None:
474 repo = _make_repo(tmp_path)
475 sid = _snap(repo)
476 r = _sd(repo, "--format", "xml", sid, sid)
477 assert r.exit_code != 0
478 assert r.stdout_bytes == b""
479 assert "error" in r.stderr.lower()
480
481 def test_no_traceback_on_bad_format(self, tmp_path: pathlib.Path) -> None:
482 repo = _make_repo(tmp_path)
483 sid = _snap(repo)
484 r = _sd(repo, "--format", "bad", sid, sid)
485 assert "Traceback" not in r.output
486
487 def test_json_shorthand(self, tmp_path: pathlib.Path) -> None:
488 repo = _make_repo(tmp_path)
489 sa = _snap(repo, manifest={"a.py": _fake_oid(1)})
490 sb = _snap(repo, manifest={"b.py": _fake_oid(2)})
491 r = _sd(repo, "--json", sa, sb)
492 assert r.exit_code == 0
493 d = json.loads(r.output)
494 assert "added" in d
495 assert "deleted" in d
496 assert "modified" in d
497
498 def test_200_sequential_snapshot_diff_calls(self, tmp_path: pathlib.Path) -> None:
499 repo = _make_repo(tmp_path)
500 sid = _snap(repo)
501 for i in range(200):
502 r = _sd(repo, sid, sid)
503 assert r.exit_code == 0, f"failed at {i}"
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 125 days ago