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