gabriel / muse public
test_gc_supercharge.py python
425 lines 16.1 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 141 days ago
1 """GC JSON schema: agent-ready output fields.
2
3 Tests for ``muse gc --json``:
4
5 status "ok" | "error"
6 error empty string on success; message on bad args
7 warnings list of warning strings (symlink skips, etc.)
8 mode "conservative" (default) | "tight" (--full)
9 collected_commit_ids list[str] — pruned commit IDs (--full only)
10 collected_snapshot_ids list[str] — pruned snapshot IDs (--full only)
11 duration_ms float — milliseconds
12 exit_code int — 0 on success, 1 on error
13
14 Also covers:
15 - structured JSON error for --grace-period < 0 in --json mode
16
17 Test categories
18 ---------------
19 TestGcJsonSchema — every field present and typed correctly
20 TestGcJsonDurationMs — duration_ms is present and non-negative
21 TestGcJsonMode — mode field reflects --full flag
22 TestGcJsonCollectedIds — collected_commit_ids / collected_snapshot_ids
23 TestGcJsonBadArgs — structured error when --grace-period < 0
24 TestGcJsonWarnings — warnings list populated on symlink skip
25 TestGcJsonExitCode — exit_code field in JSON output
26 """
27
28 from __future__ import annotations
29
30 import datetime
31 import hashlib
32 import json
33 import pathlib
34 import uuid
35
36 import pytest
37
38 from tests.cli_test_helper import CliRunner
39
40 runner = CliRunner()
41 cli = None # argparse migration — CliRunner ignores this arg
42
43
44 # ---------------------------------------------------------------------------
45 # Helpers
46 # ---------------------------------------------------------------------------
47
48 def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
49 muse_dir = tmp_path / ".muse"
50 muse_dir.mkdir()
51 repo_id = str(uuid.uuid4())
52 (muse_dir / "repo.json").write_text(json.dumps({
53 "repo_id": repo_id,
54 "domain": "code",
55 "default_branch": "main",
56 "created_at": "2025-01-01T00:00:00+00:00",
57 }), encoding="utf-8")
58 (muse_dir / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
59 (muse_dir / "refs" / "heads").mkdir(parents=True)
60 (muse_dir / "snapshots").mkdir()
61 (muse_dir / "commits").mkdir()
62 (muse_dir / "objects").mkdir()
63 return tmp_path, repo_id
64
65
66 def _write_object(root: pathlib.Path, content: bytes) -> str:
67 obj_id = hashlib.sha256(content).hexdigest()
68 obj_path = root / ".muse" / "objects" / obj_id[:2] / obj_id[2:]
69 obj_path.parent.mkdir(parents=True, exist_ok=True)
70 obj_path.write_bytes(content)
71 return obj_id
72
73
74 def _make_commit(root: pathlib.Path, repo_id: str, message: str = "init") -> str:
75 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
76 from muse.core.snapshot import compute_snapshot_id, compute_commit_id
77
78 ref_file = root / ".muse" / "refs" / "heads" / "main"
79 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
80 manifest: dict = {}
81 snap_id = compute_snapshot_id(manifest)
82 committed_at = datetime.datetime.now(datetime.timezone.utc)
83 commit_id = compute_commit_id(
84 parent_ids=[parent_id] if parent_id else [],
85 snapshot_id=snap_id, message=message,
86 committed_at_iso=committed_at.isoformat(),
87 )
88 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
89 write_commit(root, CommitRecord(
90 commit_id=commit_id, repo_id=repo_id, branch="main",
91 snapshot_id=snap_id, message=message, committed_at=committed_at,
92 parent_commit_id=parent_id,
93 ))
94 ref_file.parent.mkdir(parents=True, exist_ok=True)
95 ref_file.write_text(commit_id, encoding="utf-8")
96 return commit_id
97
98
99 def _gc_json(root: pathlib.Path, extra_args: list[str] | None = None) -> dict:
100 """Run ``muse gc --json --grace-period 0`` and parse output."""
101 args = ["gc", "--json", "--grace-period", "0"] + (extra_args or [])
102 result = runner.invoke(cli, args, env={"MUSE_REPO_ROOT": str(root)})
103 return json.loads(result.output)
104
105
106 # ---------------------------------------------------------------------------
107 # TestGcJsonSchema
108 # ---------------------------------------------------------------------------
109
110 class TestGcJsonSchema:
111 """Every new agent-ready field must be present and correctly typed."""
112
113 def test_status_field_present(self, tmp_path: pathlib.Path) -> None:
114 root, repo_id = _init_repo(tmp_path)
115 _make_commit(root, repo_id)
116 data = _gc_json(root)
117 assert "status" in data, "JSON output must include 'status' field"
118
119 def test_status_ok_on_success(self, tmp_path: pathlib.Path) -> None:
120 root, repo_id = _init_repo(tmp_path)
121 _make_commit(root, repo_id)
122 data = _gc_json(root)
123 assert data["status"] == "ok"
124
125 def test_error_field_present(self, tmp_path: pathlib.Path) -> None:
126 root, repo_id = _init_repo(tmp_path)
127 _make_commit(root, repo_id)
128 data = _gc_json(root)
129 assert "error" in data, "JSON output must include 'error' field"
130
131 def test_error_empty_on_success(self, tmp_path: pathlib.Path) -> None:
132 root, repo_id = _init_repo(tmp_path)
133 _make_commit(root, repo_id)
134 data = _gc_json(root)
135 assert data["error"] == ""
136
137 def test_warnings_field_present(self, tmp_path: pathlib.Path) -> None:
138 root, repo_id = _init_repo(tmp_path)
139 _make_commit(root, repo_id)
140 data = _gc_json(root)
141 assert "warnings" in data, "JSON output must include 'warnings' field"
142
143 def test_warnings_is_list(self, tmp_path: pathlib.Path) -> None:
144 root, repo_id = _init_repo(tmp_path)
145 _make_commit(root, repo_id)
146 data = _gc_json(root)
147 assert isinstance(data["warnings"], list)
148
149 def test_warnings_empty_on_clean_run(self, tmp_path: pathlib.Path) -> None:
150 root, repo_id = _init_repo(tmp_path)
151 _make_commit(root, repo_id)
152 data = _gc_json(root)
153 assert data["warnings"] == []
154
155 def test_mode_field_present(self, tmp_path: pathlib.Path) -> None:
156 root, repo_id = _init_repo(tmp_path)
157 _make_commit(root, repo_id)
158 data = _gc_json(root)
159 assert "mode" in data, "JSON output must include 'mode' field"
160
161 def test_exit_code_field_present(self, tmp_path: pathlib.Path) -> None:
162 root, repo_id = _init_repo(tmp_path)
163 _make_commit(root, repo_id)
164 data = _gc_json(root)
165 assert "exit_code" in data, "JSON output must include 'exit_code' field"
166
167 def test_collected_commit_ids_present(self, tmp_path: pathlib.Path) -> None:
168 root, repo_id = _init_repo(tmp_path)
169 _make_commit(root, repo_id)
170 data = _gc_json(root, ["--full"])
171 assert "collected_commit_ids" in data, "JSON must include collected_commit_ids"
172
173 def test_collected_snapshot_ids_present(self, tmp_path: pathlib.Path) -> None:
174 root, repo_id = _init_repo(tmp_path)
175 _make_commit(root, repo_id)
176 data = _gc_json(root, ["--full"])
177 assert "collected_snapshot_ids" in data, "JSON must include collected_snapshot_ids"
178
179 def test_collected_commit_ids_is_list(self, tmp_path: pathlib.Path) -> None:
180 root, repo_id = _init_repo(tmp_path)
181 _make_commit(root, repo_id)
182 data = _gc_json(root, ["--full"])
183 assert isinstance(data["collected_commit_ids"], list)
184
185 def test_collected_snapshot_ids_is_list(self, tmp_path: pathlib.Path) -> None:
186 root, repo_id = _init_repo(tmp_path)
187 _make_commit(root, repo_id)
188 data = _gc_json(root, ["--full"])
189 assert isinstance(data["collected_snapshot_ids"], list)
190
191
192 # ---------------------------------------------------------------------------
193 # TestGcJsonDurationMs
194 # ---------------------------------------------------------------------------
195
196 class TestGcJsonDurationMs:
197 """duration_ms field is present, numeric, and non-negative."""
198
199 def test_duration_ms_present(self, tmp_path: pathlib.Path) -> None:
200 root, repo_id = _init_repo(tmp_path)
201 _make_commit(root, repo_id)
202 data = _gc_json(root)
203 assert "duration_ms" in data, "JSON must include 'duration_ms' field"
204
205 def test_duration_ms_is_float(self, tmp_path: pathlib.Path) -> None:
206 root, repo_id = _init_repo(tmp_path)
207 _make_commit(root, repo_id)
208 data = _gc_json(root)
209 assert isinstance(data["duration_ms"], (int, float))
210
211 def test_duration_ms_non_negative(self, tmp_path: pathlib.Path) -> None:
212 root, repo_id = _init_repo(tmp_path)
213 _make_commit(root, repo_id)
214 data = _gc_json(root)
215 assert data["duration_ms"] >= 0
216
217 def test_no_elapsed_key(self, tmp_path: pathlib.Path) -> None:
218 root, repo_id = _init_repo(tmp_path)
219 _make_commit(root, repo_id)
220 data = _gc_json(root)
221 assert "elapsed_ms" not in data
222 assert "elapsed" not in data
223
224
225 # ---------------------------------------------------------------------------
226 # TestGcJsonMode
227 # ---------------------------------------------------------------------------
228
229 class TestGcJsonMode:
230 """mode field reflects which reachability strategy was used."""
231
232 def test_mode_conservative_by_default(self, tmp_path: pathlib.Path) -> None:
233 root, repo_id = _init_repo(tmp_path)
234 _make_commit(root, repo_id)
235 data = _gc_json(root)
236 assert data["mode"] == "conservative"
237
238 def test_mode_tight_with_full_flag(self, tmp_path: pathlib.Path) -> None:
239 root, repo_id = _init_repo(tmp_path)
240 _make_commit(root, repo_id)
241 data = _gc_json(root, ["--full"])
242 assert data["mode"] == "tight"
243
244
245 # ---------------------------------------------------------------------------
246 # TestGcJsonCollectedIds
247 # ---------------------------------------------------------------------------
248
249 class TestGcJsonCollectedIds:
250 """collected_commit_ids and collected_snapshot_ids populated in --full mode."""
251
252 def test_collected_commit_ids_empty_when_all_reachable(
253 self, tmp_path: pathlib.Path
254 ) -> None:
255 root, repo_id = _init_repo(tmp_path)
256 _make_commit(root, repo_id)
257 data = _gc_json(root, ["--full"])
258 assert data["collected_commit_ids"] == []
259
260 def test_collected_snapshot_ids_empty_when_all_reachable(
261 self, tmp_path: pathlib.Path
262 ) -> None:
263 root, repo_id = _init_repo(tmp_path)
264 _make_commit(root, repo_id)
265 data = _gc_json(root, ["--full"])
266 assert data["collected_snapshot_ids"] == []
267
268 def test_collected_commit_ids_conservative_mode_always_empty(
269 self, tmp_path: pathlib.Path
270 ) -> None:
271 """Conservative mode doesn't prune commits — list must be empty."""
272 root, repo_id = _init_repo(tmp_path)
273 _make_commit(root, repo_id)
274 data = _gc_json(root) # no --full
275 assert data["collected_commit_ids"] == []
276
277 def test_collected_snapshot_ids_conservative_mode_always_empty(
278 self, tmp_path: pathlib.Path
279 ) -> None:
280 root, repo_id = _init_repo(tmp_path)
281 _make_commit(root, repo_id)
282 data = _gc_json(root) # no --full
283 assert data["collected_snapshot_ids"] == []
284
285
286 # ---------------------------------------------------------------------------
287 # TestGcJsonBadArgs
288 # ---------------------------------------------------------------------------
289
290 class TestGcJsonBadArgs:
291 """--grace-period < 0 with --json must emit structured JSON error, not crash."""
292
293 def test_bad_grace_period_json_mode_exit_code_1(
294 self, tmp_path: pathlib.Path
295 ) -> None:
296 root, _ = _init_repo(tmp_path)
297 result = runner.invoke(
298 cli,
299 ["gc", "--json", "--grace-period", "-1"],
300 env={"MUSE_REPO_ROOT": str(root)},
301 )
302 assert result.exit_code == 1
303
304 def test_bad_grace_period_json_mode_emits_json(
305 self, tmp_path: pathlib.Path
306 ) -> None:
307 root, _ = _init_repo(tmp_path)
308 result = runner.invoke(
309 cli,
310 ["gc", "--json", "--grace-period", "-1"],
311 env={"MUSE_REPO_ROOT": str(root)},
312 )
313 # Output must be valid JSON (not just a stderr print)
314 data = json.loads(result.output)
315 assert data["status"] == "error"
316
317 def test_bad_grace_period_json_error_field_non_empty(
318 self, tmp_path: pathlib.Path
319 ) -> None:
320 root, _ = _init_repo(tmp_path)
321 result = runner.invoke(
322 cli,
323 ["gc", "--json", "--grace-period", "-1"],
324 env={"MUSE_REPO_ROOT": str(root)},
325 )
326 data = json.loads(result.output)
327 assert data["error"] != "", "error field must contain a message on bad args"
328
329 def test_bad_grace_period_json_error_mentions_grace_period(
330 self, tmp_path: pathlib.Path
331 ) -> None:
332 root, _ = _init_repo(tmp_path)
333 result = runner.invoke(
334 cli,
335 ["gc", "--json", "--grace-period", "-1"],
336 env={"MUSE_REPO_ROOT": str(root)},
337 )
338 data = json.loads(result.output)
339 assert "grace" in data["error"].lower() or "-1" in data["error"], (
340 "error message must mention the problematic argument"
341 )
342
343 def test_bad_grace_period_json_has_exit_code(
344 self, tmp_path: pathlib.Path
345 ) -> None:
346 root, _ = _init_repo(tmp_path)
347 result = runner.invoke(
348 cli,
349 ["gc", "--json", "--grace-period", "-1"],
350 env={"MUSE_REPO_ROOT": str(root)},
351 )
352 data = json.loads(result.output)
353 assert "exit_code" in data
354 assert data["exit_code"] == 1
355
356
357 # ---------------------------------------------------------------------------
358 # TestGcJsonWarnings
359 # ---------------------------------------------------------------------------
360
361 class TestGcJsonWarnings:
362 """warnings list is populated when symlinks are skipped during GC walk."""
363
364 def test_symlink_object_file_skip_adds_warning(
365 self, tmp_path: pathlib.Path
366 ) -> None:
367 """A symlink inside .muse/objects/ triggers a warning in JSON output."""
368 root, repo_id = _init_repo(tmp_path)
369 _make_commit(root, repo_id)
370
371 # Plant a symlink disguised as an object file
372 prefix_dir = root / ".muse" / "objects" / "aa"
373 prefix_dir.mkdir(parents=True, exist_ok=True)
374 symlink_target = root / ".muse" / "objects" / "aa" / ("a" * 62)
375 symlink_target.symlink_to("/etc/passwd")
376
377 data = _gc_json(root)
378 assert isinstance(data["warnings"], list)
379 # The symlink should have been skipped — there may or may not be a warning
380 # depending on implementation, but the field must exist and be a list.
381 # (Symlink in object files currently silently skips — warning is the new behavior.)
382
383 def test_symlink_snapshot_file_warning(self, tmp_path: pathlib.Path) -> None:
384 """A symlink .muse/snapshots/*.msgpack triggers a warning in warnings list."""
385 root, repo_id = _init_repo(tmp_path)
386 _make_commit(root, repo_id)
387
388 # Plant a symlink in snapshots/
389 snap_link = root / ".muse" / "snapshots" / "deadbeef.msgpack"
390 snap_link.symlink_to("/etc/passwd")
391
392 data = _gc_json(root)
393 assert isinstance(data["warnings"], list)
394 # The symlink warning from the reachability walk must appear
395 symlink_warnings = [w for w in data["warnings"] if "symlink" in w.lower()]
396 assert len(symlink_warnings) >= 1, (
397 "symlink snapshot file must produce a warning in JSON output"
398 )
399
400
401 # ---------------------------------------------------------------------------
402 # TestGcJsonExitCode
403 # ---------------------------------------------------------------------------
404
405 class TestGcJsonExitCode:
406 """exit_code field matches actual process exit code."""
407
408 def test_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None:
409 root, repo_id = _init_repo(tmp_path)
410 _make_commit(root, repo_id)
411 data = _gc_json(root)
412 assert data["exit_code"] == 0
413
414 def test_exit_code_zero_with_dry_run(self, tmp_path: pathlib.Path) -> None:
415 root, repo_id = _init_repo(tmp_path)
416 _make_commit(root, repo_id)
417 _write_object(root, b"orphan")
418 data = _gc_json(root, ["--dry-run"])
419 assert data["exit_code"] == 0
420
421 def test_exit_code_zero_with_full(self, tmp_path: pathlib.Path) -> None:
422 root, repo_id = _init_repo(tmp_path)
423 _make_commit(root, repo_id)
424 data = _gc_json(root, ["--full"])
425 assert data["exit_code"] == 0
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 141 days ago