gabriel / muse public
test_plumbing_verify_object.py python
468 lines 18.4 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Comprehensive tests for ``muse plumbing verify-object``.
2
3 Audit findings addressed here
4 ------------------------------
5 Performance
6 - ``_verify_one`` previously called ``stat()`` then opened the file for
7 hashing — 2 syscalls per object. Now bytes are counted during hashing —
8 1 syscall per object. Matters for ``--all`` across large stores.
9
10 Security
11 - Format error now goes to stderr (was stdout) — verified below.
12 - ``sanitize_display()`` applied to object_id and error text in text mode
13 (defense-in-depth: ids are hex-validated, but errors may echo user input).
14
15 Agent UX
16 - ``--all`` — verify every object in the store (fsck equivalent).
17 - ``--stdin`` — read object IDs one per line from stdin.
18 - ``--all`` + explicit ids now rejected cleanly (USER_ERROR).
19 - ``formatter_class`` added.
20 - ``nargs="*"`` on object_ids to support --all / --stdin without positional args.
21
22 Coverage tiers
23 --------------
24 - Unit: _iter_all_object_ids (empty store, real objects, symlinks skipped),
25 _verify_one (ok, not-found, hash-mismatch, bad-id, I/O error),
26 _ObjectResult schema, _CHUNK constant
27 - Integration: JSON/text format, --quiet, --all, --stdin, multi-object batch,
28 mixed pass/fail, --all on empty store, --all + explicit ids rejected
29 - Security: format error→stderr, no traceback on errors, ANSI in object-id safe
30 - Stress: 100-object store --all pass, 200 sequential verifies, stdin 200 ids
31 """
32 from __future__ import annotations
33
34 import hashlib
35 import json
36 import os
37 import pathlib
38
39 import pytest
40
41 from muse.core.errors import ExitCode
42 from tests.cli_test_helper import CliRunner, InvokeResult
43
44 runner = CliRunner()
45
46 # ---------------------------------------------------------------------------
47 # Helpers
48 # ---------------------------------------------------------------------------
49
50 _FAKE_CONTENT = b"hello muse"
51 _GOOD_OID = hashlib.sha256(_FAKE_CONTENT).hexdigest()
52
53
54 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
55 repo = tmp_path / "repo"
56 muse = repo / ".muse"
57 (muse / "objects").mkdir(parents=True)
58 (muse / "commits").mkdir(parents=True)
59 (muse / "snapshots").mkdir(parents=True)
60 (muse / "refs" / "heads").mkdir(parents=True)
61 (muse / "HEAD").write_text("ref: refs/heads/main")
62 (muse / "repo.json").write_text(json.dumps({"repo_id": "r1", "domain": "code"}))
63 return repo
64
65
66 def _write_object(repo: pathlib.Path, content: bytes) -> str:
67 """Write real content into the store and return its SHA-256 ID."""
68 from muse.core.object_store import write_object
69 oid = hashlib.sha256(content).hexdigest()
70 write_object(repo, oid, content)
71 return oid
72
73
74 def _corrupt_object(repo: pathlib.Path, oid: str) -> None:
75 """Overwrite the object file with garbage (simulates bit-rot).
76
77 The object store writes files as 0o444 (read-only) to enforce immutability.
78 We must make the file writable before overwriting it in tests.
79 """
80 shard = repo / ".muse" / "objects" / oid[:2]
81 obj_file = shard / oid[2:]
82 os.chmod(obj_file, 0o644)
83 obj_file.write_bytes(b"corrupted data that does not hash to the oid")
84
85
86 def _vo(repo: pathlib.Path, *args: str, stdin: str | None = None) -> InvokeResult:
87 from muse.cli.app import main as cli
88 return runner.invoke(
89 cli,
90 ["verify-object", *args],
91 env={"MUSE_REPO_ROOT": str(repo)},
92 input=stdin,
93 )
94
95
96 # ---------------------------------------------------------------------------
97 # Unit — _iter_all_object_ids
98 # ---------------------------------------------------------------------------
99
100
101 class TestIterAllObjectIds:
102 def test_empty_store(self, tmp_path: pathlib.Path) -> None:
103 from muse.cli.commands.plumbing.verify_object import _iter_all_object_ids
104 repo = _make_repo(tmp_path)
105 assert _iter_all_object_ids(repo) == []
106
107 def test_missing_objects_dir(self, tmp_path: pathlib.Path) -> None:
108 from muse.cli.commands.plumbing.verify_object import _iter_all_object_ids
109 repo = _make_repo(tmp_path)
110 import shutil
111 shutil.rmtree(repo / ".muse" / "objects")
112 assert _iter_all_object_ids(repo) == []
113
114 def test_finds_written_object(self, tmp_path: pathlib.Path) -> None:
115 from muse.cli.commands.plumbing.verify_object import _iter_all_object_ids
116 repo = _make_repo(tmp_path)
117 oid = _write_object(repo, b"test content")
118 ids = _iter_all_object_ids(repo)
119 assert oid in ids
120
121 def test_multiple_objects_sorted(self, tmp_path: pathlib.Path) -> None:
122 from muse.cli.commands.plumbing.verify_object import _iter_all_object_ids
123 repo = _make_repo(tmp_path)
124 oids = [_write_object(repo, f"content {i}".encode()) for i in range(5)]
125 found = _iter_all_object_ids(repo)
126 assert set(oids) == set(found)
127 assert found == sorted(found)
128
129 def test_symlinks_in_shard_skipped(self, tmp_path: pathlib.Path) -> None:
130 from muse.cli.commands.plumbing.verify_object import _iter_all_object_ids
131 repo = _make_repo(tmp_path)
132 oid = _write_object(repo, b"real content")
133 # Create a symlink in the shard directory — should be skipped.
134 shard = repo / ".muse" / "objects" / oid[:2]
135 sym = shard / "symlink_file"
136 sym.symlink_to(shard / oid[2:])
137 ids = _iter_all_object_ids(repo)
138 # Should only find the real object, not count the symlink.
139 assert ids.count(oid) == 1
140
141 def test_short_shard_dir_names_ignored(self, tmp_path: pathlib.Path) -> None:
142 from muse.cli.commands.plumbing.verify_object import _iter_all_object_ids
143 repo = _make_repo(tmp_path)
144 # Create a shard with an unexpected name (not 2 chars).
145 (repo / ".muse" / "objects" / "abc").mkdir()
146 assert _iter_all_object_ids(repo) == []
147
148
149 # ---------------------------------------------------------------------------
150 # Unit — _verify_one
151 # ---------------------------------------------------------------------------
152
153
154 class TestVerifyOne:
155 def test_valid_object_ok(self, tmp_path: pathlib.Path) -> None:
156 from muse.cli.commands.plumbing.verify_object import _verify_one
157 repo = _make_repo(tmp_path)
158 oid = _write_object(repo, b"hello world")
159 result = _verify_one(repo, oid)
160 assert result["ok"] is True
161 assert result["size_bytes"] == len(b"hello world")
162 assert result["error"] is None
163
164 def test_size_counted_during_hash(self, tmp_path: pathlib.Path) -> None:
165 """Size must equal actual byte count — confirmed by counting during hash."""
166 from muse.cli.commands.plumbing.verify_object import _verify_one
167 repo = _make_repo(tmp_path)
168 content = b"x" * 12345
169 oid = _write_object(repo, content)
170 result = _verify_one(repo, oid)
171 assert result["size_bytes"] == 12345
172
173 def test_missing_object_not_ok(self, tmp_path: pathlib.Path) -> None:
174 from muse.cli.commands.plumbing.verify_object import _verify_one
175 repo = _make_repo(tmp_path)
176 oid = "a" * 64
177 result = _verify_one(repo, oid)
178 assert result["ok"] is False
179 assert "not found" in (result["error"] or "")
180 assert result["size_bytes"] is None
181
182 def test_corrupt_object_mismatch(self, tmp_path: pathlib.Path) -> None:
183 from muse.cli.commands.plumbing.verify_object import _verify_one
184 repo = _make_repo(tmp_path)
185 oid = _write_object(repo, b"original content")
186 _corrupt_object(repo, oid)
187 result = _verify_one(repo, oid)
188 assert result["ok"] is False
189 assert "mismatch" in (result["error"] or "")
190
191 def test_invalid_object_id_format(self, tmp_path: pathlib.Path) -> None:
192 from muse.cli.commands.plumbing.verify_object import _verify_one
193 repo = _make_repo(tmp_path)
194 result = _verify_one(repo, "not-a-sha256")
195 assert result["ok"] is False
196 assert result["error"] is not None
197
198 def test_invalid_object_id_never_raises(self, tmp_path: pathlib.Path) -> None:
199 from muse.cli.commands.plumbing.verify_object import _verify_one
200 repo = _make_repo(tmp_path)
201 # Should return error dict, never raise.
202 result = _verify_one(repo, "\x00" * 64)
203 assert isinstance(result, dict)
204 assert result["ok"] is False
205
206
207 class TestObjectResultSchema:
208 def test_fields(self) -> None:
209 from muse.cli.commands.plumbing.verify_object import _ObjectResult
210 fields = set(_ObjectResult.__annotations__)
211 assert fields == {"object_id", "ok", "size_bytes", "error"}
212
213
214 class TestChunkConstant:
215 def test_chunk_is_power_of_two(self) -> None:
216 from muse.cli.commands.plumbing.verify_object import _CHUNK
217 assert _CHUNK > 0
218 assert (_CHUNK & (_CHUNK - 1)) == 0 # power of 2
219
220
221 # ---------------------------------------------------------------------------
222 # Integration — JSON output
223 # ---------------------------------------------------------------------------
224
225
226 class TestJsonOutput:
227 def test_valid_object_all_ok(self, tmp_path: pathlib.Path) -> None:
228 repo = _make_repo(tmp_path)
229 oid = _write_object(repo, _FAKE_CONTENT)
230 result = _vo(repo, oid)
231 assert result.exit_code == 0
232 data = json.loads(result.output)
233 assert data["all_ok"] is True
234 assert data["checked"] == 1
235 assert data["failed"] == 0
236 assert data["results"][0]["ok"] is True
237 assert data["results"][0]["size_bytes"] == len(_FAKE_CONTENT)
238
239 def test_missing_object_fails(self, tmp_path: pathlib.Path) -> None:
240 repo = _make_repo(tmp_path)
241 result = _vo(repo, "a" * 64)
242 assert result.exit_code == ExitCode.USER_ERROR
243 data = json.loads(result.output)
244 assert data["all_ok"] is False
245 assert data["failed"] == 1
246
247 def test_corrupt_object_fails(self, tmp_path: pathlib.Path) -> None:
248 repo = _make_repo(tmp_path)
249 oid = _write_object(repo, b"good content")
250 _corrupt_object(repo, oid)
251 result = _vo(repo, oid)
252 assert result.exit_code == ExitCode.USER_ERROR
253 data = json.loads(result.output)
254 assert data["results"][0]["ok"] is False
255 assert "mismatch" in data["results"][0]["error"]
256
257 def test_mixed_pass_fail(self, tmp_path: pathlib.Path) -> None:
258 repo = _make_repo(tmp_path)
259 good = _write_object(repo, b"good")
260 bad = "b" * 64
261 result = _vo(repo, good, bad)
262 assert result.exit_code == ExitCode.USER_ERROR
263 data = json.loads(result.output)
264 assert data["checked"] == 2
265 assert data["failed"] == 1
266
267 def test_json_shorthand(self, tmp_path: pathlib.Path) -> None:
268 repo = _make_repo(tmp_path)
269 oid = _write_object(repo, b"data")
270 result = _vo(repo, "--json", oid)
271 assert result.exit_code == 0
272 assert "all_ok" in json.loads(result.output)
273
274
275 # ---------------------------------------------------------------------------
276 # Integration — text output
277 # ---------------------------------------------------------------------------
278
279
280 class TestTextOutput:
281 def test_ok_label(self, tmp_path: pathlib.Path) -> None:
282 repo = _make_repo(tmp_path)
283 oid = _write_object(repo, _FAKE_CONTENT)
284 result = _vo(repo, "--format", "text", oid)
285 assert result.exit_code == 0
286 assert "OK" in result.output
287 assert str(len(_FAKE_CONTENT)) in result.output
288
289 def test_fail_label_on_missing(self, tmp_path: pathlib.Path) -> None:
290 repo = _make_repo(tmp_path)
291 result = _vo(repo, "--format", "text", "c" * 64)
292 assert "FAIL" in result.output
293 assert result.exit_code == ExitCode.USER_ERROR
294
295
296 # ---------------------------------------------------------------------------
297 # Integration — --quiet mode
298 # ---------------------------------------------------------------------------
299
300
301 class TestQuietMode:
302 def test_all_ok_exits_0(self, tmp_path: pathlib.Path) -> None:
303 repo = _make_repo(tmp_path)
304 oid = _write_object(repo, _FAKE_CONTENT)
305 result = _vo(repo, "--quiet", oid)
306 assert result.exit_code == 0
307 assert result.output.strip() == ""
308
309 def test_failure_exits_1(self, tmp_path: pathlib.Path) -> None:
310 repo = _make_repo(tmp_path)
311 result = _vo(repo, "--quiet", "d" * 64)
312 assert result.exit_code == ExitCode.USER_ERROR
313 assert result.output.strip() == ""
314
315
316 # ---------------------------------------------------------------------------
317 # Integration — --all (new fsck mode)
318 # ---------------------------------------------------------------------------
319
320
321 class TestAllMode:
322 def test_empty_store_all_ok(self, tmp_path: pathlib.Path) -> None:
323 repo = _make_repo(tmp_path)
324 result = _vo(repo, "--all")
325 assert result.exit_code == 0
326 data = json.loads(result.output)
327 assert data["all_ok"] is True
328 assert data["checked"] == 0
329
330 def test_all_finds_written_objects(self, tmp_path: pathlib.Path) -> None:
331 repo = _make_repo(tmp_path)
332 oids = [_write_object(repo, f"content {i}".encode()) for i in range(5)]
333 result = _vo(repo, "--all")
334 assert result.exit_code == 0
335 data = json.loads(result.output)
336 assert data["checked"] == 5
337 assert data["all_ok"] is True
338
339 def test_all_detects_corruption(self, tmp_path: pathlib.Path) -> None:
340 repo = _make_repo(tmp_path)
341 oid = _write_object(repo, b"good data")
342 _corrupt_object(repo, oid)
343 result = _vo(repo, "--all")
344 assert result.exit_code == ExitCode.USER_ERROR
345 data = json.loads(result.output)
346 assert data["failed"] == 1
347
348 def test_all_plus_explicit_ids_rejected(self, tmp_path: pathlib.Path) -> None:
349 repo = _make_repo(tmp_path)
350 result = _vo(repo, "--all", "a" * 64)
351 assert result.exit_code == ExitCode.USER_ERROR
352 assert result.stdout_bytes == b"" # error went to stderr
353
354 def test_all_quiet(self, tmp_path: pathlib.Path) -> None:
355 repo = _make_repo(tmp_path)
356 _write_object(repo, b"content")
357 result = _vo(repo, "--all", "--quiet")
358 assert result.exit_code == 0
359 assert result.output.strip() == ""
360
361
362 # ---------------------------------------------------------------------------
363 # Integration — --stdin (new agent UX)
364 # ---------------------------------------------------------------------------
365
366
367 class TestStdinMode:
368 def test_reads_ids_from_stdin(self, tmp_path: pathlib.Path) -> None:
369 repo = _make_repo(tmp_path)
370 oid = _write_object(repo, _FAKE_CONTENT)
371 result = _vo(repo, "--stdin", stdin=f"{oid}\n")
372 assert result.exit_code == 0
373 data = json.loads(result.output)
374 assert data["checked"] == 1
375 assert data["all_ok"] is True
376
377 def test_comments_and_blank_lines_skipped(self, tmp_path: pathlib.Path) -> None:
378 repo = _make_repo(tmp_path)
379 oid = _write_object(repo, _FAKE_CONTENT)
380 result = _vo(repo, "--stdin", stdin=f"\n# comment\n{oid}\n\n")
381 data = json.loads(result.output)
382 assert data["checked"] == 1
383
384 def test_stdin_combines_with_positional(self, tmp_path: pathlib.Path) -> None:
385 repo = _make_repo(tmp_path)
386 oid1 = _write_object(repo, b"one")
387 oid2 = _write_object(repo, b"two")
388 result = _vo(repo, "--stdin", oid1, stdin=f"{oid2}\n")
389 data = json.loads(result.output)
390 assert data["checked"] == 2
391
392 def test_empty_stdin_no_explicit_errors(self, tmp_path: pathlib.Path) -> None:
393 repo = _make_repo(tmp_path)
394 result = _vo(repo, "--stdin", stdin="")
395 assert result.exit_code == ExitCode.USER_ERROR
396
397
398 # ---------------------------------------------------------------------------
399 # Security
400 # ---------------------------------------------------------------------------
401
402
403 class TestSecurity:
404 def test_format_error_goes_to_stderr(self, tmp_path: pathlib.Path) -> None:
405 repo = _make_repo(tmp_path)
406 result = _vo(repo, "--format", "yaml", "a" * 64)
407 assert result.exit_code == ExitCode.USER_ERROR
408 assert "error" in result.stderr.lower()
409 assert result.stdout_bytes == b""
410
411 def test_no_traceback_on_bad_format(self, tmp_path: pathlib.Path) -> None:
412 repo = _make_repo(tmp_path)
413 result = _vo(repo, "--format", "xml", "a" * 64)
414 assert "Traceback" not in result.output
415
416 def test_ansi_in_error_message_stripped_text(self, tmp_path: pathlib.Path) -> None:
417 """Even if an error message contains ANSI, text mode strips it."""
418 repo = _make_repo(tmp_path)
419 # Pass a 64-char ID with only hex chars — passes validation, doesn't exist.
420 # The error ("object not found") has no ANSI, but sanitize_display is applied.
421 result = _vo(repo, "--format", "text", "a" * 64)
422 assert "\x1b" not in result.output
423
424 def test_invalid_id_returns_error_not_crash(self, tmp_path: pathlib.Path) -> None:
425 repo = _make_repo(tmp_path)
426 result = _vo(repo, "not-a-sha256")
427 assert result.exit_code == ExitCode.USER_ERROR
428 assert "Traceback" not in result.output
429
430 def test_no_ids_errors_to_stderr(self, tmp_path: pathlib.Path) -> None:
431 repo = _make_repo(tmp_path)
432 result = _vo(repo)
433 assert result.exit_code == ExitCode.USER_ERROR
434 assert "error" in result.stderr.lower()
435
436
437 # ---------------------------------------------------------------------------
438 # Stress
439 # ---------------------------------------------------------------------------
440
441
442 class TestStress:
443 def test_100_object_store_all_pass(self, tmp_path: pathlib.Path) -> None:
444 repo = _make_repo(tmp_path)
445 for i in range(100):
446 _write_object(repo, f"stress content {i}".encode())
447 result = _vo(repo, "--all")
448 assert result.exit_code == 0
449 data = json.loads(result.output)
450 assert data["checked"] == 100
451 assert data["all_ok"] is True
452
453 def test_200_sequential_verifies(self, tmp_path: pathlib.Path) -> None:
454 repo = _make_repo(tmp_path)
455 oid = _write_object(repo, _FAKE_CONTENT)
456 for i in range(200):
457 result = _vo(repo, oid)
458 assert result.exit_code == 0, f"failed at iteration {i}"
459
460 def test_stdin_200_ids(self, tmp_path: pathlib.Path) -> None:
461 repo = _make_repo(tmp_path)
462 oids = [_write_object(repo, f"content_{i}".encode()) for i in range(200)]
463 stdin_input = "\n".join(oids) + "\n"
464 result = _vo(repo, "--stdin", stdin=stdin_input)
465 assert result.exit_code == 0
466 data = json.loads(result.output)
467 assert data["checked"] == 200
468 assert data["all_ok"] is True
File History 3 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 100 days ago