"""Comprehensive tests for ``muse plumbing cat-object``. Coverage tiers -------------- - Unit: _CHUNK constant, _FORMAT_CHOICES - Integration: raw/info formats, --json alias, missing/invalid object_id - Security: ANSI in object_id error, path traversal object_id - Stress: 10 MiB object streaming, 200 sequential reads """ from __future__ import annotations import hashlib import json import pathlib from muse.core.errors import ExitCode from muse.core.object_store import write_object from tests.cli_test_helper import CliRunner, InvokeResult runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: """Minimal .muse/ structure.""" repo = tmp_path / "repo" muse = repo / ".muse" for sub in ("objects", "commits", "snapshots", "refs/heads"): (muse / sub).mkdir(parents=True) (muse / "HEAD").write_text("ref: refs/heads/main") (muse / "repo.json").write_text(json.dumps({"repo_id": "test", "domain": "code"})) return repo def _store(repo: pathlib.Path, content: bytes) -> str: """Write content to the object store and return its object_id.""" oid = hashlib.sha256(content).hexdigest() write_object(repo, oid, content) return oid def _cat(repo: pathlib.Path, *args: str) -> InvokeResult: from muse.cli.app import main as cli return runner.invoke( cli, ["cat-object", *args], env={"MUSE_REPO_ROOT": str(repo)}, ) # --------------------------------------------------------------------------- # Unit — module constants # --------------------------------------------------------------------------- class TestConstants: def test_chunk_size_is_64kib(self) -> None: from muse.cli.commands.plumbing.cat_object import _CHUNK assert _CHUNK == 65536 def test_format_choices_correct(self) -> None: from muse.cli.commands.plumbing.cat_object import _FORMAT_CHOICES assert "raw" in _FORMAT_CHOICES assert "info" in _FORMAT_CHOICES # "json" must NOT be a format choice — --json is an alias for "info" assert "json" not in _FORMAT_CHOICES # --------------------------------------------------------------------------- # Integration — raw format # --------------------------------------------------------------------------- class TestRawFormat: def test_raw_bytes_match_stored_content(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) content = b"hello object store" oid = _store(repo, content) result = _cat(repo, oid) assert result.exit_code == 0 assert result.stdout_bytes == content def test_raw_is_default_format(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) content = b"default format" oid = _store(repo, content) # No --format flag → should default to raw result = _cat(repo, oid) assert result.exit_code == 0 assert result.stdout_bytes == content def test_raw_binary_content_preserved(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) content = bytes(range(256)) # All byte values including null, control chars oid = _store(repo, content) result = _cat(repo, oid) assert result.exit_code == 0 assert result.stdout_bytes == content def test_raw_empty_object(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) content = b"" oid = _store(repo, content) result = _cat(repo, oid) assert result.exit_code == 0 assert result.stdout_bytes == content def test_explicit_format_raw(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) content = b"explicit raw" oid = _store(repo, content) result = _cat(repo, "--format", "raw", oid) assert result.exit_code == 0 assert result.stdout_bytes == content # --------------------------------------------------------------------------- # Integration — info / --json format # --------------------------------------------------------------------------- class TestInfoFormat: def test_info_format_shape(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) content = b"info content" oid = _store(repo, content) result = _cat(repo, "--format", "info", oid) assert result.exit_code == 0 data = json.loads(result.output) assert data["object_id"] == oid assert data["present"] is True assert data["size_bytes"] == len(content) def test_json_flag_is_alias_for_info(self, tmp_path: pathlib.Path) -> None: """--json must work and produce info-format JSON — this was broken before the audit.""" repo = _make_repo(tmp_path) content = b"json alias test" oid = _store(repo, content) result = _cat(repo, "--json", oid) assert result.exit_code == 0, f"--json failed: {result.output}" data = json.loads(result.output) assert data["object_id"] == oid assert data["present"] is True assert data["size_bytes"] == len(content) def test_info_does_not_emit_content(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) content = b"secret bytes" oid = _store(repo, content) result = _cat(repo, "--format", "info", oid) assert result.exit_code == 0 # Output must be JSON only — not the raw content data = json.loads(result.output) assert "object_id" in data assert content not in result.output.encode() def test_info_size_matches_actual_file(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) content = b"size check " * 100 oid = _store(repo, content) result = _cat(repo, "--json", oid) data = json.loads(result.output) assert data["size_bytes"] == len(content) def test_missing_object_info_has_present_false(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = "a" * 64 result = _cat(repo, "--format", "info", oid) assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert data["present"] is False assert data["size_bytes"] == 0 def test_json_flag_missing_object_has_present_false(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = "b" * 64 result = _cat(repo, "--json", oid) assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert data["present"] is False # --------------------------------------------------------------------------- # Integration — error paths # --------------------------------------------------------------------------- class TestErrorPaths: def test_missing_object_raw_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _cat(repo, "c" * 64) assert result.exit_code == ExitCode.USER_ERROR def test_invalid_object_id_too_short(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _cat(repo, "abc123") assert result.exit_code == ExitCode.USER_ERROR def test_invalid_object_id_uppercase(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _cat(repo, "A" * 64) assert result.exit_code == ExitCode.USER_ERROR def test_invalid_object_id_non_hex(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _cat(repo, "z" * 64) assert result.exit_code == ExitCode.USER_ERROR def test_invalid_format_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _cat(repo, "--format", "xml", "a" * 64) assert result.exit_code == ExitCode.USER_ERROR def test_no_repo_errors(self, tmp_path: pathlib.Path) -> None: from muse.cli.app import main as cli result = runner.invoke( cli, ["cat-object", "a" * 64], env={"MUSE_REPO_ROOT": str(tmp_path / "no_repo")}, ) assert result.exit_code != 0 # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: def test_ansi_in_invalid_id_not_in_output(self, tmp_path: pathlib.Path) -> None: """Crafted object_id with ANSI escapes must not reach output.""" repo = _make_repo(tmp_path) # validate_object_id rejects non-hex, so ANSI never reaches print. # Confirm the error is clean. evil = "\x1b[31m" + "a" * 60 # too short + has escape result = _cat(repo, evil) assert result.exit_code == ExitCode.USER_ERROR assert "\x1b" not in result.output def test_path_traversal_in_object_id_rejected(self, tmp_path: pathlib.Path) -> None: """../../../etc/passwd style object IDs must be rejected by validate_object_id.""" repo = _make_repo(tmp_path) result = _cat(repo, "../../../etc/passwd") assert result.exit_code == ExitCode.USER_ERROR def test_null_byte_in_object_id_rejected(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _cat(repo, "a" * 32 + "\x00" + "b" * 31) assert result.exit_code == ExitCode.USER_ERROR def test_no_traceback_on_invalid_id(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _cat(repo, "not-a-valid-id") assert "Traceback" not in result.output # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestStress: def test_large_object_streams_without_oom(self, tmp_path: pathlib.Path) -> None: """A 10 MiB object must stream out without memory spike.""" repo = _make_repo(tmp_path) content = b"Z" * (10 * 1024 * 1024) # 10 MiB oid = _store(repo, content) result = _cat(repo, oid) assert result.exit_code == 0 assert len(result.stdout_bytes) == len(content) assert result.stdout_bytes == content def test_large_object_info_is_fast(self, tmp_path: pathlib.Path) -> None: """Info format on a 10 MiB object reads only stat(), not content.""" repo = _make_repo(tmp_path) content = b"Y" * (10 * 1024 * 1024) oid = _store(repo, content) result = _cat(repo, "--json", oid) assert result.exit_code == 0 data = json.loads(result.output) assert data["size_bytes"] == len(content) def test_200_sequential_reads(self, tmp_path: pathlib.Path) -> None: """200 rapid cat-object calls return consistent content.""" repo = _make_repo(tmp_path) content = b"repeated read" oid = _store(repo, content) for i in range(200): result = _cat(repo, oid) assert result.exit_code == 0, f"failed at iteration {i}" assert result.stdout_bytes == content