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