"""Comprehensive tests for ``muse plumbing verify-object``. Audit findings addressed here ------------------------------ Performance - ``_verify_one`` previously called ``stat()`` then opened the file for hashing — 2 syscalls per object. Now bytes are counted during hashing — 1 syscall per object. Matters for ``--all`` across large stores. Security - Format error now goes to stderr (was stdout) — verified below. - ``sanitize_display()`` applied to object_id and error text in text mode (defense-in-depth: ids are hex-validated, but errors may echo user input). Agent UX - ``--all`` — verify every object in the store (fsck equivalent). - ``--stdin`` — read object IDs one per line from stdin. - ``--all`` + explicit ids now rejected cleanly (USER_ERROR). - ``formatter_class`` added. - ``nargs="*"`` on object_ids to support --all / --stdin without positional args. Coverage tiers -------------- - Unit: _iter_all_object_ids (empty store, real objects, symlinks skipped), _verify_one (ok, not-found, hash-mismatch, bad-id, I/O error), _ObjectResult schema, _CHUNK constant - Integration: JSON/text format, --quiet, --all, --stdin, multi-object batch, mixed pass/fail, --all on empty store, --all + explicit ids rejected - Security: format error→stderr, no traceback on errors, ANSI in object-id safe - Stress: 100-object store --all pass, 200 sequential verifies, stdin 200 ids """ from __future__ import annotations import hashlib import json import os import pathlib import pytest from muse.core.errors import ExitCode from tests.cli_test_helper import CliRunner, InvokeResult runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _FAKE_CONTENT = b"hello muse" _GOOD_OID = hashlib.sha256(_FAKE_CONTENT).hexdigest() def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: repo = tmp_path / "repo" muse = repo / ".muse" (muse / "objects").mkdir(parents=True) (muse / "commits").mkdir(parents=True) (muse / "snapshots").mkdir(parents=True) (muse / "refs" / "heads").mkdir(parents=True) (muse / "HEAD").write_text("ref: refs/heads/main") (muse / "repo.json").write_text(json.dumps({"repo_id": "r1", "domain": "code"})) return repo def _write_object(repo: pathlib.Path, content: bytes) -> str: """Write real content into the store and return its SHA-256 ID.""" from muse.core.object_store import write_object oid = hashlib.sha256(content).hexdigest() write_object(repo, oid, content) return oid def _corrupt_object(repo: pathlib.Path, oid: str) -> None: """Overwrite the object file with garbage (simulates bit-rot). The object store writes files as 0o444 (read-only) to enforce immutability. We must make the file writable before overwriting it in tests. """ shard = repo / ".muse" / "objects" / oid[:2] obj_file = shard / oid[2:] os.chmod(obj_file, 0o644) obj_file.write_bytes(b"corrupted data that does not hash to the oid") def _vo(repo: pathlib.Path, *args: str, stdin: str | None = None) -> InvokeResult: from muse.cli.app import main as cli return runner.invoke( cli, ["verify-object", *args], env={"MUSE_REPO_ROOT": str(repo)}, input=stdin, ) # --------------------------------------------------------------------------- # Unit — _iter_all_object_ids # --------------------------------------------------------------------------- class TestIterAllObjectIds: def test_empty_store(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.plumbing.verify_object import _iter_all_object_ids repo = _make_repo(tmp_path) assert _iter_all_object_ids(repo) == [] def test_missing_objects_dir(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.plumbing.verify_object import _iter_all_object_ids repo = _make_repo(tmp_path) import shutil shutil.rmtree(repo / ".muse" / "objects") assert _iter_all_object_ids(repo) == [] def test_finds_written_object(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.plumbing.verify_object import _iter_all_object_ids repo = _make_repo(tmp_path) oid = _write_object(repo, b"test content") ids = _iter_all_object_ids(repo) assert oid in ids def test_multiple_objects_sorted(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.plumbing.verify_object import _iter_all_object_ids repo = _make_repo(tmp_path) oids = [_write_object(repo, f"content {i}".encode()) for i in range(5)] found = _iter_all_object_ids(repo) assert set(oids) == set(found) assert found == sorted(found) def test_symlinks_in_shard_skipped(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.plumbing.verify_object import _iter_all_object_ids repo = _make_repo(tmp_path) oid = _write_object(repo, b"real content") # Create a symlink in the shard directory — should be skipped. shard = repo / ".muse" / "objects" / oid[:2] sym = shard / "symlink_file" sym.symlink_to(shard / oid[2:]) ids = _iter_all_object_ids(repo) # Should only find the real object, not count the symlink. assert ids.count(oid) == 1 def test_short_shard_dir_names_ignored(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.plumbing.verify_object import _iter_all_object_ids repo = _make_repo(tmp_path) # Create a shard with an unexpected name (not 2 chars). (repo / ".muse" / "objects" / "abc").mkdir() assert _iter_all_object_ids(repo) == [] # --------------------------------------------------------------------------- # Unit — _verify_one # --------------------------------------------------------------------------- class TestVerifyOne: def test_valid_object_ok(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.plumbing.verify_object import _verify_one repo = _make_repo(tmp_path) oid = _write_object(repo, b"hello world") result = _verify_one(repo, oid) assert result["ok"] is True assert result["size_bytes"] == len(b"hello world") assert result["error"] is None def test_size_counted_during_hash(self, tmp_path: pathlib.Path) -> None: """Size must equal actual byte count — confirmed by counting during hash.""" from muse.cli.commands.plumbing.verify_object import _verify_one repo = _make_repo(tmp_path) content = b"x" * 12345 oid = _write_object(repo, content) result = _verify_one(repo, oid) assert result["size_bytes"] == 12345 def test_missing_object_not_ok(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.plumbing.verify_object import _verify_one repo = _make_repo(tmp_path) oid = "a" * 64 result = _verify_one(repo, oid) assert result["ok"] is False assert "not found" in (result["error"] or "") assert result["size_bytes"] is None def test_corrupt_object_mismatch(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.plumbing.verify_object import _verify_one repo = _make_repo(tmp_path) oid = _write_object(repo, b"original content") _corrupt_object(repo, oid) result = _verify_one(repo, oid) assert result["ok"] is False assert "mismatch" in (result["error"] or "") def test_invalid_object_id_format(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.plumbing.verify_object import _verify_one repo = _make_repo(tmp_path) result = _verify_one(repo, "not-a-sha256") assert result["ok"] is False assert result["error"] is not None def test_invalid_object_id_never_raises(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.plumbing.verify_object import _verify_one repo = _make_repo(tmp_path) # Should return error dict, never raise. result = _verify_one(repo, "\x00" * 64) assert isinstance(result, dict) assert result["ok"] is False class TestObjectResultSchema: def test_fields(self) -> None: from muse.cli.commands.plumbing.verify_object import _ObjectResult fields = set(_ObjectResult.__annotations__) assert fields == {"object_id", "ok", "size_bytes", "error"} class TestChunkConstant: def test_chunk_is_power_of_two(self) -> None: from muse.cli.commands.plumbing.verify_object import _CHUNK assert _CHUNK > 0 assert (_CHUNK & (_CHUNK - 1)) == 0 # power of 2 # --------------------------------------------------------------------------- # Integration — JSON output # --------------------------------------------------------------------------- class TestJsonOutput: def test_valid_object_all_ok(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_object(repo, _FAKE_CONTENT) result = _vo(repo, oid) assert result.exit_code == 0 data = json.loads(result.output) assert data["all_ok"] is True assert data["checked"] == 1 assert data["failed"] == 0 assert data["results"][0]["ok"] is True assert data["results"][0]["size_bytes"] == len(_FAKE_CONTENT) def test_missing_object_fails(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _vo(repo, "a" * 64) assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert data["all_ok"] is False assert data["failed"] == 1 def test_corrupt_object_fails(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_object(repo, b"good content") _corrupt_object(repo, oid) result = _vo(repo, oid) assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert data["results"][0]["ok"] is False assert "mismatch" in data["results"][0]["error"] def test_mixed_pass_fail(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) good = _write_object(repo, b"good") bad = "b" * 64 result = _vo(repo, good, bad) assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert data["checked"] == 2 assert data["failed"] == 1 def test_json_shorthand(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_object(repo, b"data") result = _vo(repo, "--json", oid) assert result.exit_code == 0 assert "all_ok" in json.loads(result.output) # --------------------------------------------------------------------------- # Integration — text output # --------------------------------------------------------------------------- class TestTextOutput: def test_ok_label(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_object(repo, _FAKE_CONTENT) result = _vo(repo, "--format", "text", oid) assert result.exit_code == 0 assert "OK" in result.output assert str(len(_FAKE_CONTENT)) in result.output def test_fail_label_on_missing(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _vo(repo, "--format", "text", "c" * 64) assert "FAIL" in result.output assert result.exit_code == ExitCode.USER_ERROR # --------------------------------------------------------------------------- # Integration — --quiet mode # --------------------------------------------------------------------------- class TestQuietMode: def test_all_ok_exits_0(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_object(repo, _FAKE_CONTENT) result = _vo(repo, "--quiet", oid) assert result.exit_code == 0 assert result.output.strip() == "" def test_failure_exits_1(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _vo(repo, "--quiet", "d" * 64) assert result.exit_code == ExitCode.USER_ERROR assert result.output.strip() == "" # --------------------------------------------------------------------------- # Integration — --all (new fsck mode) # --------------------------------------------------------------------------- class TestAllMode: def test_empty_store_all_ok(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _vo(repo, "--all") assert result.exit_code == 0 data = json.loads(result.output) assert data["all_ok"] is True assert data["checked"] == 0 def test_all_finds_written_objects(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oids = [_write_object(repo, f"content {i}".encode()) for i in range(5)] result = _vo(repo, "--all") assert result.exit_code == 0 data = json.loads(result.output) assert data["checked"] == 5 assert data["all_ok"] is True def test_all_detects_corruption(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_object(repo, b"good data") _corrupt_object(repo, oid) result = _vo(repo, "--all") assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert data["failed"] == 1 def test_all_plus_explicit_ids_rejected(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _vo(repo, "--all", "a" * 64) assert result.exit_code == ExitCode.USER_ERROR assert result.stdout_bytes == b"" # error went to stderr def test_all_quiet(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _write_object(repo, b"content") result = _vo(repo, "--all", "--quiet") assert result.exit_code == 0 assert result.output.strip() == "" # --------------------------------------------------------------------------- # Integration — --stdin (new agent UX) # --------------------------------------------------------------------------- class TestStdinMode: def test_reads_ids_from_stdin(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_object(repo, _FAKE_CONTENT) result = _vo(repo, "--stdin", stdin=f"{oid}\n") assert result.exit_code == 0 data = json.loads(result.output) assert data["checked"] == 1 assert data["all_ok"] is True def test_comments_and_blank_lines_skipped(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_object(repo, _FAKE_CONTENT) result = _vo(repo, "--stdin", stdin=f"\n# comment\n{oid}\n\n") data = json.loads(result.output) assert data["checked"] == 1 def test_stdin_combines_with_positional(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid1 = _write_object(repo, b"one") oid2 = _write_object(repo, b"two") result = _vo(repo, "--stdin", oid1, stdin=f"{oid2}\n") data = json.loads(result.output) assert data["checked"] == 2 def test_empty_stdin_no_explicit_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _vo(repo, "--stdin", stdin="") assert result.exit_code == ExitCode.USER_ERROR # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: def test_format_error_goes_to_stderr(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _vo(repo, "--format", "yaml", "a" * 64) assert result.exit_code == ExitCode.USER_ERROR assert "error" in result.stderr.lower() assert result.stdout_bytes == b"" def test_no_traceback_on_bad_format(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _vo(repo, "--format", "xml", "a" * 64) assert "Traceback" not in result.output def test_ansi_in_error_message_stripped_text(self, tmp_path: pathlib.Path) -> None: """Even if an error message contains ANSI, text mode strips it.""" repo = _make_repo(tmp_path) # Pass a 64-char ID with only hex chars — passes validation, doesn't exist. # The error ("object not found") has no ANSI, but sanitize_display is applied. result = _vo(repo, "--format", "text", "a" * 64) assert "\x1b" not in result.output def test_invalid_id_returns_error_not_crash(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _vo(repo, "not-a-sha256") assert result.exit_code == ExitCode.USER_ERROR assert "Traceback" not in result.output def test_no_ids_errors_to_stderr(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _vo(repo) assert result.exit_code == ExitCode.USER_ERROR assert "error" in result.stderr.lower() # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestStress: def test_100_object_store_all_pass(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) for i in range(100): _write_object(repo, f"stress content {i}".encode()) result = _vo(repo, "--all") assert result.exit_code == 0 data = json.loads(result.output) assert data["checked"] == 100 assert data["all_ok"] is True def test_200_sequential_verifies(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_object(repo, _FAKE_CONTENT) for i in range(200): result = _vo(repo, oid) assert result.exit_code == 0, f"failed at iteration {i}" def test_stdin_200_ids(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oids = [_write_object(repo, f"content_{i}".encode()) for i in range(200)] stdin_input = "\n".join(oids) + "\n" result = _vo(repo, "--stdin", stdin=stdin_input) assert result.exit_code == 0 data = json.loads(result.output) assert data["checked"] == 200 assert data["all_ok"] is True