gabriel / muse public
test_cmd_prune.py python
357 lines 13.0 KB
Raw
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ breaking 146 days ago
1 """Tests for ``muse prune`` — surgical removal of unreachable objects.
2
3 Coverage tiers:
4 - Unit: _collect_all_reachable_ids, _find_prune_candidates helpers
5 - Integration: dry-run lists candidates without deleting, prune removes
6 unreachable objects, --json schema, object count decreases,
7 reachable objects never deleted, --expire filters by age
8 - End-to-end: full CLI via CliRunner
9 - Security: only deletes under .muse/objects/; reachable objects safe;
10 no mutation in --dry-run
11 - Stress: 200-object store with 50% unreachable
12 """
13
14 from __future__ import annotations
15
16 import datetime
17 import hashlib
18 import json
19 import pathlib
20 import time
21
22 import pytest
23
24 from tests.cli_test_helper import CliRunner
25 from muse.core.object_store import write_object, has_object
26 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
27 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
28 from muse.core._types import Manifest
29
30 runner = CliRunner()
31
32 _REPO_ID = "prune-test"
33 _counter = 0
34
35
36 # ---------------------------------------------------------------------------
37 # Helpers
38 # ---------------------------------------------------------------------------
39
40
41 def _sha(data: bytes) -> str:
42 return hashlib.sha256(data).hexdigest()
43
44
45 def _init_repo(path: pathlib.Path) -> pathlib.Path:
46 muse = path / ".muse"
47 for d in ("commits", "snapshots", "objects", "refs/heads", "code"):
48 (muse / d).mkdir(parents=True, exist_ok=True)
49 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
50 (muse / "repo.json").write_text(
51 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
52 )
53 return path
54
55
56 def _env(repo: pathlib.Path) -> dict[str, str]:
57 return {"MUSE_REPO_ROOT": str(repo)}
58
59
60 def _commit_files(
61 root: pathlib.Path,
62 files: dict[str, bytes],
63 branch: str = "main",
64 ) -> str:
65 global _counter
66 _counter += 1
67 manifest: Manifest = {}
68 for rel_path, content in files.items():
69 obj_id = _sha(content)
70 write_object(root, obj_id, content)
71 manifest[rel_path] = obj_id
72 abs_path = root / rel_path
73 abs_path.parent.mkdir(parents=True, exist_ok=True)
74 abs_path.write_bytes(content)
75 snap_id = compute_snapshot_id(manifest)
76 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
77 committed_at = datetime.datetime.now(datetime.timezone.utc)
78 ref_path = root / ".muse" / "refs" / "heads" / branch
79 parent_id = ref_path.read_text(encoding="utf-8").strip() if ref_path.exists() else None
80 parents = [parent_id] if parent_id else []
81 commit_id = compute_commit_id(
82 parents, snap_id, f"commit {_counter}", committed_at.isoformat()
83 )
84 write_commit(
85 root,
86 CommitRecord(
87 commit_id=commit_id,
88 repo_id=_REPO_ID,
89 branch=branch,
90 snapshot_id=snap_id,
91 message=f"commit {_counter}",
92 committed_at=committed_at,
93 parent_commit_id=parent_id,
94 ),
95 )
96 ref_path.write_text(commit_id, encoding="utf-8")
97 return commit_id
98
99
100 def _invoke(repo: pathlib.Path, *args: str):
101 from muse.cli.app import main as cli
102 return runner.invoke(cli, ["prune", *args], env=_env(repo))
103
104
105 def _object_count(root: pathlib.Path) -> int:
106 """Count loose objects directly in the store."""
107 store = root / ".muse" / "objects"
108 if not store.exists():
109 return 0
110 count = 0
111 for shard in store.iterdir():
112 if shard.is_dir():
113 count += sum(1 for f in shard.iterdir() if f.is_file())
114 return count
115
116
117 # ---------------------------------------------------------------------------
118 # Unit — _collect_all_reachable_ids
119 # ---------------------------------------------------------------------------
120
121
122 def test_collect_all_reachable_ids_empty_repo(tmp_path: pathlib.Path) -> None:
123 from muse.cli.commands.prune import _collect_all_reachable_ids
124 root = _init_repo(tmp_path)
125 ids = _collect_all_reachable_ids(root)
126 assert isinstance(ids, set)
127
128
129 def test_collect_all_reachable_ids_after_commit(tmp_path: pathlib.Path) -> None:
130 from muse.cli.commands.prune import _collect_all_reachable_ids
131 root = _init_repo(tmp_path)
132 _commit_files(root, {"a.py": b"# a\n", "b.py": b"# b\n"})
133 ids = _collect_all_reachable_ids(root)
134 assert _sha(b"# a\n") in ids
135 assert _sha(b"# b\n") in ids
136
137
138 def test_collect_all_reachable_ids_excludes_orphan(tmp_path: pathlib.Path) -> None:
139 from muse.cli.commands.prune import _collect_all_reachable_ids
140 root = _init_repo(tmp_path)
141 _commit_files(root, {"a.py": b"# a\n"})
142 orphan = b"orphan not in any snapshot"
143 write_object(root, _sha(orphan), orphan)
144 ids = _collect_all_reachable_ids(root)
145 assert _sha(orphan) not in ids
146 assert _sha(b"# a\n") in ids
147
148
149 # ---------------------------------------------------------------------------
150 # Unit — _find_prune_candidates
151 # ---------------------------------------------------------------------------
152
153
154 def test_find_prune_candidates_returns_orphan(tmp_path: pathlib.Path) -> None:
155 from muse.cli.commands.prune import _find_prune_candidates, _collect_all_reachable_ids
156 root = _init_repo(tmp_path)
157 _commit_files(root, {"a.py": b"# a\n"})
158 orphan = b"i am orphaned"
159 orphan_id = _sha(orphan)
160 write_object(root, orphan_id, orphan)
161 reachable = _collect_all_reachable_ids(root)
162 candidates = _find_prune_candidates(root, reachable, expire_before=None)
163 candidate_ids = {c["object_id"] for c in candidates}
164 assert orphan_id in candidate_ids
165
166
167 def test_find_prune_candidates_excludes_reachable(tmp_path: pathlib.Path) -> None:
168 from muse.cli.commands.prune import _find_prune_candidates, _collect_all_reachable_ids
169 root = _init_repo(tmp_path)
170 _commit_files(root, {"a.py": b"# a\n"})
171 reachable = _collect_all_reachable_ids(root)
172 candidates = _find_prune_candidates(root, reachable, expire_before=None)
173 candidate_ids = {c["object_id"] for c in candidates}
174 assert _sha(b"# a\n") not in candidate_ids
175
176
177 def test_find_prune_candidates_expire_before_filters_recent(tmp_path: pathlib.Path) -> None:
178 from muse.cli.commands.prune import _find_prune_candidates, _collect_all_reachable_ids
179 root = _init_repo(tmp_path)
180 orphan = b"recent orphan"
181 orphan_id = _sha(orphan)
182 write_object(root, orphan_id, orphan)
183 reachable = set()
184 # expire_before = 1 hour AGO — a just-written file is newer, so it should be skipped
185 one_hour_ago = time.time() - 3600
186 candidates = _find_prune_candidates(root, reachable, expire_before=one_hour_ago)
187 candidate_ids = {c["object_id"] for c in candidates}
188 assert orphan_id not in candidate_ids, "Recent orphan should be kept by --expire"
189
190
191 def test_find_prune_candidates_expire_old_objects(tmp_path: pathlib.Path) -> None:
192 from muse.cli.commands.prune import _find_prune_candidates
193 root = _init_repo(tmp_path)
194 orphan = b"old orphan"
195 orphan_id = _sha(orphan)
196 write_object(root, orphan_id, orphan)
197 # Backdate the file's mtime to 2 hours ago
198 obj_path = next(
199 (root / ".muse" / "objects").rglob(orphan_id[-62:]), None
200 )
201 if obj_path:
202 two_hours_ago = time.time() - 7200
203 import os
204 os.utime(obj_path, (two_hours_ago, two_hours_ago))
205 # expire_before = 1 hour ago — the file is older, so it IS a candidate
206 one_hour_ago = time.time() - 3600
207 candidates = _find_prune_candidates(root, set(), expire_before=one_hour_ago)
208 candidate_ids = {c["object_id"] for c in candidates}
209 assert orphan_id in candidate_ids
210
211
212 # ---------------------------------------------------------------------------
213 # Integration — --dry-run
214 # ---------------------------------------------------------------------------
215
216
217 def test_prune_dry_run_does_not_delete_objects(tmp_path: pathlib.Path) -> None:
218 root = _init_repo(tmp_path)
219 _commit_files(root, {"a.py": b"# a\n"})
220 orphan = b"orphan to not delete"
221 orphan_id = _sha(orphan)
222 write_object(root, orphan_id, orphan)
223 before = _object_count(root)
224 result = _invoke(root, "--dry-run")
225 assert result.exit_code == 0
226 after = _object_count(root)
227 assert after == before, "dry-run must not delete any objects"
228
229
230 def test_prune_dry_run_json_lists_candidates(tmp_path: pathlib.Path) -> None:
231 root = _init_repo(tmp_path)
232 _commit_files(root, {"a.py": b"# a\n"})
233 orphan = b"orphan candidate"
234 orphan_id = _sha(orphan)
235 write_object(root, orphan_id, orphan)
236 result = _invoke(root, "--dry-run", "--json")
237 assert result.exit_code == 0
238 data = json.loads(result.stdout)
239 assert "candidates" in data
240 assert data["dry_run"] is True
241 candidate_ids = [c["object_id"] for c in data["candidates"]]
242 assert orphan_id in candidate_ids
243
244
245 def test_prune_dry_run_text_mentions_candidates(tmp_path: pathlib.Path) -> None:
246 root = _init_repo(tmp_path)
247 _commit_files(root, {"a.py": b"# a\n"})
248 write_object(root, _sha(b"orphan x"), b"orphan x")
249 result = _invoke(root, "--dry-run")
250 assert result.exit_code == 0
251 assert result.stdout.strip()
252
253
254 # ---------------------------------------------------------------------------
255 # Integration — actual pruning
256 # ---------------------------------------------------------------------------
257
258
259 def test_prune_removes_unreachable_objects(tmp_path: pathlib.Path) -> None:
260 root = _init_repo(tmp_path)
261 _commit_files(root, {"a.py": b"# a\n"})
262 orphan = b"i am unreachable"
263 orphan_id = _sha(orphan)
264 write_object(root, orphan_id, orphan)
265 assert has_object(root, orphan_id)
266 result = _invoke(root)
267 assert result.exit_code == 0
268 assert not has_object(root, orphan_id), "Orphan blob must be deleted by prune"
269
270
271 def test_prune_keeps_reachable_objects(tmp_path: pathlib.Path) -> None:
272 root = _init_repo(tmp_path)
273 _commit_files(root, {"a.py": b"# a\n", "b.py": b"# b\n"})
274 write_object(root, _sha(b"orphan"), b"orphan")
275 result = _invoke(root)
276 assert result.exit_code == 0
277 assert has_object(root, _sha(b"# a\n")), "Reachable blob must survive prune"
278 assert has_object(root, _sha(b"# b\n")), "Reachable blob must survive prune"
279
280
281 def test_prune_json_schema(tmp_path: pathlib.Path) -> None:
282 root = _init_repo(tmp_path)
283 _commit_files(root, {"a.py": b"# a\n"})
284 write_object(root, _sha(b"orphan"), b"orphan")
285 result = _invoke(root, "--json")
286 assert result.exit_code == 0
287 data = json.loads(result.stdout)
288 assert "pruned" in data
289 assert "bytes_freed" in data
290 assert "dry_run" in data
291 assert data["dry_run"] is False
292
293
294 def test_prune_json_pruned_count(tmp_path: pathlib.Path) -> None:
295 root = _init_repo(tmp_path)
296 _commit_files(root, {"a.py": b"# a\n"})
297 for i in range(3):
298 write_object(root, _sha(f"orphan {i}".encode()), f"orphan {i}".encode())
299 result = _invoke(root, "--json")
300 data = json.loads(result.stdout)
301 assert data["pruned"] >= 3
302
303
304 def test_prune_empty_repo_exits_zero(tmp_path: pathlib.Path) -> None:
305 root = _init_repo(tmp_path)
306 result = _invoke(root, "--json")
307 assert result.exit_code == 0
308 data = json.loads(result.stdout)
309 assert data["pruned"] == 0
310
311
312 def test_prune_no_orphans_exits_zero(tmp_path: pathlib.Path) -> None:
313 root = _init_repo(tmp_path)
314 _commit_files(root, {"a.py": b"# a\n"})
315 result = _invoke(root, "--json")
316 assert result.exit_code == 0
317 data = json.loads(result.stdout)
318 assert data["pruned"] == 0
319
320
321 # ---------------------------------------------------------------------------
322 # Security — only deletes under .muse/objects/
323 # ---------------------------------------------------------------------------
324
325
326 def test_prune_does_not_touch_commits_or_snapshots(tmp_path: pathlib.Path) -> None:
327 root = _init_repo(tmp_path)
328 _commit_files(root, {"a.py": b"# a\n"})
329 write_object(root, _sha(b"orphan"), b"orphan")
330 commits_before = list((root / ".muse" / "commits").glob("*.msgpack"))
331 snaps_before = list((root / ".muse" / "snapshots").glob("*.msgpack"))
332 _invoke(root)
333 commits_after = list((root / ".muse" / "commits").glob("*.msgpack"))
334 snaps_after = list((root / ".muse" / "snapshots").glob("*.msgpack"))
335 assert len(commits_before) == len(commits_after), "prune must not delete commits"
336 assert len(snaps_before) == len(snaps_after), "prune must not delete snapshots"
337
338
339 # ---------------------------------------------------------------------------
340 # Stress
341 # ---------------------------------------------------------------------------
342
343
344 def test_prune_50_percent_unreachable(tmp_path: pathlib.Path) -> None:
345 """200 objects: 100 reachable (committed), 100 orphaned. Prune removes exactly 100."""
346 root = _init_repo(tmp_path)
347 # 100 reachable
348 files = {f"file_{i}.py": f"# {i}\n".encode() for i in range(100)}
349 _commit_files(root, files)
350 # 100 orphans
351 for i in range(100):
352 content = f"orphan blob {i:04d}".encode()
353 write_object(root, _sha(content), content)
354 result = _invoke(root, "--json")
355 assert result.exit_code == 0
356 data = json.loads(result.stdout)
357 assert data["pruned"] == 100
File History 1 commit
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 146 days ago