"""Tests for muse/core/gc.py — garbage collection.""" from __future__ import annotations import json import pathlib import pytest from muse.core.gc import GcResult, run_gc from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot from muse.core._types import Manifest # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: """Create a minimal .muse repo structure.""" muse = tmp_path / ".muse" for d in ("objects", "commits", "snapshots", "refs/heads"): (muse / d).mkdir(parents=True, exist_ok=True) (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo"})) (muse / "HEAD").write_text("ref: refs/heads/main\n") return tmp_path def _write_object(repo: pathlib.Path, content: bytes) -> str: import hashlib sha = hashlib.sha256(content).hexdigest() obj_dir = repo / ".muse" / "objects" / sha[:2] obj_dir.mkdir(parents=True, exist_ok=True) (obj_dir / sha[2:]).write_bytes(content) return sha def _write_snapshot(repo: pathlib.Path, manifest: Manifest) -> str: """Write a snapshot with a valid content-hash snapshot_id. Returns the snapshot_id.""" snap_id = compute_snapshot_id(manifest) write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) return snap_id def _write_commit(repo: pathlib.Path, snapshot_id: str) -> str: """Write a commit record with a valid content-hash commit_id. Returns the commit_id.""" import datetime committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) commit_id = compute_commit_id([], snapshot_id, "test", committed_at.isoformat()) write_commit(repo, CommitRecord( commit_id=commit_id, repo_id="test-repo", branch="main", snapshot_id=snapshot_id, message="test", committed_at=committed_at, )) ref_path = repo / ".muse" / "refs" / "heads" / "main" ref_path.parent.mkdir(parents=True, exist_ok=True) ref_path.write_text(commit_id) return commit_id # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- def test_gc_empty_repo(tmp_path: pathlib.Path) -> None: """GC on an empty repo should report 0 collected.""" repo = _make_repo(tmp_path) result = run_gc(repo, grace_period_seconds=0) assert isinstance(result, GcResult) assert result.collected_count == 0 def test_gc_removes_unreachable_object(tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) # Write an object but don't reference it in any commit. orphan_id = _write_object(repo, b"orphan data") obj_path = repo / ".muse" / "objects" / orphan_id[:2] / orphan_id[2:] assert obj_path.exists() result = run_gc(repo, grace_period_seconds=0) assert result.collected_count == 1 assert orphan_id in result.collected_ids assert not obj_path.exists() def test_gc_preserves_reachable_object(tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) content = b"reachable file content" obj_id = _write_object(repo, content) snap_id = _write_snapshot(repo, {"file.txt": obj_id}) _write_commit(repo, snap_id) result = run_gc(repo, grace_period_seconds=0) assert result.collected_count == 0 obj_path = repo / ".muse" / "objects" / obj_id[:2] / obj_id[2:] assert obj_path.exists() def test_gc_dry_run_does_not_delete(tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) orphan_id = _write_object(repo, b"orphan") obj_path = repo / ".muse" / "objects" / orphan_id[:2] / orphan_id[2:] result = run_gc(repo, dry_run=True, grace_period_seconds=0) assert result.dry_run is True assert result.collected_count == 1 # File should still exist. assert obj_path.exists() def test_gc_collected_bytes(tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) content = b"x" * 1000 _write_object(repo, content) result = run_gc(repo, grace_period_seconds=0) assert result.collected_bytes >= 1000 def test_gc_multiple_orphans(tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) for i in range(5): _write_object(repo, f"orphan {i}".encode()) result = run_gc(repo, grace_period_seconds=0) assert result.collected_count == 5 def test_gc_mixed_reachable_and_orphans(tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) # One reachable object. reachable_id = _write_object(repo, b"reachable") snap_id = _write_snapshot(repo, {"file.txt": reachable_id}) _write_commit(repo, snap_id) # Two orphans. _write_object(repo, b"orphan A") _write_object(repo, b"orphan B") result = run_gc(repo, grace_period_seconds=0) assert result.collected_count == 2 assert result.reachable_count == 1 def test_gc_elapsed_time_positive(tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = run_gc(repo, grace_period_seconds=0) assert result.elapsed_seconds >= 0.0 # --------------------------------------------------------------------------- # Stress test # --------------------------------------------------------------------------- def test_gc_preserves_stash_objects(tmp_path: pathlib.Path) -> None: """Objects referenced only by stash.json must NOT be GCed. This is the critical safety case: `muse stash` writes file blobs to the object store and records their IDs in stash.json. Without walking the stash, a subsequent `muse gc` would delete those blobs and make `muse stash pop` fail with missing objects. """ repo = _make_repo(tmp_path) # Simulate stash writing two objects. stash_obj_a = _write_object(repo, b"stashed file A") stash_obj_b = _write_object(repo, b"stashed file B") stash_path = repo / ".muse" / "stash.json" stash_path.write_text(json.dumps([{ "snapshot_id": "s" * 64, "branch": "main", "stashed_at": "2026-01-01T00:00:00+00:00", "manifest": {"a.py": stash_obj_a, "b.py": stash_obj_b}, }])) result = run_gc(repo, grace_period_seconds=0) assert result.collected_count == 0, "Stash objects must not be GCed" # The blobs must still exist. assert (repo / ".muse" / "objects" / stash_obj_a[:2] / stash_obj_a[2:]).exists() assert (repo / ".muse" / "objects" / stash_obj_b[:2] / stash_obj_b[2:]).exists() def test_gc_collects_objects_not_in_stash(tmp_path: pathlib.Path) -> None: """Objects that are neither committed nor stashed ARE unreachable and must be GCed.""" repo = _make_repo(tmp_path) stash_obj = _write_object(repo, b"stashed") orphan_obj = _write_object(repo, b"truly orphaned") stash_path = repo / ".muse" / "stash.json" stash_path.write_text(json.dumps([{ "snapshot_id": "s" * 64, "branch": "main", "stashed_at": "2026-01-01T00:00:00+00:00", "manifest": {"a.py": stash_obj}, }])) result = run_gc(repo, grace_period_seconds=0) assert result.collected_count == 1 assert orphan_obj in result.collected_ids assert stash_obj not in result.collected_ids def test_gc_ignores_stray_non_hex_files_in_objects_dir(tmp_path: pathlib.Path) -> None: """Non-hex filenames in .muse/objects/ are skipped, not mistakenly deleted.""" repo = _make_repo(tmp_path) # Create a stray file that should be ignored. stray_dir = repo / ".muse" / "objects" / "ab" stray_dir.mkdir(parents=True, exist_ok=True) stray = stray_dir / ".DS_Store" stray.write_bytes(b"stray") result = run_gc(repo, grace_period_seconds=0) assert result.collected_count == 0 assert stray.exists(), ".DS_Store should survive GC" def test_gc_stress_many_orphans(tmp_path: pathlib.Path) -> None: """GC should handle 200 orphaned objects efficiently.""" repo = _make_repo(tmp_path) for i in range(200): _write_object(repo, f"orphan-{i:04d}".encode()) result = run_gc(repo, grace_period_seconds=0) assert result.collected_count == 200 # Verify the objects directory is clean. obj_dir = repo / ".muse" / "objects" remaining = list(obj_dir.rglob("*")) remaining_files = [p for p in remaining if p.is_file()] assert remaining_files == []