gabriel / muse public
test_cmd_verify.py python
445 lines 18.6 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 140 days ago
1 """Tests for ``muse verify`` and ``muse/core/verify.py``.
2
3 Covers: empty repo, healthy repo, missing commit, missing snapshot,
4 missing object, corrupted object (hash mismatch), --no-objects flag,
5 --quiet flag, --format json, stress: 100-commit chain.
6 """
7
8 from __future__ import annotations
9
10 import datetime
11 import hashlib
12 import json
13 import pathlib
14
15 import pytest
16 from tests.cli_test_helper import CliRunner
17
18 cli = None # argparse migration — CliRunner ignores this arg
19 import os
20
21 from muse.core.object_store import object_path, write_object
22 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
23 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
24 from muse.core.verify import run_verify
25 from muse.core._types import Manifest, long_id
26
27 runner = CliRunner()
28
29 _REPO_ID = "verify-test"
30
31
32 # ---------------------------------------------------------------------------
33 # Helpers
34 # ---------------------------------------------------------------------------
35
36
37 def _sha(data: bytes) -> str:
38 return long_id(hashlib.sha256(data).hexdigest())
39
40
41 def _init_repo(path: pathlib.Path) -> pathlib.Path:
42 muse = path / ".muse"
43 for d in ("commits", "snapshots", "objects", "refs/heads"):
44 (muse / d).mkdir(parents=True, exist_ok=True)
45 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
46 (muse / "repo.json").write_text(
47 json.dumps({"repo_id": _REPO_ID, "domain": "midi"}), encoding="utf-8"
48 )
49 return path
50
51
52 def _env(repo: pathlib.Path) -> Manifest:
53 return {"MUSE_REPO_ROOT": str(repo)}
54
55
56 def _make_commit(
57 root: pathlib.Path,
58 parent_id: str | None = None,
59 content: bytes = b"data",
60 branch: str = "main",
61 idx: int = 0,
62 ) -> str:
63 raw = content + str(idx).encode()
64 obj_id = _sha(raw)
65 write_object(root, obj_id, raw)
66 manifest = {f"file_{idx}.txt": obj_id}
67 snap_id = compute_snapshot_id(manifest)
68 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
69 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) + datetime.timedelta(hours=idx)
70 parent_ids = [parent_id] if parent_id else []
71 commit_id = compute_commit_id(parent_ids, snap_id, f"commit {idx}", committed_at.isoformat())
72 write_commit(root, CommitRecord(
73 commit_id=commit_id,
74 repo_id=_REPO_ID,
75 branch=branch,
76 snapshot_id=snap_id,
77 message=f"commit {idx}",
78 committed_at=committed_at,
79 parent_commit_id=parent_id,
80 ))
81 (root / ".muse" / "refs" / "heads" / branch).write_text(commit_id, encoding="utf-8")
82 return commit_id
83
84
85 # ---------------------------------------------------------------------------
86 # Unit: core run_verify
87 # ---------------------------------------------------------------------------
88
89
90 def test_verify_empty_repo_no_failures(tmp_path: pathlib.Path) -> None:
91 _init_repo(tmp_path)
92 result = run_verify(tmp_path)
93 assert result["all_ok"] is True
94 assert result["failures"] == []
95 assert result["nothing_checked"] is True
96
97
98 # ---------------------------------------------------------------------------
99 # Supercharged verify — snapshot sweep, nothing_checked, zero-byte detection
100 # ---------------------------------------------------------------------------
101
102
103 class TestVerifySupercharged:
104 """Tests for the three supercharged verify capabilities:
105
106 1. Snapshot store sweep — finds missing objects even when branch refs are absent.
107 2. nothing_checked flag — distinguishes "empty repo" from "all healthy".
108 3. Truncated objects are caught by the hash check (check_objects=True);
109 existence-only mode (check_objects=False) does not hash-verify content.
110 """
111
112 def test_nothing_checked_false_when_commits_exist(self, tmp_path: pathlib.Path) -> None:
113 _init_repo(tmp_path)
114 _make_commit(tmp_path, content=b"data", idx=0)
115 result = run_verify(tmp_path)
116 assert result["nothing_checked"] is False
117
118 def test_nothing_checked_true_when_no_refs_and_no_snapshots(self, tmp_path: pathlib.Path) -> None:
119 _init_repo(tmp_path)
120 result = run_verify(tmp_path)
121 assert result["nothing_checked"] is True
122
123 def test_orphan_snapshot_with_missing_object_detected(self, tmp_path: pathlib.Path) -> None:
124 """Snapshot exists in .muse/snapshots/ but no commit or branch ref points to it.
125 Its objects are missing. Verify should catch this via the snapshot store sweep."""
126 _init_repo(tmp_path)
127 obj_id = long_id("a" * 64) # non-existent object
128 manifest = {"orphan.py": obj_id}
129 snap_id = compute_snapshot_id(manifest)
130 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
131 # No branch ref written, no commit written.
132
133 result = run_verify(tmp_path)
134 assert result["all_ok"] is False
135 assert any(f["kind"] == "object" and f["id"] == obj_id for f in result["failures"])
136 assert result["nothing_checked"] is False # sweep found something to check
137
138 def test_orphan_snapshot_with_present_object_passes(self, tmp_path: pathlib.Path) -> None:
139 """Orphan snapshot whose object IS present should not cause failures."""
140 _init_repo(tmp_path)
141 content = b"orphan content"
142 obj_id = _sha(content)
143 write_object(tmp_path, obj_id, content)
144 manifest = {"file.py": obj_id}
145 snap_id = compute_snapshot_id(manifest)
146 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
147
148 result = run_verify(tmp_path)
149 assert result["all_ok"] is True
150 assert result["nothing_checked"] is False # sweep found the snapshot
151
152 def test_partial_clone_missing_objects_detected(self, tmp_path: pathlib.Path) -> None:
153 """Simulate a failed clone: commits and snapshots written to store,
154 but the branch ref file was never created and objects are absent.
155 Verify must detect the missing objects via the snapshot sweep."""
156 import datetime
157 _init_repo(tmp_path)
158 obj_id = _sha(b"important file content")
159 manifest = {"src/main.py": obj_id}
160 snap_id = compute_snapshot_id(manifest)
161 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
162 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
163 commit_id = compute_commit_id([], snap_id, "partial clone", committed_at.isoformat())
164 write_commit(tmp_path, CommitRecord(
165 commit_id=commit_id,
166 repo_id=_REPO_ID,
167 branch="main",
168 snapshot_id=snap_id,
169 message="partial clone",
170 committed_at=committed_at,
171 ))
172 # Critically: the branch ref file is NOT written (simulates clone crash).
173 # The object is also NOT written (simulates R2 gap).
174
175 result = run_verify(tmp_path)
176 assert result["all_ok"] is False
177 object_failures = [f for f in result["failures"] if f["kind"] == "object"]
178 assert any(f["id"] == obj_id for f in object_failures)
179
180 def test_truncated_object_caught_by_hash_check(self, tmp_path: pathlib.Path) -> None:
181 """An object file truncated to empty bytes is caught as a hash mismatch
182 when check_objects=True. Empty bytes have OID sha256:e3b0c44… which
183 differs from the stored OID unless the file was always empty."""
184 import os as _os
185 _init_repo(tmp_path)
186 content = b"real content here"
187 obj_id = _sha(content)
188 write_object(tmp_path, obj_id, content)
189 manifest = {"real.py": obj_id}
190 snap_id = compute_snapshot_id(manifest)
191 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
192 committed_at = datetime.datetime(2026, 4, 1, tzinfo=datetime.timezone.utc)
193 commit_id = compute_commit_id([], snap_id, "truncated test", committed_at.isoformat())
194 write_commit(tmp_path, CommitRecord(
195 commit_id=commit_id, repo_id=_REPO_ID, branch="main",
196 snapshot_id=snap_id, message="truncated test", committed_at=committed_at,
197 ))
198 (tmp_path / ".muse" / "refs" / "heads" / "main").write_text(commit_id)
199
200 # Simulate truncation (e.g. R2 serving empty body for a non-empty OID).
201 obj_file = object_path(tmp_path, obj_id)
202 _os.chmod(obj_file, 0o644)
203 obj_file.write_bytes(b"")
204
205 # Hash check catches the mismatch.
206 result = run_verify(tmp_path, check_objects=True)
207 assert result["all_ok"] is False
208 assert any(f["kind"] == "object" and f["id"] == obj_id for f in result["failures"])
209
210 def test_truncated_object_passes_existence_check(self, tmp_path: pathlib.Path) -> None:
211 """check_objects=False only verifies the object file exists — it does not
212 re-hash. A truncated file passes existence-only mode."""
213 import os as _os
214 _init_repo(tmp_path)
215 content = b"real content here"
216 obj_id = _sha(content)
217 write_object(tmp_path, obj_id, content)
218 manifest = {"real.py": obj_id}
219 snap_id = compute_snapshot_id(manifest)
220 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
221 committed_at = datetime.datetime(2026, 4, 1, tzinfo=datetime.timezone.utc)
222 commit_id = compute_commit_id([], snap_id, "existence test", committed_at.isoformat())
223 write_commit(tmp_path, CommitRecord(
224 commit_id=commit_id, repo_id=_REPO_ID, branch="main",
225 snapshot_id=snap_id, message="existence test", committed_at=committed_at,
226 ))
227 (tmp_path / ".muse" / "refs" / "heads" / "main").write_text(commit_id)
228
229 obj_file = object_path(tmp_path, obj_id)
230 _os.chmod(obj_file, 0o644)
231 obj_file.write_bytes(b"")
232
233 result = run_verify(tmp_path, check_objects=False)
234 assert result["all_ok"] is True
235
236 def test_genuinely_empty_file_passes_hash_check(self, tmp_path: pathlib.Path) -> None:
237 """A file whose content is genuinely empty bytes has OID sha256:e3b0c44…
238 The object file is zero bytes and the hash check must pass — empty is valid."""
239 _init_repo(tmp_path)
240 content = b""
241 obj_id = _sha(content) # sha256:e3b0c44...
242 write_object(tmp_path, obj_id, content)
243 manifest = {"__init__.py": obj_id}
244 snap_id = compute_snapshot_id(manifest)
245 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
246 committed_at = datetime.datetime(2026, 4, 3, tzinfo=datetime.timezone.utc)
247 commit_id = compute_commit_id([], snap_id, "empty file test", committed_at.isoformat())
248 write_commit(tmp_path, CommitRecord(
249 commit_id=commit_id, repo_id=_REPO_ID, branch="main",
250 snapshot_id=snap_id, message="empty file test", committed_at=committed_at,
251 ))
252 (tmp_path / ".muse" / "refs" / "heads" / "main").write_text(commit_id)
253
254 result = run_verify(tmp_path, check_objects=True)
255 assert result["all_ok"] is True, f"Failures: {result['failures']}"
256
257 def test_truncated_object_reported_exactly_once(self, tmp_path: pathlib.Path) -> None:
258 """A truncated object should appear exactly once in failures — the hash
259 mismatch check, not duplicated by any secondary check."""
260 import os as _os
261 _init_repo(tmp_path)
262 content = b"will be truncated"
263 obj_id = _sha(content)
264 write_object(tmp_path, obj_id, content)
265 manifest = {"f.py": obj_id}
266 snap_id = compute_snapshot_id(manifest)
267 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
268 committed_at = datetime.datetime(2026, 4, 2, tzinfo=datetime.timezone.utc)
269 commit_id = compute_commit_id([], snap_id, "dup test", committed_at.isoformat())
270 write_commit(tmp_path, CommitRecord(
271 commit_id=commit_id, repo_id=_REPO_ID, branch="main",
272 snapshot_id=snap_id, message="dup test", committed_at=committed_at,
273 ))
274 (tmp_path / ".muse" / "refs" / "heads" / "main").write_text(commit_id)
275
276 obj_file = object_path(tmp_path, obj_id)
277 _os.chmod(obj_file, 0o644)
278 obj_file.write_bytes(b"")
279
280 result = run_verify(tmp_path, check_objects=True)
281 matching = [f for f in result["failures"] if f["id"] == obj_id]
282 assert len(matching) == 1, f"Expected 1 failure for {obj_id[:12]}, got {len(matching)}"
283
284 def test_snapshot_sweep_does_not_recheck_already_verified(self, tmp_path: pathlib.Path) -> None:
285 """Snapshots reachable from branch refs should not be double-counted
286 by the orphan sweep pass."""
287 _init_repo(tmp_path)
288 commit_id = _make_commit(tmp_path, content=b"data", idx=0)
289 result = run_verify(tmp_path)
290 assert result["snapshots_checked"] == 1 # not 2
291
292 def test_json_output_includes_nothing_checked(self, tmp_path: pathlib.Path) -> None:
293 """The --json output must include nothing_checked so scripts can distinguish
294 empty repos from healthy ones."""
295 _init_repo(tmp_path)
296 result = runner.invoke(cli, ["verify", "--json"], env=_env(tmp_path))
297 assert result.exit_code == 0
298 data = json.loads(result.output)
299 assert "nothing_checked" in data
300 assert data["nothing_checked"] is True
301
302
303 def test_verify_healthy_repo(tmp_path: pathlib.Path) -> None:
304 _init_repo(tmp_path)
305 _make_commit(tmp_path, content=b"healthy", idx=0)
306 result = run_verify(tmp_path)
307 assert result["all_ok"] is True
308 assert result["commits_checked"] == 1
309 assert result["objects_checked"] >= 1
310
311
312 def test_verify_missing_commit_fails(tmp_path: pathlib.Path) -> None:
313 _init_repo(tmp_path)
314 # Write a ref pointing to a nonexistent commit.
315 fake_id = "a" * 64
316 (tmp_path / ".muse" / "refs" / "heads" / "main").write_text(fake_id, encoding="utf-8")
317 result = run_verify(tmp_path)
318 assert result["all_ok"] is False
319 kinds = [f["kind"] for f in result["failures"]]
320 assert "commit" in kinds
321
322
323 def test_verify_corrupted_object_detected(tmp_path: pathlib.Path) -> None:
324 _init_repo(tmp_path)
325 content = b"original content"
326 obj_id = _sha(content)
327 write_object(tmp_path, obj_id, content)
328 manifest = {"file.txt": obj_id}
329 snap_id = compute_snapshot_id(manifest)
330 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
331 committed_at = datetime.datetime(2026, 3, 1, tzinfo=datetime.timezone.utc)
332 commit_id = compute_commit_id([], snap_id, "corrupt test", committed_at.isoformat())
333 write_commit(tmp_path, CommitRecord(
334 commit_id=commit_id,
335 repo_id=_REPO_ID,
336 branch="main",
337 snapshot_id=snap_id,
338 message="corrupt test",
339 committed_at=committed_at,
340 ))
341 (tmp_path / ".muse" / "refs" / "heads" / "main").write_text(commit_id, encoding="utf-8")
342
343 # Object store writes files as 0o444 (immutable) — chmod before corrupting.
344 obj_file = object_path(tmp_path, obj_id)
345 os.chmod(obj_file, 0o644)
346 obj_file.write_bytes(b"tampered data!")
347
348 result = run_verify(tmp_path, check_objects=True)
349 assert result["all_ok"] is False
350 kinds = [f["kind"] for f in result["failures"]]
351 assert "object" in kinds
352
353
354 def test_verify_no_objects_flag_skips_rehash(tmp_path: pathlib.Path) -> None:
355 _init_repo(tmp_path)
356 content = b"clean"
357 obj_id = _sha(content)
358 write_object(tmp_path, obj_id, content)
359 manifest = {"f.txt": obj_id}
360 snap_id = compute_snapshot_id(manifest)
361 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
362 committed_at = datetime.datetime(2026, 3, 2, tzinfo=datetime.timezone.utc)
363 commit_id = compute_commit_id([], snap_id, "test", committed_at.isoformat())
364 write_commit(tmp_path, CommitRecord(
365 commit_id=commit_id, repo_id=_REPO_ID, branch="main",
366 snapshot_id=snap_id, message="test", committed_at=committed_at,
367 ))
368 (tmp_path / ".muse" / "refs" / "heads" / "main").write_text(commit_id, encoding="utf-8")
369
370 # Object store writes files as 0o444 (immutable) — chmod before corrupting.
371 obj_file = object_path(tmp_path, obj_id)
372 os.chmod(obj_file, 0o644)
373 obj_file.write_bytes(b"corrupted!")
374
375 result = run_verify(tmp_path, check_objects=False)
376 # Should not flag the corruption since we skipped re-hashing.
377 assert result["all_ok"] is True
378
379
380 # ---------------------------------------------------------------------------
381 # CLI: muse verify
382 # ---------------------------------------------------------------------------
383
384
385 def test_verify_cli_help() -> None:
386 result = runner.invoke(cli, ["verify", "--help"])
387 assert result.exit_code == 0
388 # Rich injects ANSI codes between '--' dashes; the short flag '-O' is reliable.
389 assert "--no-objects" in result.output or "-O" in result.output
390
391
392 def test_verify_cli_healthy(tmp_path: pathlib.Path) -> None:
393 _init_repo(tmp_path)
394 _make_commit(tmp_path, content=b"cli healthy", idx=99)
395 result = runner.invoke(cli, ["verify"], env=_env(tmp_path))
396 assert result.exit_code == 0
397 assert "healthy" in result.output.lower()
398
399
400 def test_verify_cli_json(tmp_path: pathlib.Path) -> None:
401 _init_repo(tmp_path)
402 _make_commit(tmp_path, content=b"json verify", idx=88)
403 result = runner.invoke(cli, ["verify", "--json"], env=_env(tmp_path))
404 assert result.exit_code == 0
405 data = json.loads(result.output)
406 assert data["all_ok"] is True
407 assert data["failures"] == []
408
409
410 def test_verify_cli_quiet_exit_zero_when_clean(tmp_path: pathlib.Path) -> None:
411 _init_repo(tmp_path)
412 _make_commit(tmp_path, content=b"quiet clean", idx=77)
413 result = runner.invoke(cli, ["verify", "--quiet"], env=_env(tmp_path))
414 assert result.exit_code == 0
415
416
417 def test_verify_cli_quiet_exit_one_when_broken(tmp_path: pathlib.Path) -> None:
418 _init_repo(tmp_path)
419 fake_id = "b" * 64
420 (tmp_path / ".muse" / "refs" / "heads" / "main").write_text(fake_id, encoding="utf-8")
421 result = runner.invoke(cli, ["verify", "-q"], env=_env(tmp_path))
422 assert result.exit_code != 0
423
424
425 def test_verify_cli_no_objects_flag(tmp_path: pathlib.Path) -> None:
426 _init_repo(tmp_path)
427 _make_commit(tmp_path, content=b"no-obj flag", idx=66)
428 result = runner.invoke(cli, ["verify", "--no-objects"], env=_env(tmp_path))
429 assert result.exit_code == 0
430
431
432 # ---------------------------------------------------------------------------
433 # Stress: 100-commit chain
434 # ---------------------------------------------------------------------------
435
436
437 def test_verify_stress_100_commit_chain(tmp_path: pathlib.Path) -> None:
438 _init_repo(tmp_path)
439 prev: str | None = None
440 for i in range(100):
441 prev = _make_commit(tmp_path, parent_id=prev, content=b"chain", idx=i)
442
443 result = run_verify(tmp_path, check_objects=True)
444 assert result["all_ok"] is True
445 assert result["commits_checked"] == 100
File History 2 commits
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 140 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor 143 days ago