gabriel / muse public
test_cmd_maintenance.py python
319 lines 11.6 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 142 days ago
1 """Tests for ``muse maintenance`` — scheduled store maintenance orchestration.
2
3 Coverage tiers:
4 - Unit: run records timestamp; run --task gc; run --task verify-objects;
5 run --all; run --dry-run; run --json schema; status text + JSON;
6 schedule --period-hours; schedule --enable/--disable;
7 status reflects schedule config; no-config defaults
8 - Integration: verify-objects detects corrupt objects; gc cleans unreachable
9 blobs; json result keys correct; dry-run produces no mutations
10 - Security: no ANSI injection in task output
11 - Stress: verify-objects on 100 objects completes and counts correctly
12 """
13
14 from __future__ import annotations
15
16 import datetime
17 import hashlib
18 import json
19 import pathlib
20 import time
21 import unittest.mock
22
23 import pytest
24
25 from tests.cli_test_helper import CliRunner
26 from muse.core.object_store import object_path, write_object
27 from muse.core.snapshot import compute_snapshot_id
28 from muse.core.store import SnapshotRecord, write_snapshot
29 from muse.core._types import Manifest, long_id
30
31 runner = CliRunner()
32
33 _REPO_ID = "maintenance-test"
34
35
36 # ---------------------------------------------------------------------------
37 # Helpers
38 # ---------------------------------------------------------------------------
39
40
41 def _sha(data: bytes) -> str:
42 return long_id(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 _invoke(args: list[str], repo: pathlib.Path):
61 result = runner.invoke(None, args, env=_env(repo))
62 return result.exit_code, result.stdout, result.stderr
63
64
65 def _maint_config(repo: pathlib.Path) -> pathlib.Path:
66 return repo / ".muse" / "maintenance.json"
67
68
69 def _write_object(repo: pathlib.Path, content: bytes) -> str:
70 obj_id = _sha(content)
71 write_object(repo, obj_id, content)
72 return obj_id
73
74
75 # ---------------------------------------------------------------------------
76 # Unit — run (default)
77 # ---------------------------------------------------------------------------
78
79
80 class TestRun:
81 def test_run_exits_zero(self, tmp_path):
82 repo = _init_repo(tmp_path / "repo")
83 rc, out, err = _invoke(["maintenance", "run"], repo)
84 assert rc == 0
85
86 def test_run_records_timestamp(self, tmp_path):
87 repo = _init_repo(tmp_path / "repo")
88 _invoke(["maintenance", "run"], repo)
89 cfg = json.loads(_maint_config(repo).read_text())
90 assert "last_run" in cfg
91 assert "gc" in cfg["last_run"]
92
93 def test_run_creates_config_if_missing(self, tmp_path):
94 repo = _init_repo(tmp_path / "repo")
95 assert not _maint_config(repo).exists()
96 _invoke(["maintenance", "run"], repo)
97 assert _maint_config(repo).exists()
98
99 def test_run_task_gc(self, tmp_path):
100 repo = _init_repo(tmp_path / "repo")
101 rc, out, err = _invoke(["maintenance", "run", "--task", "gc"], repo)
102 assert rc == 0
103 cfg = json.loads(_maint_config(repo).read_text())
104 assert "gc" in cfg["last_run"]
105
106 def test_run_task_verify_objects(self, tmp_path):
107 repo = _init_repo(tmp_path / "repo")
108 _write_object(repo, b"hello")
109 rc, out, err = _invoke(
110 ["maintenance", "run", "--task", "verify-objects"], repo
111 )
112 assert rc == 0
113 cfg = json.loads(_maint_config(repo).read_text())
114 assert "verify-objects" in cfg["last_run"]
115
116 def test_run_all(self, tmp_path):
117 repo = _init_repo(tmp_path / "repo")
118 rc, out, err = _invoke(["maintenance", "run", "--all"], repo)
119 assert rc == 0
120 cfg = json.loads(_maint_config(repo).read_text())
121 assert "gc" in cfg["last_run"]
122 assert "verify-objects" in cfg["last_run"]
123
124 def test_run_dry_run_no_config_written(self, tmp_path):
125 repo = _init_repo(tmp_path / "repo")
126 _invoke(["maintenance", "run", "--dry-run"], repo)
127 # dry-run should NOT persist timestamps
128 if _maint_config(repo).exists():
129 cfg = json.loads(_maint_config(repo).read_text())
130 assert "last_run" not in cfg or not cfg.get("last_run")
131
132 def test_run_json_schema(self, tmp_path):
133 repo = _init_repo(tmp_path / "repo")
134 rc, out, err = _invoke(["maintenance", "run", "--json"], repo)
135 assert rc == 0
136 data = json.loads(out)
137 assert "tasks_run" in data
138 assert "results" in data
139 assert "dry_run" in data
140 assert "duration_ms" in data
141
142 def test_run_json_tasks_run_is_list(self, tmp_path):
143 repo = _init_repo(tmp_path / "repo")
144 rc, out, err = _invoke(
145 ["maintenance", "run", "--task", "gc", "--json"], repo
146 )
147 data = json.loads(out)
148 assert isinstance(data["tasks_run"], list)
149 assert "gc" in data["tasks_run"]
150
151 def test_run_unknown_task_exits_nonzero(self, tmp_path):
152 repo = _init_repo(tmp_path / "repo")
153 rc, out, err = _invoke(
154 ["maintenance", "run", "--task", "bogus-task"], repo
155 )
156 assert rc != 0
157
158 def test_run_dry_run_flag_in_json(self, tmp_path):
159 repo = _init_repo(tmp_path / "repo")
160 rc, out, err = _invoke(
161 ["maintenance", "run", "--dry-run", "--json"], repo
162 )
163 data = json.loads(out)
164 assert data["dry_run"] is True
165
166
167 # ---------------------------------------------------------------------------
168 # Unit — status
169 # ---------------------------------------------------------------------------
170
171
172 class TestStatus:
173 def test_status_no_config(self, tmp_path):
174 repo = _init_repo(tmp_path / "repo")
175 rc, out, err = _invoke(["maintenance", "status"], repo)
176 assert rc == 0
177 assert "never" in out.lower() or "no" in out.lower() or "disabled" in out.lower()
178
179 def test_status_json_no_config(self, tmp_path):
180 repo = _init_repo(tmp_path / "repo")
181 rc, out, err = _invoke(["maintenance", "status", "--json"], repo)
182 assert rc == 0
183 data = json.loads(out)
184 assert "enabled" in data
185 assert "period_hours" in data
186 assert "last_run" in data
187
188 def test_status_shows_last_run_after_run(self, tmp_path):
189 repo = _init_repo(tmp_path / "repo")
190 _invoke(["maintenance", "run", "--task", "gc"], repo)
191 rc, out, err = _invoke(["maintenance", "status"], repo)
192 assert rc == 0
193 assert "gc" in out.lower()
194
195 def test_status_json_has_last_run_timestamp(self, tmp_path):
196 repo = _init_repo(tmp_path / "repo")
197 _invoke(["maintenance", "run", "--task", "gc"], repo)
198 rc, out, err = _invoke(["maintenance", "status", "--json"], repo)
199 data = json.loads(out)
200 assert "gc" in data["last_run"]
201 # Timestamp should be ISO 8601
202 ts = data["last_run"]["gc"]
203 assert "T" in ts or ts is None
204
205
206 # ---------------------------------------------------------------------------
207 # Unit — schedule
208 # ---------------------------------------------------------------------------
209
210
211 class TestSchedule:
212 def test_schedule_period_hours(self, tmp_path):
213 repo = _init_repo(tmp_path / "repo")
214 rc, out, err = _invoke(
215 ["maintenance", "schedule", "--period-hours", "48"], repo
216 )
217 assert rc == 0
218 cfg = json.loads(_maint_config(repo).read_text())
219 assert cfg["period_hours"] == 48
220
221 def test_schedule_disable(self, tmp_path):
222 repo = _init_repo(tmp_path / "repo")
223 _invoke(["maintenance", "schedule", "--period-hours", "24"], repo)
224 rc, out, err = _invoke(["maintenance", "schedule", "--disable"], repo)
225 assert rc == 0
226 cfg = json.loads(_maint_config(repo).read_text())
227 assert cfg["enabled"] is False
228
229 def test_schedule_enable(self, tmp_path):
230 repo = _init_repo(tmp_path / "repo")
231 _invoke(["maintenance", "schedule", "--disable"], repo)
232 rc, out, err = _invoke(["maintenance", "schedule", "--enable"], repo)
233 assert rc == 0
234 cfg = json.loads(_maint_config(repo).read_text())
235 assert cfg["enabled"] is True
236
237 def test_schedule_status_reflects_period(self, tmp_path):
238 repo = _init_repo(tmp_path / "repo")
239 _invoke(["maintenance", "schedule", "--period-hours", "72"], repo)
240 rc, out, err = _invoke(["maintenance", "status", "--json"], repo)
241 data = json.loads(out)
242 assert data["period_hours"] == 72
243
244 def test_schedule_invalid_period_exits_nonzero(self, tmp_path):
245 repo = _init_repo(tmp_path / "repo")
246 rc, out, err = _invoke(
247 ["maintenance", "schedule", "--period-hours", "-1"], repo
248 )
249 assert rc != 0
250
251 def test_schedule_default_period_is_24(self, tmp_path):
252 repo = _init_repo(tmp_path / "repo")
253 _invoke(["maintenance", "schedule"], repo)
254 cfg = json.loads(_maint_config(repo).read_text())
255 assert cfg.get("period_hours", 24) == 24
256
257
258 # ---------------------------------------------------------------------------
259 # Integration — verify-objects detects corruption
260 # ---------------------------------------------------------------------------
261
262
263 class TestVerifyObjectsIntegration:
264 def test_verify_objects_passes_on_clean_store(self, tmp_path):
265 repo = _init_repo(tmp_path / "repo")
266 for i in range(5):
267 _write_object(repo, f"content-{i}".encode())
268 rc, out, err = _invoke(
269 ["maintenance", "run", "--task", "verify-objects", "--json"], repo
270 )
271 assert rc == 0
272 data = json.loads(out)
273 assert data["results"]["verify-objects"]["failed"] == 0
274
275 def test_verify_objects_detects_corrupt_object(self, tmp_path):
276 import os
277 repo = _init_repo(tmp_path / "repo")
278 obj_id = _write_object(repo, b"good content")
279 # corrupt the file (objects are stored read-only, chmod first)
280 obj_path = object_path(repo, obj_id)
281 obj_path.chmod(0o644)
282 obj_path.write_bytes(b"corrupted data")
283
284 rc, out, err = _invoke(
285 ["maintenance", "run", "--task", "verify-objects", "--json"], repo
286 )
287 # Should still exit 0 but report failures
288 data = json.loads(out)
289 assert data["results"]["verify-objects"]["failed"] >= 1
290
291 def test_verify_objects_json_has_checked_count(self, tmp_path):
292 repo = _init_repo(tmp_path / "repo")
293 for i in range(3):
294 _write_object(repo, f"item-{i}".encode())
295 rc, out, err = _invoke(
296 ["maintenance", "run", "--task", "verify-objects", "--json"], repo
297 )
298 data = json.loads(out)
299 assert data["results"]["verify-objects"]["checked"] == 3
300
301
302 # ---------------------------------------------------------------------------
303 # Stress — verify-objects on 100 objects
304 # ---------------------------------------------------------------------------
305
306
307 class TestStress:
308 def test_100_objects_verify_all_pass(self, tmp_path):
309 repo = _init_repo(tmp_path / "repo")
310 for i in range(100):
311 _write_object(repo, f"stress-object-{i:03d}".encode())
312 rc, out, err = _invoke(
313 ["maintenance", "run", "--task", "verify-objects", "--json"], repo
314 )
315 assert rc == 0
316 data = json.loads(out)
317 v = data["results"]["verify-objects"]
318 assert v["checked"] == 100
319 assert v["failed"] == 0
File History 2 commits
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 142 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 145 days ago