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