gabriel / muse public
test_cmd_gc_hardening.py python
705 lines 27.8 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 122 days ago
1 """Comprehensive hardening tests for ``muse gc``.
2
3 Coverage dimensions:
4
5 Unit
6 ~~~~
7 - ``_is_hex`` edge cases (empty string, uppercase, mixed, valid)
8 - ``_list_stored_objects`` symlink guard for prefix dirs
9 - ``_list_stored_objects`` symlink guard for object files
10 - ``_list_stored_objects`` grace period filters recent files
11 - ``_list_stored_objects`` grace_period=0 includes all files
12 - ``_collect_reachable_objects`` symlink guard on shelf.json
13 - ``_collect_reachable_objects`` size cap on shelf.json
14 - ``_collect_reachable_objects`` malformed shelf.json is skipped gracefully
15 - ``run_gc`` grace_period_seconds stored in GcResult
16 - ``_fmt_bytes`` all size ranges
17 - ``run_gc`` negative grace period rejected by CLI
18
19 Security
20 ~~~~~~~~
21 - Symlink in .muse/objects/ prefix dir not deleted or followed
22 - Symlink object file not deleted or followed
23 - Symlink shelf.json skipped during reachability walk
24 - ANSI escape sequences in object IDs sanitized in text output
25 - Invalid --format rejected with error to stderr
26 - Negative --grace-period rejected with non-zero exit
27
28 Integration (CLI)
29 ~~~~~~~~~~~~~~~~~
30 - ``--json`` output schema matches ``_GcJson`` TypedDict
31 - ``--json`` includes ``grace_period_seconds`` field
32 - ``--grace-period`` value propagated to GcResult
33 - ``--dry-run`` combined with ``--json`` reports correctly
34 - ``--verbose`` combined with ``--json`` shows IDs in JSON
35 - ``--format text`` is the default
36 - Repeated GC runs are idempotent (JSON)
37
38 E2E
39 ~~~
40 - Full lifecycle: orphan accumulates across branches, GC reclaims
41 - GC after shelf save does NOT delete shelved objects
42 - GC with corrupt shelf.json succeeds (skips shelf walk)
43 - ``--grace-period 0`` collects freshly-written orphan
44 - ``--grace-period 9999`` protects freshly-written orphan
45
46 Stress
47 ~~~~~~
48 - 500 orphaned objects across 256 prefix dirs collected correctly
49 - Concurrent read-only GC (dry-run) on same repo is safe
50 """
51
52 from __future__ import annotations
53
54 import json
55 import os
56 import pathlib
57 import stat
58 import threading
59 import time
60 from collections.abc import Mapping
61 from typing import TypedDict
62
63 import msgpack
64
65 import pytest
66 from tests.cli_test_helper import CliRunner, InvokeResult
67 from muse.core.types import fake_id, long_id
68 from muse.core.object_store import object_path
69 from muse.core.paths import heads_dir, merge_state_path, muse_dir, objects_dir, shelf_dir
70
71 cli = None # argparse bridge — CliRunner ignores this
72 runner = CliRunner()
73
74
75 # ---------------------------------------------------------------------------
76 # Helpers
77 # ---------------------------------------------------------------------------
78
79
80 def _env(root: pathlib.Path) -> Manifest:
81 return {"MUSE_REPO_ROOT": str(root)}
82
83
84 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
85 muse = muse_dir(tmp_path)
86 for sub in ("objects", "commits", "snapshots", "refs/heads"):
87 (muse / sub).mkdir(parents=True, exist_ok=True)
88 repo_id = fake_id("repo")
89 (muse / "repo.json").write_text(json.dumps({
90 "repo_id": repo_id,
91 "domain": "code",
92 "default_branch": "main",
93 "created_at": "2026-01-01T00:00:00+00:00",
94 }), encoding="utf-8")
95 (muse / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
96 return tmp_path
97
98
99 def _write_object(root: pathlib.Path, content: bytes) -> str:
100 from muse.core.types import blob_id
101 from muse.core.object_store import write_object
102 oid = blob_id(content)
103 write_object(root, oid, content)
104 return oid
105
106
107 def _write_shelf_entry(root: pathlib.Path, snapshot: Mapping[str, str]) -> pathlib.Path:
108 """Write a shelf entry msgpack file under .muse/shelf/sha256/ and return its path."""
109 from muse.core.types import blob_id, split_id
110 entry = {"snapshot": snapshot, "branch": "main", "created_at": "2026-01-01T00:00:00+00:00"}
111 packed = msgpack.packb(entry, use_bin_type=True)
112 _, hex_id = split_id(blob_id(packed))
113 s_dir = shelf_dir(root) / "sha256"
114 s_dir.mkdir(parents=True, exist_ok=True)
115 path = s_dir / f"{hex_id}.msgpack"
116 path.write_bytes(packed)
117 return path
118
119
120 def _make_commit(root: pathlib.Path, manifest: Manifest | None = None) -> str:
121 import datetime
122 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
123 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
124
125 mfst: Manifest = manifest or {}
126 snap_id = compute_snapshot_id(mfst)
127 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
128 commit_id = compute_commit_id( parent_ids=[],
129 snapshot_id=snap_id,
130 message="test",
131 committed_at_iso=committed_at.isoformat(),
132 )
133 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=mfst))
134 write_commit(root, CommitRecord(
135 commit_id=commit_id,
136 repo_id="test-repo",
137 branch="main",
138 snapshot_id=snap_id,
139 message="test",
140 committed_at=committed_at,
141 ))
142 ref_path = heads_dir(root) / "main"
143 ref_path.parent.mkdir(parents=True, exist_ok=True)
144 ref_path.write_text(commit_id, encoding="utf-8")
145 return commit_id
146
147
148 def _invoke_gc(root: pathlib.Path, *extra_args: str) -> InvokeResult:
149 """Invoke ``muse gc`` with ``--grace-period 0`` unless caller overrides."""
150 args = list(extra_args)
151 if "--grace-period" not in args:
152 args = ["--grace-period", "0"] + args
153 return runner.invoke(cli, ["gc"] + args, env=_env(root), catch_exceptions=False)
154
155
156 # ---------------------------------------------------------------------------
157 # _GcJson TypedDict for test assertions
158 # ---------------------------------------------------------------------------
159
160
161 class _GcJson(TypedDict):
162 collected_count: int
163 collected_bytes: int
164 reachable_count: int
165 duration_ms: float
166 grace_period_seconds: int
167 dry_run: bool
168 collected_ids: list[str]
169
170
171 def _parse_gc_json(output: str) -> _GcJson:
172 """Extract and parse the JSON blob from CliRunner output."""
173 for line in output.splitlines():
174 line = line.strip()
175 if line.startswith("{"):
176 raw = json.loads(line)
177 return _GcJson(
178 collected_count=int(raw["collected_count"]),
179 collected_bytes=int(raw["collected_bytes"]),
180 reachable_count=int(raw["reachable_count"]),
181 duration_ms=float(raw["duration_ms"]),
182 grace_period_seconds=int(raw["grace_period_seconds"]),
183 dry_run=bool(raw["dry_run"]),
184 collected_ids=[str(x) for x in raw["collected_ids"]],
185 )
186 raise AssertionError(f"No JSON object found in output:\n{output}")
187
188
189 # ---------------------------------------------------------------------------
190 # Unit — _is_hex
191 # ---------------------------------------------------------------------------
192
193
194 class TestIsHex:
195 def test_empty_string_is_not_hex(self) -> None:
196 from muse.core.gc import _is_hex
197 assert not _is_hex("")
198
199 def test_valid_lowercase_hex(self) -> None:
200 from muse.core.gc import _is_hex
201 assert _is_hex("0123456789abcdef")
202
203 def test_uppercase_rejected(self) -> None:
204 from muse.core.gc import _is_hex
205 assert not _is_hex("ABCDEF")
206
207 def test_mixed_case_rejected(self) -> None:
208 from muse.core.gc import _is_hex
209 assert not _is_hex("0aF")
210
211 def test_non_hex_chars_rejected(self) -> None:
212 from muse.core.gc import _is_hex
213 assert not _is_hex("xyz")
214
215 def test_single_valid_char(self) -> None:
216 from muse.core.gc import _is_hex
217 assert _is_hex("a")
218
219 def test_64_char_sha256(self) -> None:
220 from muse.core.gc import _is_hex
221 sha = "a" * 64
222 assert _is_hex(sha)
223
224
225 # ---------------------------------------------------------------------------
226 # Unit — _fmt_bytes
227 # ---------------------------------------------------------------------------
228
229
230 class TestFmtBytes:
231 def test_bytes_range(self) -> None:
232 from muse.cli.commands.gc import _fmt_bytes
233 assert _fmt_bytes(0) == "0 B"
234 assert _fmt_bytes(1023) == "1023 B"
235
236 def test_kib_range(self) -> None:
237 from muse.cli.commands.gc import _fmt_bytes
238 assert "KiB" in _fmt_bytes(1024)
239 assert "KiB" in _fmt_bytes(1024 * 1024 - 1)
240
241 def test_mib_range(self) -> None:
242 from muse.cli.commands.gc import _fmt_bytes
243 assert "MiB" in _fmt_bytes(1024 * 1024)
244 assert "MiB" in _fmt_bytes(1024 * 1024 * 100)
245
246
247 # ---------------------------------------------------------------------------
248 # Unit — _list_stored_objects
249 # ---------------------------------------------------------------------------
250
251
252 class TestListStoredObjects:
253 def test_symlink_prefix_dir_is_skipped(self, tmp_path: pathlib.Path) -> None:
254 """A symlinked prefix directory must not be entered."""
255 from muse.core.gc import _list_stored_objects
256 root = _make_repo(tmp_path)
257 real_dir = tmp_path / "external_objects"
258 real_dir.mkdir()
259 sha = "a" * 64
260 real_file = real_dir / sha[2:]
261 real_file.write_bytes(b"content")
262
263 # Create a symlink at .muse/objects/sha256/<prefix> → external dir
264 algo_dir = objects_dir(root) / "sha256"
265 algo_dir.mkdir(parents=True, exist_ok=True)
266 link = algo_dir / sha[:2]
267 link.symlink_to(real_dir)
268
269 pairs = _list_stored_objects(root, grace_period_seconds=0)
270 found_ids = {oid for oid, _ in pairs}
271 assert sha not in found_ids, "Symlinked prefix dir must not be entered"
272
273 def test_symlink_object_file_is_skipped(self, tmp_path: pathlib.Path) -> None:
274 """A symlinked object file must not be listed or ever unlinked."""
275 from muse.core.gc import _list_stored_objects
276 root = _make_repo(tmp_path)
277 # Write a real file outside the repo.
278 external = tmp_path / "external_secret"
279 external.write_bytes(b"secret content")
280
281 sha = "b" * 64
282 link = object_path(root, long_id(sha))
283 link.parent.mkdir(parents=True, exist_ok=True)
284 link.symlink_to(external)
285
286 pairs = _list_stored_objects(root, grace_period_seconds=0)
287 found_ids = {oid for oid, _ in pairs}
288 assert sha not in found_ids, "Symlinked object file must not be listed"
289 # The external file must be untouched.
290 assert external.exists()
291
292 def test_grace_period_filters_recent_files(self, tmp_path: pathlib.Path) -> None:
293 """Objects written within the grace window are excluded."""
294 from muse.core.gc import _list_stored_objects
295 root = _make_repo(tmp_path)
296 _write_object(root, b"new orphan")
297 # Grace period of 60 s — the object was written <1 s ago.
298 pairs = _list_stored_objects(root, grace_period_seconds=60)
299 assert len(pairs) == 0
300
301 def test_grace_period_zero_includes_all_files(self, tmp_path: pathlib.Path) -> None:
302 """grace_period_seconds=0 bypasses the mtime check."""
303 from muse.core.gc import _list_stored_objects
304 root = _make_repo(tmp_path)
305 _write_object(root, b"orphan")
306 pairs = _list_stored_objects(root, grace_period_seconds=0)
307 assert len(pairs) == 1
308
309 def test_non_hex_prefix_dir_skipped(self, tmp_path: pathlib.Path) -> None:
310 from muse.core.gc import _list_stored_objects
311 root = _make_repo(tmp_path)
312 (objects_dir(root) / "sha256" / "zz").mkdir(parents=True)
313 pairs = _list_stored_objects(root, grace_period_seconds=0)
314 assert len(pairs) == 0
315
316 def test_non_hex_object_file_skipped(self, tmp_path: pathlib.Path) -> None:
317 from muse.core.gc import _list_stored_objects
318 root = _make_repo(tmp_path)
319 prefix = objects_dir(root) / "sha256" / "ab"
320 prefix.mkdir(parents=True)
321 (prefix / "not-valid-hex!").write_bytes(b"x")
322 pairs = _list_stored_objects(root, grace_period_seconds=0)
323 assert len(pairs) == 0
324
325 def test_valid_object_included(self, tmp_path: pathlib.Path) -> None:
326 from muse.core.gc import _list_stored_objects
327 root = _make_repo(tmp_path)
328 oid = _write_object(root, b"valid object")
329 pairs = _list_stored_objects(root, grace_period_seconds=0)
330 found_ids = {o for o, _ in pairs}
331 assert oid in found_ids
332
333
334 # ---------------------------------------------------------------------------
335 # Unit — _collect_reachable_objects
336 # ---------------------------------------------------------------------------
337
338
339 class TestCollectReachableObjects:
340 def test_shelf_symlink_skipped(self, tmp_path: pathlib.Path) -> None:
341 """A symlinked shelf.json is ignored during the reachability walk."""
342 from muse.core.gc import _collect_reachable_objects
343 root = _make_repo(tmp_path)
344 # Write a real object and make it look shelved via a symlink.
345 obj_id = _write_object(root, b"shelved content")
346 external = tmp_path / "real_shelf.json"
347 external.write_text(json.dumps([{
348 "snapshot_id": "s" * 64,
349 "branch": "main",
350 "created_at": "2026-01-01T00:00:00+00:00",
351 "snapshot": {"a.py": obj_id},
352 }]))
353 link = muse_dir(root) / "shelf.json"
354 link.symlink_to(external)
355
356 reachable = _collect_reachable_objects(root)
357 # The object should NOT be marked reachable (symlink was skipped).
358 assert obj_id not in reachable
359
360 def test_shelf_oversized_file_skipped(self, tmp_path: pathlib.Path) -> None:
361 """A shelf entry exceeding the size cap is skipped, not OOM-killed."""
362 from muse.core.gc import _collect_reachable_objects, _MAX_SHELF_BYTES
363 import unittest.mock as mock
364 root = _make_repo(tmp_path)
365 obj_id = _write_object(root, b"shelved content")
366 entry_path = _write_shelf_entry(root, {"a.py": obj_id})
367 fake_stat = os.stat_result((
368 stat.S_IFREG | 0o644, 0, 0, 1, 0, 0,
369 _MAX_SHELF_BYTES + 1, 0, 0, 0,
370 ))
371 with mock.patch.object(pathlib.Path, "stat", return_value=fake_stat):
372 reachable = _collect_reachable_objects(root)
373 assert obj_id not in reachable
374
375 def test_malformed_shelf_json_skipped(self, tmp_path: pathlib.Path) -> None:
376 from muse.core.gc import _collect_reachable_objects
377 root = _make_repo(tmp_path)
378 (muse_dir(root) / "shelf.json").write_text("not valid json{{{}}", encoding="utf-8")
379 # Should not raise.
380 reachable = _collect_reachable_objects(root)
381 assert isinstance(reachable, set)
382
383 def test_valid_shelf_objects_marked_reachable(self, tmp_path: pathlib.Path) -> None:
384 from muse.core.gc import _collect_reachable_objects
385 root = _make_repo(tmp_path)
386 obj_id = _write_object(root, b"shelved content")
387 _write_shelf_entry(root, {"a.py": obj_id})
388 reachable = _collect_reachable_objects(root)
389 assert obj_id in reachable
390
391
392 # ---------------------------------------------------------------------------
393 # Unit — run_gc result fields
394 # ---------------------------------------------------------------------------
395
396
397 class TestRunGcResult:
398 def test_grace_period_stored_in_result(self, tmp_path: pathlib.Path) -> None:
399 from muse.core.gc import run_gc
400 root = _make_repo(tmp_path)
401 result = run_gc(root, grace_period_seconds=42)
402 assert result.grace_period_seconds == 42
403
404 def test_dry_run_flag_stored_in_result(self, tmp_path: pathlib.Path) -> None:
405 from muse.core.gc import run_gc
406 root = _make_repo(tmp_path)
407 result = run_gc(root, dry_run=True, grace_period_seconds=0)
408 assert result.dry_run is True
409
410 def test_duration_ms_is_non_negative(self, tmp_path: pathlib.Path) -> None:
411 from muse.core.gc import run_gc
412 root = _make_repo(tmp_path)
413 result = run_gc(root, grace_period_seconds=0)
414 assert result.duration_ms >= 0.0
415
416
417 # ---------------------------------------------------------------------------
418 # Security — CLI
419 # ---------------------------------------------------------------------------
420
421
422 class TestSecurity:
423 def test_symlink_in_objects_not_deleted(self, tmp_path: pathlib.Path) -> None:
424 """GC must never delete a file outside the repo via a symlink."""
425 root = _make_repo(tmp_path)
426 _make_commit(root)
427 external = tmp_path / "precious_file"
428 external.write_bytes(b"important data")
429 sha = "c" * 64
430 link = object_path(root, long_id(sha))
431 link.parent.mkdir(parents=True, exist_ok=True)
432 link.symlink_to(external)
433
434 _invoke_gc(root)
435
436 assert external.exists(), "External file must not be deleted via symlink"
437
438 def test_ansi_in_object_id_sanitized(self, tmp_path: pathlib.Path) -> None:
439 """sanitize_display must strip ANSI sequences from object IDs in verbose output."""
440 root = _make_repo(tmp_path)
441 _make_commit(root)
442 # Write a real orphan (we can't control its SHA, but we test the path is taken).
443 _write_object(root, b"orphan for sanitize test")
444 result = _invoke_gc(root, "--verbose")
445 assert result.exit_code == 0
446 # The output must not contain raw ESC bytes.
447 assert "\x1b" not in result.output
448
449 def test_invalid_format_exits_nonzero_and_writes_stderr(
450 self, tmp_path: pathlib.Path
451 ) -> None:
452 root = _make_repo(tmp_path)
453 # argparse now uses choices= so invalid format triggers argparse error.
454 result = runner.invoke(cli, ["gc", "--format", "csv"], env=_env(root))
455 assert result.exit_code != 0
456
457 def test_negative_grace_period_rejected(self, tmp_path: pathlib.Path) -> None:
458 root = _make_repo(tmp_path)
459 result = runner.invoke(cli, ["gc", "--grace-period", "-1"], env=_env(root))
460 assert result.exit_code != 0
461
462
463 # ---------------------------------------------------------------------------
464 # Integration — JSON output schema
465 # ---------------------------------------------------------------------------
466
467
468 class TestJsonSchema:
469 def test_json_schema_all_fields_present(self, tmp_path: pathlib.Path) -> None:
470 root = _make_repo(tmp_path)
471 _make_commit(root)
472 _write_object(root, b"orphan for json test")
473 result = _invoke_gc(root, "--json")
474 assert result.exit_code == 0
475 payload = _parse_gc_json(result.output)
476 assert payload["collected_count"] == 1
477 assert payload["collected_bytes"] > 0
478 assert payload["reachable_count"] == 0
479 assert payload["duration_ms"] >= 0.0
480 assert payload["grace_period_seconds"] == 0
481 assert payload["dry_run"] is False
482 assert len(payload["collected_ids"]) == 1
483
484 def test_json_dry_run_does_not_delete(self, tmp_path: pathlib.Path) -> None:
485 root = _make_repo(tmp_path)
486 _make_commit(root)
487 orphan_id = _write_object(root, b"dry run orphan")
488 result = _invoke_gc(root, "--dry-run", "--json")
489 assert result.exit_code == 0
490 payload = _parse_gc_json(result.output)
491 assert payload["dry_run"] is True
492 assert payload["collected_count"] == 1
493 # File must still exist.
494 from muse.core.object_store import has_object
495 assert has_object(root, orphan_id)
496
497 def test_json_grace_period_field_reflects_flag(self, tmp_path: pathlib.Path) -> None:
498 root = _make_repo(tmp_path)
499 result = runner.invoke(
500 cli, ["gc", "--grace-period", "99", "--json"],
501 env=_env(root), catch_exceptions=False,
502 )
503 assert result.exit_code == 0
504 payload = _parse_gc_json(result.output)
505 assert payload["grace_period_seconds"] == 99
506
507 def test_json_collected_ids_sorted(self, tmp_path: pathlib.Path) -> None:
508 root = _make_repo(tmp_path)
509 for i in range(5):
510 _write_object(root, f"sort test {i}".encode())
511 result = _invoke_gc(root, "--json")
512 assert result.exit_code == 0
513 payload = _parse_gc_json(result.output)
514 assert payload["collected_ids"] == sorted(payload["collected_ids"])
515
516 def test_json_clean_repo_shows_zero_counts(self, tmp_path: pathlib.Path) -> None:
517 root = _make_repo(tmp_path)
518 _make_commit(root)
519 result = _invoke_gc(root, "--json")
520 assert result.exit_code == 0
521 payload = _parse_gc_json(result.output)
522 assert payload["collected_count"] == 0
523 assert payload["collected_bytes"] == 0
524 assert payload["collected_ids"] == []
525
526 def test_shorthand_json_flag(self, tmp_path: pathlib.Path) -> None:
527 root = _make_repo(tmp_path)
528 result = _invoke_gc(root, "--json")
529 assert result.exit_code == 0
530 _parse_gc_json(result.output) # must not raise
531
532
533 # ---------------------------------------------------------------------------
534 # E2E — full lifecycle
535 # ---------------------------------------------------------------------------
536
537
538 class TestE2E:
539 def test_orphan_from_abandoned_branch_reclaimed(self, tmp_path: pathlib.Path) -> None:
540 """Objects written for a branch that was never committed are reclaimed."""
541 root = _make_repo(tmp_path)
542 # Write objects that were staged but never committed.
543 orphan_a = _write_object(root, b"branch work A")
544 orphan_b = _write_object(root, b"branch work B")
545 # Now run GC.
546 result = _invoke_gc(root, "--json")
547 assert result.exit_code == 0
548 payload = _parse_gc_json(result.output)
549 assert orphan_a in payload["collected_ids"]
550 assert orphan_b in payload["collected_ids"]
551
552 def test_gc_after_shelf_save_preserves_shelf_objects(self, tmp_path: pathlib.Path) -> None:
553 root = _make_repo(tmp_path)
554 shelf_obj = _write_object(root, b"shelved file content")
555 _write_shelf_entry(root, {"file.py": shelf_obj})
556
557 result = _invoke_gc(root, "--json")
558 assert result.exit_code == 0
559 payload = _parse_gc_json(result.output)
560 assert shelf_obj not in payload["collected_ids"]
561 # Blob must still be on disk.
562 from muse.core.object_store import has_object
563 assert has_object(root, shelf_obj)
564
565 def test_gc_with_corrupt_shelf_json_succeeds(self, tmp_path: pathlib.Path) -> None:
566 root = _make_repo(tmp_path)
567 orphan = _write_object(root, b"orphan despite corrupt shelf")
568 (muse_dir(root) / "shelf.json").write_text("{not json", encoding="utf-8")
569 result = _invoke_gc(root, "--json")
570 assert result.exit_code == 0
571 payload = _parse_gc_json(result.output)
572 # Orphan is still collected even though shelf was corrupt.
573 assert orphan in payload["collected_ids"]
574
575 def test_grace_period_zero_collects_fresh_orphan(self, tmp_path: pathlib.Path) -> None:
576 root = _make_repo(tmp_path)
577 orphan = _write_object(root, b"fresh orphan")
578 result = _invoke_gc(root, "--grace-period", "0", "--json")
579 assert result.exit_code == 0
580 payload = _parse_gc_json(result.output)
581 assert orphan in payload["collected_ids"]
582
583 def test_grace_period_large_protects_fresh_orphan(self, tmp_path: pathlib.Path) -> None:
584 root = _make_repo(tmp_path)
585 _write_object(root, b"fresh orphan protected")
586 result = runner.invoke(
587 cli, ["gc", "--grace-period", "9999", "--json"],
588 env=_env(root), catch_exceptions=False,
589 )
590 assert result.exit_code == 0
591 payload = _parse_gc_json(result.output)
592 assert payload["collected_count"] == 0
593
594 def test_repeated_gc_is_idempotent(self, tmp_path: pathlib.Path) -> None:
595 root = _make_repo(tmp_path)
596 _write_object(root, b"first orphan")
597 _invoke_gc(root)
598 result2 = _invoke_gc(root, "--json")
599 assert result2.exit_code == 0
600 payload = _parse_gc_json(result2.output)
601 assert payload["collected_count"] == 0
602
603 def test_gc_removes_empty_prefix_dirs(self, tmp_path: pathlib.Path) -> None:
604 """After GC, empty prefix dirs under .muse/objects/sha256/ are cleaned up."""
605 root = _make_repo(tmp_path)
606 sha = _write_object(root, b"sole object in prefix")
607 from muse.core.object_store import object_path
608 prefix_dir = object_path(root, sha).parent
609 assert prefix_dir.exists()
610 _invoke_gc(root)
611 # Directory should be removed since it's empty now.
612 assert not prefix_dir.exists()
613
614 def test_verbose_lists_full_sha256_ids(self, tmp_path: pathlib.Path) -> None:
615 root = _make_repo(tmp_path)
616 orphan = _write_object(root, b"verbose test object")
617 result = _invoke_gc(root, "--verbose")
618 assert result.exit_code == 0
619 assert orphan in result.output
620
621 def test_dry_run_verbose_lists_without_deleting(self, tmp_path: pathlib.Path) -> None:
622 root = _make_repo(tmp_path)
623 orphan = _write_object(root, b"dry verbose test")
624 result = _invoke_gc(root, "--dry-run", "--verbose")
625 assert result.exit_code == 0
626 assert orphan in result.output
627 from muse.core.object_store import object_path
628 assert object_path(root, orphan).exists()
629
630 def test_dry_run_prefix_present_in_text_output(self, tmp_path: pathlib.Path) -> None:
631 root = _make_repo(tmp_path)
632 result = _invoke_gc(root, "--dry-run")
633 assert result.exit_code == 0
634 assert "[dry-run]" in result.output
635
636 def test_reachable_count_reflects_committed_objects(self, tmp_path: pathlib.Path) -> None:
637 root = _make_repo(tmp_path)
638 obj = _write_object(root, b"committed content")
639 _make_commit(root, manifest={"file.txt": obj})
640 result = _invoke_gc(root, "--json")
641 payload = _parse_gc_json(result.output)
642 assert payload["reachable_count"] == 1
643 assert payload["collected_count"] == 0
644
645
646 # ---------------------------------------------------------------------------
647 # Stress
648 # ---------------------------------------------------------------------------
649
650
651 class TestStress:
652 def test_500_orphans_all_collected(self, tmp_path: pathlib.Path) -> None:
653 root = _make_repo(tmp_path)
654 _make_commit(root)
655 orphan_ids = [_write_object(root, f"stress-{i:04d}".encode()) for i in range(500)]
656 result = _invoke_gc(root, "--json")
657 assert result.exit_code == 0
658 payload = _parse_gc_json(result.output)
659 assert payload["collected_count"] == 500
660 assert set(payload["collected_ids"]) == set(orphan_ids)
661 # Objects directory should be empty after GC.
662 obj_dir = objects_dir(root)
663 remaining_files = [p for p in obj_dir.rglob("*") if p.is_file()]
664 assert remaining_files == []
665
666 def test_concurrent_dry_run_does_not_crash(self, tmp_path: pathlib.Path) -> None:
667 """Multiple concurrent dry-run GCs on the same repo must not crash."""
668 root = _make_repo(tmp_path)
669 _make_commit(root)
670 for i in range(20):
671 _write_object(root, f"concurrent-orphan-{i}".encode())
672
673 errors: list[str] = []
674
675 def _run_dry() -> None:
676 try:
677 from muse.core.gc import run_gc
678 run_gc(root, dry_run=True, grace_period_seconds=0)
679 except Exception as exc:
680 errors.append(str(exc))
681
682 threads = [threading.Thread(target=_run_dry) for _ in range(8)]
683 for t in threads:
684 t.start()
685 for t in threads:
686 t.join()
687
688 assert not errors, f"Concurrent dry-run GC failures: {errors}"
689
690 def test_gc_across_many_prefix_dirs(self, tmp_path: pathlib.Path) -> None:
691 """Objects spread across many prefix dirs are all found and collected."""
692 root = _make_repo(tmp_path)
693 # Force objects into many distinct prefix dirs by varying content.
694 ids: list[str] = []
695 for i in range(100):
696 ids.append(_write_object(root, f"spread-{i:08d}".encode()))
697 # Verify we have multiple prefix dirs.
698 algo_dir = objects_dir(root) / "sha256"
699 prefix_dirs = [d for d in algo_dir.iterdir() if d.is_dir()]
700 assert len(prefix_dirs) > 1, "Test needs objects in multiple prefix dirs"
701
702 result = _invoke_gc(root, "--json")
703 payload = _parse_gc_json(result.output)
704 assert payload["collected_count"] == 100
705 assert set(payload["collected_ids"]) == set(ids)
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 122 days ago