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