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