gabriel / muse public
test_cmd_gc.py python
258 lines 9.3 KB
Raw
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 5 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 from muse.core.object_store import write_object
56 oid = blob_id(content)
57 write_object(root, oid, content)
58 return oid
59
60
61 def _make_commit(root: pathlib.Path, repo_id: str, message: str = "init") -> str:
62 from muse.core.commits import (
63 CommitRecord,
64 write_commit,
65 )
66 from muse.core.snapshots import (
67 SnapshotRecord,
68 write_snapshot,
69 )
70 from muse.core.ids import hash_snapshot, hash_commit
71
72 ref_file = heads_dir(root) / "main"
73 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
74 manifest: Manifest = {}
75 snap_id = hash_snapshot(manifest)
76 committed_at = datetime.datetime.now(datetime.timezone.utc)
77 commit_id = hash_commit( parent_ids=[parent_id] if parent_id else [],
78 snapshot_id=snap_id,
79 message=message,
80 committed_at_iso=committed_at.isoformat(),
81 )
82 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
83 write_commit(root, CommitRecord(
84 commit_id=commit_id, branch="main",
85 snapshot_id=snap_id, message=message, committed_at=committed_at,
86 parent_commit_id=parent_id,
87 ))
88 ref_file.parent.mkdir(parents=True, exist_ok=True)
89 ref_file.write_text(commit_id, encoding="utf-8")
90 return commit_id
91
92
93 # ---------------------------------------------------------------------------
94 # Unit tests
95 # ---------------------------------------------------------------------------
96
97
98 class TestRegisterFlags:
99 def _parse(self, *args: str) -> "argparse.Namespace":
100 import argparse
101 from muse.cli.commands.gc import register
102 p = argparse.ArgumentParser()
103 sub = p.add_subparsers()
104 register(sub)
105 return p.parse_args(["gc", *args])
106
107 def test_default_json_out_is_false(self) -> None:
108 ns = self._parse()
109 assert ns.json_out is False
110
111 def test_json_flag_sets_json_out(self) -> None:
112 ns = self._parse("--json")
113 assert ns.json_out is True
114
115 def test_j_shorthand_sets_json_out(self) -> None:
116 ns = self._parse("-j")
117 assert ns.json_out is True
118
119
120 class TestGcUnit:
121 def test_run_gc_empty_repo(self, tmp_path: pathlib.Path) -> None:
122 root, _ = _init_repo(tmp_path)
123 from muse.core.gc import run_gc
124 result = run_gc(root, dry_run=False)
125 assert result.collected_count == 0
126
127 def test_run_gc_dry_run_does_not_delete(self, tmp_path: pathlib.Path) -> None:
128 root, _ = _init_repo(tmp_path)
129 orphan_id = _write_object(root, b"orphaned content")
130 from muse.core.gc import run_gc
131 result = run_gc(root, dry_run=True, grace_period_seconds=0)
132 assert object_path(root, orphan_id).exists()
133 assert result.collected_count >= 1
134
135 def test_run_gc_collects_unreachable_objects(self, tmp_path: pathlib.Path) -> None:
136 root, repo_id = _init_repo(tmp_path)
137 _make_commit(root, repo_id, message="committed")
138 orphan_id = _write_object(root, b"never committed content")
139 from muse.core.gc import run_gc
140 result = run_gc(root, dry_run=False, grace_period_seconds=0)
141 assert not object_path(root, orphan_id).exists()
142 assert orphan_id in result.collected_ids
143
144
145 # ---------------------------------------------------------------------------
146 # Integration (CLI) tests
147 # ---------------------------------------------------------------------------
148
149 class TestGcIntegration:
150 def test_gc_default_clean_repo(self, tmp_path: pathlib.Path) -> None:
151 root, repo_id = _init_repo(tmp_path)
152 _make_commit(root, repo_id)
153 result = runner.invoke(cli, ["gc"], env=_env(root), catch_exceptions=False)
154 assert result.exit_code == 0
155
156 def test_gc_dry_run_reports_orphans(self, tmp_path: pathlib.Path) -> None:
157 root, repo_id = _init_repo(tmp_path)
158 _make_commit(root, repo_id)
159 _write_object(root, b"orphan1")
160 _write_object(root, b"orphan2")
161 result = runner.invoke(
162 cli, ["gc", "--dry-run", "--grace-period", "0"],
163 env=_env(root), catch_exceptions=False,
164 )
165 assert result.exit_code == 0
166 assert "2" in result.output or "collect" in result.output.lower()
167
168 def test_gc_verbose_shows_ids(self, tmp_path: pathlib.Path) -> None:
169 root, repo_id = _init_repo(tmp_path)
170 _make_commit(root, repo_id)
171 orphan_id = _write_object(root, b"verbose orphan")
172 result = runner.invoke(
173 cli, ["gc", "--verbose", "--grace-period", "0"],
174 env=_env(root), catch_exceptions=False,
175 )
176 assert result.exit_code == 0
177 assert short_id(orphan_id, strip=True) in result.output
178
179 def test_gc_output_includes_count(self, tmp_path: pathlib.Path) -> None:
180 root, repo_id = _init_repo(tmp_path)
181 _write_object(root, b"orphan for count test")
182 result = runner.invoke(
183 cli, ["gc", "--grace-period", "0"],
184 env=_env(root), catch_exceptions=False,
185 )
186 assert result.exit_code == 0
187 assert "Removed" in result.output or "object" in result.output
188
189 def test_gc_keeps_referenced_objects(self, tmp_path: pathlib.Path) -> None:
190 root, repo_id = _init_repo(tmp_path)
191 content = b"referenced file content"
192 obj_id = _write_object(root, content)
193
194 from muse.core.commits import (
195 CommitRecord,
196 write_commit,
197 )
198 from muse.core.snapshots import (
199 SnapshotRecord,
200 write_snapshot,
201 )
202 from muse.core.ids import hash_snapshot, hash_commit
203
204 manifest = {"file.mid": obj_id}
205 snap_id = hash_snapshot(manifest)
206 committed_at = datetime.datetime.now(datetime.timezone.utc)
207 commit_id = hash_commit( parent_ids=[],
208 snapshot_id=snap_id,
209 message="with file",
210 committed_at_iso=committed_at.isoformat(),
211 )
212 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
213 write_commit(root, CommitRecord(
214 commit_id=commit_id, branch="main",
215 snapshot_id=snap_id, message="with file",
216 committed_at=committed_at, parent_commit_id=None,
217 ))
218 (heads_dir(root) / "main").write_text(commit_id)
219
220 runner.invoke(cli, ["gc", "--grace-period", "0"], env=_env(root), catch_exceptions=False)
221 assert object_path(root, obj_id).exists()
222
223 def test_gc_short_flags(self, tmp_path: pathlib.Path) -> None:
224 root, repo_id = _init_repo(tmp_path)
225 _make_commit(root, repo_id)
226 _write_object(root, b"short flag orphan")
227 result = runner.invoke(
228 cli, ["gc", "-n", "-v", "--grace-period", "0"],
229 env=_env(root), catch_exceptions=False,
230 )
231 assert result.exit_code == 0
232
233
234 # ---------------------------------------------------------------------------
235 # Stress tests
236 # ---------------------------------------------------------------------------
237
238 class TestGcStress:
239 def test_gc_many_orphaned_objects(self, tmp_path: pathlib.Path) -> None:
240 root, repo_id = _init_repo(tmp_path)
241 _make_commit(root, repo_id)
242 orphan_ids = [_write_object(root, f"orphan {i}".encode()) for i in range(100)]
243
244 result = runner.invoke(
245 cli, ["gc", "--grace-period", "0"], env=_env(root), catch_exceptions=False,
246 )
247 assert result.exit_code == 0
248 assert "100" in result.output
249
250 for oid in orphan_ids:
251 assert not object_path(root, oid).exists()
252
253 def test_gc_repeated_runs_idempotent(self, tmp_path: pathlib.Path) -> None:
254 root, repo_id = _init_repo(tmp_path)
255 _make_commit(root, repo_id)
256 for _ in range(3):
257 result = runner.invoke(cli, ["gc"], env=_env(root), catch_exceptions=False)
258 assert result.exit_code == 0
File History 1 commit
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 5 days ago