gabriel / muse public
test_gc_full.py python
921 lines 38.8 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 139 days ago
1 """Comprehensive tests for ``muse gc --full`` — orphaned commit + snapshot pruning.
2
3 Coverage dimensions
4 -------------------
5
6 Unit
7 ~~~~
8 - ``_collect_reachable_commits``: empty repo, single branch, multi-branch,
9 parent chain traversal, merge commits (2 parents), missing files, corrupt
10 files, symlink guard, cycle resistance, tags included
11 - ``_collect_reachable_snapshots``: snapshot IDs from reachable commits,
12 blob IDs from manifests, shelf objects preserved
13 - ``_list_stored_msgpack``: enumerates files, grace period, symlink guard,
14 non-.msgpack files skipped
15 - ``GcResult``: new fields default to zero
16
17 Integration (run_gc)
18 ~~~~~~~~~~~~~~~~~~~~
19 - Orphaned commit deleted; reachable commit preserved
20 - Orphaned snapshot deleted; reachable snapshot preserved
21 - Orphaned blobs from orphaned commits deleted under --full
22 - dry_run=True never deletes commits or snapshots
23 - Multiple branches: any-branch reachability preserved
24 - Linear commit chain: all intermediates preserved
25 - Merge commit (2 parents): both parent chains preserved
26 - Grace period protects recently-written commits and snapshots
27 - GcResult fields populated correctly
28 - run_gc(full=False) does NOT delete orphaned commits/snapshots
29 - Idempotency: second run collects nothing
30
31 CLI
32 ~~~
33 - ``muse gc --full`` text output has commits + snapshots lines
34 - ``muse gc --full --dry-run`` text output prefixed with [dry-run]
35 - ``muse gc --full --json`` output includes all new fields with correct types
36 - ``muse gc --full --json --dry-run`` dry_run field is True
37 - ``muse gc --full`` without orphans reports 0 collected
38 - ``muse gc --json`` (no --full) schema unchanged — new fields present at 0
39
40 E2E
41 ~~~
42 - Full lifecycle: orphaned commits/snapshots accumulate, gc --full reclaims them
43 - After branch deletion, unique commits/snapshots GCed under --full
44 - Shelf blob objects protected under --full
45 - Rewritten history: old commits removed, new commits preserved
46
47 Security
48 ~~~~~~~~
49 - Symlinked commit file not deleted by --full
50 - Symlinked snapshot file not deleted by --full
51 - Non-.msgpack file in commits dir skipped (not deleted)
52 - Non-.msgpack file in snapshots dir skipped (not deleted)
53
54 Stress
55 ~~~~~~
56 - 200 orphaned commits + 200 orphaned snapshots collected correctly
57 - Deep 100-commit chain: all commits preserved under --full
58 """
59
60 from __future__ import annotations
61
62 type _FileStore = dict[str, bytes]
63
64 import datetime
65 import hashlib
66 import json
67 import os
68 import pathlib
69
70 import msgpack
71 import pytest
72
73 from muse.core.gc import (
74 GcResult,
75 _collect_reachable_commits,
76 _collect_reachable_snapshots,
77 _list_stored_msgpack,
78 run_gc,
79 )
80 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
81 from muse.core.store import CommitRecord, SnapshotRecord, commit_path, snapshot_path, write_commit, write_snapshot
82 from muse.core._types import Manifest, blob_id, long_id, fake_id, split_id
83 from muse.core.object_store import write_object as _write_obj_atomic
84 from muse.core.object_store import object_path
85
86 from tests.cli_test_helper import CliRunner, InvokeResult
87
88 cli = None
89 runner = CliRunner()
90
91 _EPOCH = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
92
93
94 # ---------------------------------------------------------------------------
95 # Helpers
96 # ---------------------------------------------------------------------------
97
98
99 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
100 muse = tmp_path / ".muse"
101 for sub in ("objects", "commits", "snapshots", "refs/heads"):
102 (muse / sub).mkdir(parents=True, exist_ok=True)
103 (muse / "repo.json").write_text(
104 json.dumps({"repo_id": fake_id("repo"), "domain": "code"}),
105 encoding="utf-8",
106 )
107 (muse / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
108 return tmp_path
109
110
111 def _write_object(root: pathlib.Path, content: bytes) -> str:
112 oid = blob_id(content)
113 _write_obj_atomic(root, oid, content)
114 return oid
115
116
117 def _write_snapshot_with_objects(
118 root: pathlib.Path, files: _FileStore
119 ) -> tuple[str, dict[str, str]]:
120 """Write objects + snapshot. Returns (snapshot_id, manifest)."""
121 manifest: Manifest = {}
122 for name, content in files.items():
123 manifest[name] = _write_object(root, content)
124 snap_id = compute_snapshot_id(manifest)
125 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
126 return snap_id, manifest
127
128
129 def _write_commit_record(
130 root: pathlib.Path,
131 snapshot_id: str,
132 *,
133 parent1: str | None = None,
134 parent2: str | None = None,
135 message: str = "test",
136 ts_offset: int = 0,
137 ) -> str:
138 """Write a commit msgpack and return its commit_id."""
139 parent_ids = [p for p in [parent1, parent2] if p]
140 ts = (_EPOCH + datetime.timedelta(seconds=ts_offset)).isoformat()
141 commit_id = compute_commit_id(
142 repo_id="test-repo",
143 parent_ids=parent_ids,
144 snapshot_id=snapshot_id,
145 message=message,
146 committed_at_iso=ts,
147 )
148 data = {
149 "commit_id": commit_id,
150 "repo_id": "test-repo",
151 "created_on_branch": "main",
152 "snapshot_id": snapshot_id,
153 "message": message,
154 "committed_at": ts,
155 "parent_commit_id": parent1,
156 "parent2_commit_id": parent2,
157 "author": "test",
158 "metadata": {},
159 }
160 p = commit_path(root, commit_id)
161 p.parent.mkdir(parents=True, exist_ok=True)
162 p.write_bytes(msgpack.packb(data, use_bin_type=True))
163 return commit_id
164
165
166 def _set_branch(root: pathlib.Path, branch: str, commit_id: str) -> None:
167 ref_path = root / ".muse" / "refs" / "heads" / branch
168 ref_path.parent.mkdir(parents=True, exist_ok=True)
169 ref_path.write_text(commit_id, encoding="utf-8")
170
171
172 def _make_linear_chain(
173 root: pathlib.Path,
174 length: int,
175 branch: str = "main",
176 ) -> list[str]:
177 """Create a linear chain of *length* commits on *branch*. Returns all commit IDs."""
178 snap_id, _ = _write_snapshot_with_objects(root, {f"f{i}.txt": f"v{i}".encode() for i in range(length)})
179 commit_ids: list[str] = []
180 parent: str | None = None
181 for i in range(length):
182 cid = _write_commit_record(root, snap_id, parent1=parent, message=f"commit {i}", ts_offset=i)
183 commit_ids.append(cid)
184 parent = cid
185 _set_branch(root, branch, commit_ids[-1])
186 return commit_ids
187
188
189 def _env(root: pathlib.Path) -> Manifest:
190 return {"MUSE_REPO_ROOT": str(root)}
191
192
193 def _invoke_gc(root: pathlib.Path, *extra_args: str) -> InvokeResult:
194 args = list(extra_args)
195 if "--grace-period" not in args:
196 args = ["--grace-period", "0"] + args
197 return runner.invoke(cli, ["gc"] + args, env=_env(root), catch_exceptions=False)
198
199
200 # ---------------------------------------------------------------------------
201 # Unit — GcResult new fields default to zero
202 # ---------------------------------------------------------------------------
203
204
205 class TestGcResultDefaults:
206 def test_commits_fields_default_to_zero(self) -> None:
207 r = GcResult()
208 assert r.commits_reachable == 0
209 assert r.commits_collected == 0
210 assert r.commits_collected_bytes == 0
211
212 def test_snapshots_fields_default_to_zero(self) -> None:
213 r = GcResult()
214 assert r.snapshots_reachable == 0
215 assert r.snapshots_collected == 0
216 assert r.snapshots_collected_bytes == 0
217
218 def test_full_field_defaults_to_false(self) -> None:
219 assert GcResult().full is False
220
221
222 # ---------------------------------------------------------------------------
223 # Unit — _collect_reachable_commits
224 # ---------------------------------------------------------------------------
225
226
227 class TestCollectReachableCommits:
228 def test_empty_repo_returns_empty_set(self, tmp_path: pathlib.Path) -> None:
229 root = _make_repo(tmp_path)
230 assert _collect_reachable_commits(root) == set()
231
232 def test_single_branch_single_commit(self, tmp_path: pathlib.Path) -> None:
233 root = _make_repo(tmp_path)
234 snap_id, _ = _write_snapshot_with_objects(root, {"a.py": b"x"})
235 cid = _write_commit_record(root, snap_id)
236 _set_branch(root, "main", cid)
237 assert _collect_reachable_commits(root) == {split_id(cid)[1]}
238
239 def test_traverses_parent_chain(self, tmp_path: pathlib.Path) -> None:
240 root = _make_repo(tmp_path)
241 snap_id, _ = _write_snapshot_with_objects(root, {})
242 c1 = _write_commit_record(root, snap_id, message="c1", ts_offset=0)
243 c2 = _write_commit_record(root, snap_id, parent1=c1, message="c2", ts_offset=1)
244 c3 = _write_commit_record(root, snap_id, parent1=c2, message="c3", ts_offset=2)
245 _set_branch(root, "main", c3)
246 reachable = _collect_reachable_commits(root)
247 assert {split_id(c1)[1], split_id(c2)[1], split_id(c3)[1]} == reachable
248
249 def test_merge_commit_both_parents_reachable(self, tmp_path: pathlib.Path) -> None:
250 root = _make_repo(tmp_path)
251 snap_id, _ = _write_snapshot_with_objects(root, {})
252 base = _write_commit_record(root, snap_id, message="base", ts_offset=0)
253 feat = _write_commit_record(root, snap_id, parent1=base, message="feat", ts_offset=1)
254 merge = _write_commit_record(root, snap_id, parent1=base, parent2=feat, message="merge", ts_offset=2)
255 _set_branch(root, "main", merge)
256 reachable = _collect_reachable_commits(root)
257 assert {split_id(base)[1], split_id(feat)[1], split_id(merge)[1]} == reachable
258
259 def test_multiple_branches_union(self, tmp_path: pathlib.Path) -> None:
260 root = _make_repo(tmp_path)
261 snap_id, _ = _write_snapshot_with_objects(root, {})
262 c1 = _write_commit_record(root, snap_id, message="c1", ts_offset=0)
263 c2 = _write_commit_record(root, snap_id, message="c2", ts_offset=1)
264 _set_branch(root, "main", c1)
265 _set_branch(root, "dev", c2)
266 reachable = _collect_reachable_commits(root)
267 assert {split_id(c1)[1], split_id(c2)[1]} == reachable
268
269 def test_nested_branch_ref_traversed(self, tmp_path: pathlib.Path) -> None:
270 root = _make_repo(tmp_path)
271 snap_id, _ = _write_snapshot_with_objects(root, {})
272 cid = _write_commit_record(root, snap_id)
273 _set_branch(root, "feat/my-feature", cid)
274 reachable = _collect_reachable_commits(root)
275 assert split_id(cid)[1] in reachable
276
277 def test_missing_commit_file_skipped(self, tmp_path: pathlib.Path) -> None:
278 root = _make_repo(tmp_path)
279 # Write a branch ref pointing to a commit that doesn't exist in store.
280 ghost_id = "a" * 64
281 _set_branch(root, "main", ghost_id)
282 # Should not raise; ghost_id counted as reachable (it's the tip).
283 reachable = _collect_reachable_commits(root)
284 assert ghost_id in reachable
285
286 def test_corrupt_commit_file_skipped_gracefully(self, tmp_path: pathlib.Path) -> None:
287 root = _make_repo(tmp_path)
288 snap_id, _ = _write_snapshot_with_objects(root, {})
289 cid = _write_commit_record(root, snap_id)
290 _set_branch(root, "main", cid)
291 # Corrupt the commit file.
292 commit_path(root, cid).write_bytes(b"not msgpack")
293 # Should not raise.
294 reachable = _collect_reachable_commits(root)
295 assert split_id(cid)[1] in reachable # tip is still reachable; parents can't be walked
296
297 def test_symlinked_ref_file_skipped(self, tmp_path: pathlib.Path) -> None:
298 root = _make_repo(tmp_path)
299 # Create a symlinked ref file — should be ignored.
300 ref_dir = root / ".muse" / "refs" / "heads"
301 link = ref_dir / "evil"
302 target = tmp_path / "target.txt"
303 target.write_text("a" * 64)
304 link.symlink_to(target)
305 # Shouldn't crash; symlinked ref is not followed.
306 reachable = _collect_reachable_commits(root)
307 assert len(reachable) == 0
308
309 def test_commit_only_in_orphaned_file_not_reachable(self, tmp_path: pathlib.Path) -> None:
310 root = _make_repo(tmp_path)
311 snap_id, _ = _write_snapshot_with_objects(root, {})
312 orphan = _write_commit_record(root, snap_id, message="orphan")
313 # No branch ref points to orphan.
314 reachable = _collect_reachable_commits(root)
315 assert split_id(orphan)[1] not in reachable
316
317 def test_diamond_dag_no_duplicate_walk(self, tmp_path: pathlib.Path) -> None:
318 root = _make_repo(tmp_path)
319 snap_id, _ = _write_snapshot_with_objects(root, {})
320 base = _write_commit_record(root, snap_id, message="base", ts_offset=0)
321 left = _write_commit_record(root, snap_id, parent1=base, message="left", ts_offset=1)
322 right = _write_commit_record(root, snap_id, parent1=base, message="right", ts_offset=2)
323 tip = _write_commit_record(root, snap_id, parent1=left, parent2=right, message="tip", ts_offset=3)
324 _set_branch(root, "main", tip)
325 reachable = _collect_reachable_commits(root)
326 assert reachable == {split_id(base)[1], split_id(left)[1], split_id(right)[1], split_id(tip)[1]}
327
328
329 # ---------------------------------------------------------------------------
330 # Unit — _collect_reachable_snapshots
331 # ---------------------------------------------------------------------------
332
333
334 class TestCollectReachableSnapshots:
335 def test_returns_snapshot_ids_from_reachable_commits(self, tmp_path: pathlib.Path) -> None:
336 root = _make_repo(tmp_path)
337 snap_id, manifest = _write_snapshot_with_objects(root, {"f.py": b"code"})
338 cid = _write_commit_record(root, snap_id)
339 snaps, objs = _collect_reachable_snapshots(root, {split_id(cid)[1]})
340 assert split_id(snap_id)[1] in snaps
341
342 def test_returns_blob_ids_from_manifest(self, tmp_path: pathlib.Path) -> None:
343 root = _make_repo(tmp_path)
344 obj_id = _write_object(root, b"file content")
345 snap_id = compute_snapshot_id({"f.py": obj_id})
346 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest={"f.py": obj_id}))
347 cid = _write_commit_record(root, snap_id)
348 _, objs = _collect_reachable_snapshots(root, {split_id(cid)[1]})
349 assert obj_id in objs
350
351 def test_empty_reachable_commits_returns_empty(self, tmp_path: pathlib.Path) -> None:
352 root = _make_repo(tmp_path)
353 snaps, objs = _collect_reachable_snapshots(root, set())
354 assert snaps == set()
355 assert objs == set()
356
357 def test_multiple_commits_same_snapshot_deduplicated(self, tmp_path: pathlib.Path) -> None:
358 root = _make_repo(tmp_path)
359 snap_id, _ = _write_snapshot_with_objects(root, {"f": b"x"})
360 c1 = _write_commit_record(root, snap_id, message="c1", ts_offset=0)
361 c2 = _write_commit_record(root, snap_id, message="c2", ts_offset=1)
362 snaps, _ = _collect_reachable_snapshots(root, {split_id(c1)[1], split_id(c2)[1]})
363 assert snaps == {split_id(snap_id)[1]}
364
365 def test_shelf_blobs_included(self, tmp_path: pathlib.Path) -> None:
366 root = _make_repo(tmp_path)
367 shelf_obj = _write_object(root, b"shelved content")
368 shelf_path = root / ".muse" / "shelf.json"
369 shelf_path.write_text(json.dumps([{
370 "snapshot_id": "s" * 64,
371 "branch": "main",
372 "created_at": "2026-01-01T00:00:00+00:00",
373 "snapshot": {"file.py": shelf_obj},
374 }]))
375 _, objs = _collect_reachable_snapshots(root, set())
376 assert shelf_obj in objs
377
378 def test_missing_snapshot_file_skipped(self, tmp_path: pathlib.Path) -> None:
379 root = _make_repo(tmp_path)
380 ghost_snap_id = "b" * 64
381 cid = _write_commit_record(root, ghost_snap_id)
382 # No snapshot file on disk — should not crash.
383 snaps, objs = _collect_reachable_snapshots(root, {split_id(cid)[1]})
384 # ghost snapshot is in snaps set but its blobs can't be collected
385 assert ghost_snap_id in snaps
386 assert len(objs) == 0
387
388
389 # ---------------------------------------------------------------------------
390 # Unit — _list_stored_msgpack
391 # ---------------------------------------------------------------------------
392
393
394 class TestListStoredMsgpack:
395 def test_returns_msgpack_files(self, tmp_path: pathlib.Path) -> None:
396 d = tmp_path / "store"
397 shard = d / "sha256"
398 shard.mkdir(parents=True)
399 (shard / "abc123.msgpack").write_bytes(b"data")
400 (shard / "def456.msgpack").write_bytes(b"data2")
401 pairs = _list_stored_msgpack(d, grace_period_seconds=0)
402 stems = {stem for stem, _ in pairs}
403 assert stems == {"abc123", "def456"}
404
405 def test_non_msgpack_files_excluded(self, tmp_path: pathlib.Path) -> None:
406 d = tmp_path / "store"
407 d.mkdir()
408 (d / "abc.json").write_bytes(b"data")
409 (d / "abc.txt").write_bytes(b"data")
410 pairs = _list_stored_msgpack(d, grace_period_seconds=0)
411 assert pairs == []
412
413 def test_symlinked_file_excluded(self, tmp_path: pathlib.Path) -> None:
414 d = tmp_path / "store"
415 d.mkdir()
416 real = tmp_path / "real.msgpack"
417 real.write_bytes(b"data")
418 (d / "linked.msgpack").symlink_to(real)
419 pairs = _list_stored_msgpack(d, grace_period_seconds=0)
420 assert pairs == []
421
422 def test_grace_period_protects_recent_files(self, tmp_path: pathlib.Path) -> None:
423 d = tmp_path / "store"
424 shard = d / "sha256"
425 shard.mkdir(parents=True)
426 (shard / "recent.msgpack").write_bytes(b"data")
427 pairs = _list_stored_msgpack(d, grace_period_seconds=9999)
428 assert pairs == []
429
430 def test_grace_period_zero_includes_all(self, tmp_path: pathlib.Path) -> None:
431 d = tmp_path / "store"
432 shard = d / "sha256"
433 shard.mkdir(parents=True)
434 (shard / "old.msgpack").write_bytes(b"data")
435 pairs = _list_stored_msgpack(d, grace_period_seconds=0)
436 assert len(pairs) == 1
437
438 def test_nonexistent_directory_returns_empty(self, tmp_path: pathlib.Path) -> None:
439 pairs = _list_stored_msgpack(tmp_path / "does_not_exist", grace_period_seconds=0)
440 assert pairs == []
441
442
443 # ---------------------------------------------------------------------------
444 # Integration — run_gc(full=True)
445 # ---------------------------------------------------------------------------
446
447
448 class TestRunGcFull:
449 def test_orphaned_commit_deleted(self, tmp_path: pathlib.Path) -> None:
450 root = _make_repo(tmp_path)
451 snap_id, _ = _write_snapshot_with_objects(root, {})
452 orphan = _write_commit_record(root, snap_id, message="orphan")
453 orphan_path = commit_path(root, orphan)
454 assert orphan_path.exists()
455
456 result = run_gc(root, full=True, grace_period_seconds=0)
457 assert result.commits_collected == 1
458 assert not orphan_path.exists()
459
460 def test_reachable_commit_preserved(self, tmp_path: pathlib.Path) -> None:
461 root = _make_repo(tmp_path)
462 snap_id, _ = _write_snapshot_with_objects(root, {})
463 cid = _write_commit_record(root, snap_id)
464 _set_branch(root, "main", cid)
465 cp = commit_path(root, cid)
466
467 result = run_gc(root, full=True, grace_period_seconds=0)
468 assert result.commits_collected == 0
469 assert cp.exists()
470
471 def test_orphaned_snapshot_deleted(self, tmp_path: pathlib.Path) -> None:
472 root = _make_repo(tmp_path)
473 snap_id, _ = _write_snapshot_with_objects(root, {})
474 snap_path = snapshot_path(root, snap_id)
475 assert snap_path.exists()
476 # No commit references this snapshot.
477
478 result = run_gc(root, full=True, grace_period_seconds=0)
479 assert result.snapshots_collected == 1
480 assert not snap_path.exists()
481
482 def test_reachable_snapshot_preserved(self, tmp_path: pathlib.Path) -> None:
483 root = _make_repo(tmp_path)
484 snap_id, _ = _write_snapshot_with_objects(root, {"f.py": b"code"})
485 cid = _write_commit_record(root, snap_id)
486 _set_branch(root, "main", cid)
487 snap_path = snapshot_path(root, snap_id)
488
489 result = run_gc(root, full=True, grace_period_seconds=0)
490 assert result.snapshots_collected == 0
491 assert snap_path.exists()
492
493 def test_orphaned_blob_from_orphaned_commit_deleted(self, tmp_path: pathlib.Path) -> None:
494 root = _make_repo(tmp_path)
495 orphan_blob = _write_object(root, b"only in orphaned commit")
496 snap_id = compute_snapshot_id({"f": orphan_blob})
497 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest={"f": orphan_blob}))
498 _write_commit_record(root, snap_id, message="orphan")
499 # No branch ref → orphan commit + snapshot + blob all unreachable.
500 blob_path = object_path(root, orphan_blob)
501
502 result = run_gc(root, full=True, grace_period_seconds=0)
503 assert result.commits_collected == 1
504 assert result.snapshots_collected == 1
505 assert result.collected_count == 1
506 assert not blob_path.exists()
507
508 def test_dry_run_never_deletes(self, tmp_path: pathlib.Path) -> None:
509 root = _make_repo(tmp_path)
510 snap_id, _ = _write_snapshot_with_objects(root, {})
511 orphan = _write_commit_record(root, snap_id, message="orphan")
512
513 result = run_gc(root, full=True, dry_run=True, grace_period_seconds=0)
514 assert result.dry_run is True
515 assert result.commits_collected == 1
516 assert commit_path(root, orphan).exists()
517 assert snapshot_path(root, snap_id).exists()
518
519 def test_commit_reachable_from_any_branch_preserved(self, tmp_path: pathlib.Path) -> None:
520 root = _make_repo(tmp_path)
521 snap_id, _ = _write_snapshot_with_objects(root, {})
522 shared_base = _write_commit_record(root, snap_id, message="base", ts_offset=0)
523 tip_main = _write_commit_record(root, snap_id, parent1=shared_base, message="main-tip", ts_offset=1)
524 tip_dev = _write_commit_record(root, snap_id, parent1=shared_base, message="dev-tip", ts_offset=2)
525 _set_branch(root, "main", tip_main)
526 _set_branch(root, "dev", tip_dev)
527
528 result = run_gc(root, full=True, grace_period_seconds=0)
529 assert result.commits_collected == 0
530 for cid in (shared_base, tip_main, tip_dev):
531 assert commit_path(root, cid).exists()
532
533 def test_linear_chain_all_intermediates_preserved(self, tmp_path: pathlib.Path) -> None:
534 root = _make_repo(tmp_path)
535 commit_ids = _make_linear_chain(root, 10)
536
537 result = run_gc(root, full=True, grace_period_seconds=0)
538 assert result.commits_collected == 0
539 for cid in commit_ids:
540 assert commit_path(root, cid).exists()
541
542 def test_merge_commit_both_parent_chains_preserved(self, tmp_path: pathlib.Path) -> None:
543 root = _make_repo(tmp_path)
544 snap_id, _ = _write_snapshot_with_objects(root, {})
545 base = _write_commit_record(root, snap_id, message="base", ts_offset=0)
546 left = _write_commit_record(root, snap_id, parent1=base, message="left", ts_offset=1)
547 right = _write_commit_record(root, snap_id, parent1=base, message="right", ts_offset=2)
548 merge = _write_commit_record(root, snap_id, parent1=left, parent2=right, message="merge", ts_offset=3)
549 _set_branch(root, "main", merge)
550
551 result = run_gc(root, full=True, grace_period_seconds=0)
552 assert result.commits_collected == 0
553 for cid in (base, left, right, merge):
554 assert commit_path(root, cid).exists()
555
556 def test_grace_period_protects_recent_commit(self, tmp_path: pathlib.Path) -> None:
557 root = _make_repo(tmp_path)
558 snap_id, _ = _write_snapshot_with_objects(root, {})
559 orphan = _write_commit_record(root, snap_id)
560 # No branch ref; orphan is unreachable — but grace period protects it.
561 result = run_gc(root, full=True, grace_period_seconds=9999)
562 assert result.commits_collected == 0
563 assert commit_path(root, orphan).exists()
564
565 def test_grace_period_protects_recent_snapshot(self, tmp_path: pathlib.Path) -> None:
566 root = _make_repo(tmp_path)
567 snap_id, _ = _write_snapshot_with_objects(root, {})
568 result = run_gc(root, full=True, grace_period_seconds=9999)
569 assert result.snapshots_collected == 0
570 assert snapshot_path(root, snap_id).exists()
571
572 def test_gcresult_commits_reachable_count(self, tmp_path: pathlib.Path) -> None:
573 root = _make_repo(tmp_path)
574 snap_id, _ = _write_snapshot_with_objects(root, {})
575 cid = _write_commit_record(root, snap_id)
576 _set_branch(root, "main", cid)
577
578 result = run_gc(root, full=True, grace_period_seconds=0)
579 assert result.commits_reachable == 1
580 assert result.commits_collected == 0
581
582 def test_gcresult_snapshots_reachable_count(self, tmp_path: pathlib.Path) -> None:
583 root = _make_repo(tmp_path)
584 snap_id, _ = _write_snapshot_with_objects(root, {})
585 cid = _write_commit_record(root, snap_id)
586 _set_branch(root, "main", cid)
587
588 result = run_gc(root, full=True, grace_period_seconds=0)
589 assert result.snapshots_reachable == 1
590 assert result.snapshots_collected == 0
591
592 def test_gcresult_full_flag_set(self, tmp_path: pathlib.Path) -> None:
593 root = _make_repo(tmp_path)
594 result = run_gc(root, full=True, grace_period_seconds=0)
595 assert result.full is True
596
597 def test_without_full_flag_orphaned_commits_not_deleted(self, tmp_path: pathlib.Path) -> None:
598 """Default run_gc (full=False) must NOT prune commits or snapshots."""
599 root = _make_repo(tmp_path)
600 snap_id, _ = _write_snapshot_with_objects(root, {})
601 orphan = _write_commit_record(root, snap_id, message="orphan")
602
603 result = run_gc(root, full=False, grace_period_seconds=0)
604 assert result.commits_collected == 0
605 assert result.snapshots_collected == 0
606 assert commit_path(root, orphan).exists()
607
608 def test_idempotent_second_run_collects_nothing(self, tmp_path: pathlib.Path) -> None:
609 root = _make_repo(tmp_path)
610 snap_id, _ = _write_snapshot_with_objects(root, {})
611 orphan = _write_commit_record(root, snap_id)
612
613 run_gc(root, full=True, grace_period_seconds=0)
614 result2 = run_gc(root, full=True, grace_period_seconds=0)
615 assert result2.commits_collected == 0
616 assert result2.snapshots_collected == 0
617 assert result2.collected_count == 0
618
619 def test_collected_bytes_nonzero_for_deleted_commit(self, tmp_path: pathlib.Path) -> None:
620 root = _make_repo(tmp_path)
621 snap_id, _ = _write_snapshot_with_objects(root, {})
622 _write_commit_record(root, snap_id, message="orphan")
623
624 result = run_gc(root, full=True, grace_period_seconds=0)
625 assert result.commits_collected_bytes > 0
626
627 def test_collected_bytes_nonzero_for_deleted_snapshot(self, tmp_path: pathlib.Path) -> None:
628 root = _make_repo(tmp_path)
629 snap_id, _ = _write_snapshot_with_objects(root, {"f": b"content"})
630
631 result = run_gc(root, full=True, grace_period_seconds=0)
632 assert result.snapshots_collected_bytes > 0
633
634
635 # ---------------------------------------------------------------------------
636 # CLI integration
637 # ---------------------------------------------------------------------------
638
639
640 class TestCliGcFull:
641 def test_full_text_output_has_three_lines(self, tmp_path: pathlib.Path) -> None:
642 root = _make_repo(tmp_path)
643 r = _invoke_gc(root, "--full")
644 assert r.exit_code == 0
645 lines = [ln for ln in r.output.strip().splitlines() if ln.strip()]
646 assert len(lines) == 3 # objects, commits, snapshots
647
648 def test_full_text_output_contains_commit_line(self, tmp_path: pathlib.Path) -> None:
649 root = _make_repo(tmp_path)
650 r = _invoke_gc(root, "--full")
651 assert "commit" in r.output
652
653 def test_full_text_output_contains_snapshot_line(self, tmp_path: pathlib.Path) -> None:
654 root = _make_repo(tmp_path)
655 r = _invoke_gc(root, "--full")
656 assert "snapshot" in r.output
657
658 def test_full_dry_run_prefix_on_all_lines(self, tmp_path: pathlib.Path) -> None:
659 root = _make_repo(tmp_path)
660 r = _invoke_gc(root, "--full", "--dry-run")
661 assert r.exit_code == 0
662 content_lines = [ln for ln in r.output.strip().splitlines() if ln.strip()]
663 assert all("[dry-run]" in ln for ln in content_lines)
664
665 def test_full_json_includes_commits_fields(self, tmp_path: pathlib.Path) -> None:
666 root = _make_repo(tmp_path)
667 r = _invoke_gc(root, "--full", "--json")
668 assert r.exit_code == 0
669 data = json.loads(r.output.strip())
670 assert "commits_reachable" in data
671 assert "commits_collected" in data
672 assert "commits_collected_bytes" in data
673
674 def test_full_json_includes_snapshots_fields(self, tmp_path: pathlib.Path) -> None:
675 root = _make_repo(tmp_path)
676 r = _invoke_gc(root, "--full", "--json")
677 data = json.loads(r.output.strip())
678 assert "snapshots_reachable" in data
679 assert "snapshots_collected" in data
680 assert "snapshots_collected_bytes" in data
681
682 def test_full_json_field_types(self, tmp_path: pathlib.Path) -> None:
683 root = _make_repo(tmp_path)
684 r = _invoke_gc(root, "--full", "--json")
685 data = json.loads(r.output.strip())
686 assert isinstance(data["commits_reachable"], int)
687 assert isinstance(data["commits_collected"], int)
688 assert isinstance(data["commits_collected_bytes"], int)
689 assert isinstance(data["snapshots_reachable"], int)
690 assert isinstance(data["snapshots_collected"], int)
691 assert isinstance(data["snapshots_collected_bytes"], int)
692 assert isinstance(data["full"], bool)
693
694 def test_full_json_full_field_true(self, tmp_path: pathlib.Path) -> None:
695 root = _make_repo(tmp_path)
696 r = _invoke_gc(root, "--full", "--json")
697 data = json.loads(r.output.strip())
698 assert data["full"] is True
699
700 def test_no_full_json_full_field_false(self, tmp_path: pathlib.Path) -> None:
701 root = _make_repo(tmp_path)
702 r = _invoke_gc(root, "--json")
703 data = json.loads(r.output.strip())
704 assert data["full"] is False
705
706 def test_full_json_dry_run_field(self, tmp_path: pathlib.Path) -> None:
707 root = _make_repo(tmp_path)
708 r = _invoke_gc(root, "--full", "--dry-run", "--json")
709 data = json.loads(r.output.strip())
710 assert data["dry_run"] is True
711
712 def test_full_zero_orphans_reports_zeros(self, tmp_path: pathlib.Path) -> None:
713 root = _make_repo(tmp_path)
714 snap_id, _ = _write_snapshot_with_objects(root, {})
715 cid = _write_commit_record(root, snap_id)
716 _set_branch(root, "main", cid)
717 r = _invoke_gc(root, "--full", "--json")
718 data = json.loads(r.output.strip())
719 assert data["commits_collected"] == 0
720 assert data["snapshots_collected"] == 0
721
722 def test_full_reports_correct_collected_counts(self, tmp_path: pathlib.Path) -> None:
723 root = _make_repo(tmp_path)
724 # 3 orphaned commits, 3 orphaned snapshots
725 for i in range(3):
726 snap_id, _ = _write_snapshot_with_objects(root, {f"f{i}": f"v{i}".encode()})
727 _write_commit_record(root, snap_id, message=f"orphan-{i}", ts_offset=i)
728 r = _invoke_gc(root, "--full", "--json")
729 data = json.loads(r.output.strip())
730 assert data["commits_collected"] == 3
731 assert data["snapshots_collected"] == 3
732
733 def test_no_full_json_schema_unchanged(self, tmp_path: pathlib.Path) -> None:
734 """Without --full, the new fields are present but zero."""
735 root = _make_repo(tmp_path)
736 r = _invoke_gc(root, "--json")
737 data = json.loads(r.output.strip())
738 # Old fields still present.
739 assert "collected_count" in data
740 assert "reachable_count" in data
741 # New fields present but zero.
742 assert data["commits_collected"] == 0
743 assert data["snapshots_collected"] == 0
744
745
746 # ---------------------------------------------------------------------------
747 # E2E tests
748 # ---------------------------------------------------------------------------
749
750
751 class TestGcFullE2E:
752 def test_full_lifecycle_orphans_accumulate_then_freed(self, tmp_path: pathlib.Path) -> None:
753 root = _make_repo(tmp_path)
754
755 # Create a live commit on main.
756 live_snap, _ = _write_snapshot_with_objects(root, {"app.py": b"app code"})
757 live_cid = _write_commit_record(root, live_snap)
758 _set_branch(root, "main", live_cid)
759
760 # Simulate abandoned work: write orphaned commits/snapshots.
761 for i in range(5):
762 snap_id, _ = _write_snapshot_with_objects(root, {f"draft{i}.py": f"draft{i}".encode()})
763 _write_commit_record(root, snap_id, message=f"abandoned-{i}", ts_offset=i + 10)
764
765 result = run_gc(root, full=True, grace_period_seconds=0)
766 assert result.commits_collected == 5
767 assert result.snapshots_collected == 5
768 assert result.commits_reachable == 1
769 assert result.snapshots_reachable == 1
770
771 # Live commit and snapshot still intact.
772 assert commit_path(root, live_cid).exists()
773 assert snapshot_path(root, live_snap).exists()
774
775 def test_shelf_blobs_protected_under_full(self, tmp_path: pathlib.Path) -> None:
776 root = _make_repo(tmp_path)
777 shelf_obj = _write_object(root, b"shelved work")
778 shelf_path = root / ".muse" / "shelf.json"
779 shelf_path.write_text(json.dumps([{
780 "snapshot_id": long_id("0b" * 32),
781 "branch": "main",
782 "created_at": "2026-01-01T00:00:00+00:00",
783 "snapshot": {"work.py": shelf_obj},
784 }]))
785
786 result = run_gc(root, full=True, grace_period_seconds=0)
787 assert result.collected_count == 0
788 blob_path = object_path(root, shelf_obj)
789 assert blob_path.exists()
790
791 def test_rewrite_history_old_commits_removed(self, tmp_path: pathlib.Path) -> None:
792 """Simulate a history rewrite: old commits orphaned, new commits on branch."""
793 root = _make_repo(tmp_path)
794 snap1, _ = _write_snapshot_with_objects(root, {"v1.py": b"v1"})
795 old_cid = _write_commit_record(root, snap1, message="old")
796
797 snap2, _ = _write_snapshot_with_objects(root, {"v2.py": b"v2"})
798 new_cid = _write_commit_record(root, snap2, message="new (rewrite)")
799 _set_branch(root, "main", new_cid)
800 # old_cid is now orphaned.
801
802 result = run_gc(root, full=True, grace_period_seconds=0)
803 assert result.commits_collected == 1
804 assert result.snapshots_collected == 1
805 assert not commit_path(root, old_cid).exists()
806 assert commit_path(root, new_cid).exists()
807
808 def test_two_branches_then_one_deleted_unique_commits_freed(
809 self, tmp_path: pathlib.Path
810 ) -> None:
811 root = _make_repo(tmp_path)
812 shared_snap, _ = _write_snapshot_with_objects(root, {"base.py": b"base"})
813 base_cid = _write_commit_record(root, shared_snap, message="base", ts_offset=0)
814
815 feat_snap, _ = _write_snapshot_with_objects(root, {"feat.py": b"feat"})
816 feat_cid = _write_commit_record(root, feat_snap, parent1=base_cid, message="feat", ts_offset=1)
817
818 _set_branch(root, "main", base_cid)
819 _set_branch(root, "dev", feat_cid)
820
821 # "Delete" feat branch by removing its ref.
822 (root / ".muse" / "refs" / "heads" / "dev").unlink()
823
824 result = run_gc(root, full=True, grace_period_seconds=0)
825 # feat_cid unique to dev is now orphaned.
826 assert result.commits_collected == 1
827 assert not commit_path(root, feat_cid).exists()
828 # base_cid still on main is preserved.
829 assert commit_path(root, base_cid).exists()
830
831
832 # ---------------------------------------------------------------------------
833 # Security tests
834 # ---------------------------------------------------------------------------
835
836
837 class TestGcFullSecurity:
838 def test_symlinked_commit_file_not_deleted(self, tmp_path: pathlib.Path) -> None:
839 root = _make_repo(tmp_path)
840 real_file = tmp_path / "real_commit.msgpack"
841 real_file.write_bytes(msgpack.packb({"commit_id": "a" * 64}, use_bin_type=True))
842 link = root / ".muse" / "commits" / "linked.msgpack"
843 link.symlink_to(real_file)
844
845 run_gc(root, full=True, grace_period_seconds=0)
846 assert real_file.exists(), "Target of symlink must not be deleted"
847
848 def test_symlinked_snapshot_file_not_deleted(self, tmp_path: pathlib.Path) -> None:
849 root = _make_repo(tmp_path)
850 real_file = tmp_path / "real_snap.msgpack"
851 real_file.write_bytes(msgpack.packb({"snapshot_id": "b" * 64}, use_bin_type=True))
852 link = root / ".muse" / "snapshots" / "linked.msgpack"
853 link.symlink_to(real_file)
854
855 run_gc(root, full=True, grace_period_seconds=0)
856 assert real_file.exists(), "Target of snapshot symlink must not be deleted"
857
858 def test_non_msgpack_file_in_commits_dir_not_deleted(self, tmp_path: pathlib.Path) -> None:
859 root = _make_repo(tmp_path)
860 stray = root / ".muse" / "commits" / "README.txt"
861 stray.write_text("not a commit")
862
863 run_gc(root, full=True, grace_period_seconds=0)
864 assert stray.exists()
865
866 def test_non_msgpack_file_in_snapshots_dir_not_deleted(self, tmp_path: pathlib.Path) -> None:
867 root = _make_repo(tmp_path)
868 stray = root / ".muse" / "snapshots" / ".DS_Store"
869 stray.write_bytes(b"junk")
870
871 run_gc(root, full=True, grace_period_seconds=0)
872 assert stray.exists()
873
874
875 # ---------------------------------------------------------------------------
876 # Stress tests
877 # ---------------------------------------------------------------------------
878
879
880 class TestGcFullStress:
881 def test_200_orphaned_commits_and_snapshots_collected(self, tmp_path: pathlib.Path) -> None:
882 root = _make_repo(tmp_path)
883 # Live commit that must survive.
884 live_snap, _ = _write_snapshot_with_objects(root, {"live.py": b"live"})
885 live_cid = _write_commit_record(root, live_snap)
886 _set_branch(root, "main", live_cid)
887
888 # 200 orphaned commit+snapshot pairs.
889 for i in range(200):
890 snap_id, _ = _write_snapshot_with_objects(root, {f"f{i}.py": f"v{i}".encode()})
891 _write_commit_record(root, snap_id, message=f"orphan-{i}", ts_offset=i + 1)
892
893 result = run_gc(root, full=True, grace_period_seconds=0)
894 assert result.commits_collected == 200
895 assert result.snapshots_collected == 200
896 assert result.commits_reachable == 1
897 assert result.snapshots_reachable == 1
898 # Live commit and snapshot intact.
899 assert commit_path(root, live_cid).exists()
900 assert snapshot_path(root, live_snap).exists()
901
902 def test_deep_100_commit_chain_all_preserved(self, tmp_path: pathlib.Path) -> None:
903 root = _make_repo(tmp_path)
904 commit_ids = _make_linear_chain(root, 100)
905
906 result = run_gc(root, full=True, grace_period_seconds=0)
907 assert result.commits_collected == 0
908 assert result.commits_reachable == 100
909 for cid in commit_ids:
910 assert commit_path(root, cid).exists()
911
912 def test_50_branches_all_commits_preserved(self, tmp_path: pathlib.Path) -> None:
913 root = _make_repo(tmp_path)
914 snap_id, _ = _write_snapshot_with_objects(root, {})
915 for i in range(50):
916 cid = _write_commit_record(root, snap_id, message=f"branch-{i}", ts_offset=i)
917 _set_branch(root, f"feat/branch-{i:02d}", cid)
918
919 result = run_gc(root, full=True, grace_period_seconds=0)
920 assert result.commits_collected == 0
921 assert result.commits_reachable == 50
File History 3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 139 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 145 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 148 days ago