gabriel / muse public
test_cmd_gc.py python
245 lines 9.3 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 123 days ago
1 """Comprehensive tests for ``muse gc``.
2
3 Covers:
4 - Unit: run_gc core logic (reachable vs unreachable objects)
5 - Integration: gc cleans up orphaned objects after commits
6 - E2E: full CLI via CliRunner (--dry-run, --verbose, --format json)
7 - Security: only objects dir affected, no path traversal
8 - Stress: gc with many orphaned objects
9 """
10
11 from __future__ import annotations
12
13 import datetime
14 import json
15 import pathlib
16
17 import pytest
18 from tests.cli_test_helper import CliRunner
19 from muse.core.types import blob_id, fake_id, short_id
20 from muse.core.object_store import object_path
21 from muse.core.paths import heads_dir, muse_dir
22
23 cli = None # argparse migration — CliRunner ignores this arg
24
25 runner = CliRunner()
26
27
28 # ---------------------------------------------------------------------------
29 # Helpers
30 # ---------------------------------------------------------------------------
31
32 def _env(root: pathlib.Path) -> Manifest:
33 return {"MUSE_REPO_ROOT": str(root)}
34
35
36 def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
37 dot_muse = muse_dir(tmp_path)
38 dot_muse.mkdir()
39 repo_id = fake_id("repo")
40 (dot_muse / "repo.json").write_text(json.dumps({
41 "repo_id": repo_id,
42 "domain": "midi",
43 "default_branch": "main",
44 "created_at": "2025-01-01T00:00:00+00:00",
45 }), encoding="utf-8")
46 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
47 (dot_muse / "refs" / "heads").mkdir(parents=True)
48 (dot_muse / "snapshots").mkdir()
49 (dot_muse / "commits").mkdir()
50 (dot_muse / "objects" / "sha256").mkdir(parents=True)
51 return tmp_path, repo_id
52
53
54 def _write_object(root: pathlib.Path, content: bytes) -> str:
55 oid = blob_id(content)
56 p = object_path(root, oid)
57 p.parent.mkdir(parents=True, exist_ok=True)
58 p.write_bytes(content)
59 return oid
60
61
62 def _make_commit(root: pathlib.Path, repo_id: str, message: str = "init") -> str:
63 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
64 from muse.core.snapshot import compute_snapshot_id, compute_commit_id
65
66 ref_file = heads_dir(root) / "main"
67 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
68 manifest: Manifest = {}
69 snap_id = compute_snapshot_id(manifest)
70 committed_at = datetime.datetime.now(datetime.timezone.utc)
71 commit_id = compute_commit_id( parent_ids=[parent_id] if parent_id else [],
72 snapshot_id=snap_id,
73 message=message,
74 committed_at_iso=committed_at.isoformat(),
75 )
76 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
77 write_commit(root, CommitRecord(
78 commit_id=commit_id, repo_id=repo_id, branch="main",
79 snapshot_id=snap_id, message=message, committed_at=committed_at,
80 parent_commit_id=parent_id,
81 ))
82 ref_file.parent.mkdir(parents=True, exist_ok=True)
83 ref_file.write_text(commit_id, encoding="utf-8")
84 return commit_id
85
86
87 # ---------------------------------------------------------------------------
88 # Unit tests
89 # ---------------------------------------------------------------------------
90
91
92 class TestRegisterFlags:
93 def _parse(self, *args: str) -> "argparse.Namespace":
94 import argparse
95 from muse.cli.commands.gc import register
96 p = argparse.ArgumentParser()
97 sub = p.add_subparsers()
98 register(sub)
99 return p.parse_args(["gc", *args])
100
101 def test_default_json_out_is_false(self) -> None:
102 ns = self._parse()
103 assert ns.json_out is False
104
105 def test_json_flag_sets_json_out(self) -> None:
106 ns = self._parse("--json")
107 assert ns.json_out is True
108
109 def test_j_shorthand_sets_json_out(self) -> None:
110 ns = self._parse("-j")
111 assert ns.json_out is True
112
113
114 class TestGcUnit:
115 def test_run_gc_empty_repo(self, tmp_path: pathlib.Path) -> None:
116 root, _ = _init_repo(tmp_path)
117 from muse.core.gc import run_gc
118 result = run_gc(root, dry_run=False)
119 assert result.collected_count == 0
120
121 def test_run_gc_dry_run_does_not_delete(self, tmp_path: pathlib.Path) -> None:
122 root, _ = _init_repo(tmp_path)
123 orphan_id = _write_object(root, b"orphaned content")
124 from muse.core.gc import run_gc
125 result = run_gc(root, dry_run=True, grace_period_seconds=0)
126 assert object_path(root, orphan_id).exists()
127 assert result.collected_count >= 1
128
129 def test_run_gc_collects_unreachable_objects(self, tmp_path: pathlib.Path) -> None:
130 root, repo_id = _init_repo(tmp_path)
131 _make_commit(root, repo_id, message="committed")
132 orphan_id = _write_object(root, b"never committed content")
133 from muse.core.gc import run_gc
134 result = run_gc(root, dry_run=False, grace_period_seconds=0)
135 assert not object_path(root, orphan_id).exists()
136 assert orphan_id in result.collected_ids
137
138
139 # ---------------------------------------------------------------------------
140 # Integration (CLI) tests
141 # ---------------------------------------------------------------------------
142
143 class TestGcIntegration:
144 def test_gc_default_clean_repo(self, tmp_path: pathlib.Path) -> None:
145 root, repo_id = _init_repo(tmp_path)
146 _make_commit(root, repo_id)
147 result = runner.invoke(cli, ["gc"], env=_env(root), catch_exceptions=False)
148 assert result.exit_code == 0
149
150 def test_gc_dry_run_reports_orphans(self, tmp_path: pathlib.Path) -> None:
151 root, repo_id = _init_repo(tmp_path)
152 _make_commit(root, repo_id)
153 _write_object(root, b"orphan1")
154 _write_object(root, b"orphan2")
155 result = runner.invoke(
156 cli, ["gc", "--dry-run", "--grace-period", "0"],
157 env=_env(root), catch_exceptions=False,
158 )
159 assert result.exit_code == 0
160 assert "2" in result.output or "collect" in result.output.lower()
161
162 def test_gc_verbose_shows_ids(self, tmp_path: pathlib.Path) -> None:
163 root, repo_id = _init_repo(tmp_path)
164 _make_commit(root, repo_id)
165 orphan_id = _write_object(root, b"verbose orphan")
166 result = runner.invoke(
167 cli, ["gc", "--verbose", "--grace-period", "0"],
168 env=_env(root), catch_exceptions=False,
169 )
170 assert result.exit_code == 0
171 assert short_id(orphan_id, strip=True) in result.output
172
173 def test_gc_output_includes_count(self, tmp_path: pathlib.Path) -> None:
174 root, repo_id = _init_repo(tmp_path)
175 _write_object(root, b"orphan for count test")
176 result = runner.invoke(
177 cli, ["gc", "--grace-period", "0"],
178 env=_env(root), catch_exceptions=False,
179 )
180 assert result.exit_code == 0
181 assert "Removed" in result.output or "object" in result.output
182
183 def test_gc_keeps_referenced_objects(self, tmp_path: pathlib.Path) -> None:
184 root, repo_id = _init_repo(tmp_path)
185 content = b"referenced file content"
186 obj_id = _write_object(root, content)
187
188 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
189 from muse.core.snapshot import compute_snapshot_id, compute_commit_id
190
191 manifest = {"file.mid": obj_id}
192 snap_id = compute_snapshot_id(manifest)
193 committed_at = datetime.datetime.now(datetime.timezone.utc)
194 commit_id = compute_commit_id( parent_ids=[],
195 snapshot_id=snap_id,
196 message="with file",
197 committed_at_iso=committed_at.isoformat(),
198 )
199 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
200 write_commit(root, CommitRecord(
201 commit_id=commit_id, repo_id=repo_id, branch="main",
202 snapshot_id=snap_id, message="with file",
203 committed_at=committed_at, parent_commit_id=None,
204 ))
205 (heads_dir(root) / "main").write_text(commit_id)
206
207 runner.invoke(cli, ["gc", "--grace-period", "0"], env=_env(root), catch_exceptions=False)
208 assert object_path(root, obj_id).exists()
209
210 def test_gc_short_flags(self, tmp_path: pathlib.Path) -> None:
211 root, repo_id = _init_repo(tmp_path)
212 _make_commit(root, repo_id)
213 _write_object(root, b"short flag orphan")
214 result = runner.invoke(
215 cli, ["gc", "-n", "-v", "--grace-period", "0"],
216 env=_env(root), catch_exceptions=False,
217 )
218 assert result.exit_code == 0
219
220
221 # ---------------------------------------------------------------------------
222 # Stress tests
223 # ---------------------------------------------------------------------------
224
225 class TestGcStress:
226 def test_gc_many_orphaned_objects(self, tmp_path: pathlib.Path) -> None:
227 root, repo_id = _init_repo(tmp_path)
228 _make_commit(root, repo_id)
229 orphan_ids = [_write_object(root, f"orphan {i}".encode()) for i in range(100)]
230
231 result = runner.invoke(
232 cli, ["gc", "--grace-period", "0"], env=_env(root), catch_exceptions=False,
233 )
234 assert result.exit_code == 0
235 assert "100" in result.output
236
237 for oid in orphan_ids:
238 assert not object_path(root, oid).exists()
239
240 def test_gc_repeated_runs_idempotent(self, tmp_path: pathlib.Path) -> None:
241 root, repo_id = _init_repo(tmp_path)
242 _make_commit(root, repo_id)
243 for _ in range(3):
244 result = runner.invoke(cli, ["gc"], env=_env(root), catch_exceptions=False)
245 assert result.exit_code == 0
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 123 days ago