"""Supercharge tests for ``muse ls-files``. Coverage tiers -------------- I JSON envelope schema — status, error, branch, path_prefix, duration_ms, exit_code II Error payload shape — consistent {status, error, exit_code}; no prose in JSON mode III branch field — reflects the branch HEAD resolved to IV path_prefix echoed — agents can verify which filter was applied V TypedDicts — _LsFilesJson and _LsFilesErrorJson exist with correct annotations VI Docstring — documents all envelope fields VII Data integrity — object_ids are sha256:-prefixed in all output modes VIII No prose pollution in JSON mode """ from __future__ import annotations from collections.abc import Mapping import datetime import json import pathlib import pytest from muse.core.object_store import write_object from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) from muse.core.types import Manifest, blob_id, long_id from muse.core.paths import ref_path, muse_dir from tests.cli_test_helper import CliRunner, InvokeResult runner = CliRunner() _ENVELOPE_KEYS = {"duration_ms", "exit_code", "muse_version", "schema", "timestamp", "warnings"} _REQUIRED_SUCCESS_KEYS = { "status", "error", "commit_id", "snapshot_id", "branch", "path_prefix", "file_count", "files", } | _ENVELOPE_KEYS _REQUIRED_ERROR_KEYS = {"status", "error"} | _ENVELOPE_KEYS _TS = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _oid(content: bytes) -> str: return blob_id(content) def _make_repo(tmp_path: pathlib.Path, branch: str = "main") -> pathlib.Path: repo = tmp_path / "repo" dot_muse = muse_dir(repo) for sub in ("objects", "commits", "snapshots", "refs/heads"): (dot_muse / sub).mkdir(parents=True) (dot_muse / "HEAD").write_text(f"ref: refs/heads/{branch}") (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test", "domain": "code"})) return repo def _add_commit( repo: pathlib.Path, files: dict[str, bytes], *, branch: str = "main", set_head: bool = True, ) -> str: stored: Manifest = {} for path, content in files.items(): oid = _oid(content) write_object(repo, oid, content) stored[path] = oid snap_id = compute_snapshot_id(stored) write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=stored, created_at=_TS)) commit_id = compute_commit_id( parent_ids=[], snapshot_id=snap_id, message="test", committed_at_iso=_TS.isoformat(), author="tester",) write_commit(repo, CommitRecord( commit_id=commit_id, branch=branch, snapshot_id=snap_id, message="test", committed_at=_TS, author="tester", parent_commit_id=None, )) if set_head: ref = ref_path(repo, branch) ref.parent.mkdir(parents=True, exist_ok=True) ref.write_text(commit_id) return commit_id def _ls(repo: pathlib.Path, *args: str) -> InvokeResult: from muse.cli.app import main as cli return runner.invoke(cli, ["ls-files", *args], env={"MUSE_REPO_ROOT": str(repo)}) def _ls_json(repo: pathlib.Path, *args: str) -> Mapping[str, object]: result = _ls(repo, "--json", *args) assert result.exit_code == 0, f"ls-files --json failed:\n{result.output}" return json.loads(result.output.strip()) # --------------------------------------------------------------------------- # I JSON envelope schema # --------------------------------------------------------------------------- class TestJsonEnvelopeSchema: def test_all_required_keys_present(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a"}) data = _ls_json(repo) missing = _REQUIRED_SUCCESS_KEYS - set(data.keys()) assert not missing, f"Missing envelope keys: {missing}" def test_no_extra_undocumented_keys(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a"}) data = _ls_json(repo) extra = set(data.keys()) - _REQUIRED_SUCCESS_KEYS assert not extra, f"Undocumented extra keys: {extra}" def test_status_ok_on_success(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a"}) data = _ls_json(repo) assert data["status"] == "ok" def test_error_empty_string_on_success(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a"}) data = _ls_json(repo) assert data["error"] == "" def test_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a"}) data = _ls_json(repo) assert data["exit_code"] == 0 def test_duration_ms_nonnegative_float(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a"}) data = _ls_json(repo) assert isinstance(data["duration_ms"], float) assert data["duration_ms"] >= 0.0 def test_file_count_matches_files_length(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a", "b.py": b"b", "c.py": b"c"}) data = _ls_json(repo) assert data["file_count"] == len(data["files"]) def test_files_is_list(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a"}) data = _ls_json(repo) assert isinstance(data["files"], list) def test_path_prefix_none_when_not_filtered(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a"}) data = _ls_json(repo) assert data["path_prefix"] is None def test_path_prefix_echoed_when_filtered(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"src/a.py": b"a", "tests/b.py": b"b"}) data = _ls_json(repo, "--path-prefix", "src/") assert data["path_prefix"] == "src/" def test_commit_id_sha256_prefixed(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a"}) data = _ls_json(repo) assert data["commit_id"].startswith("sha256:") def test_snapshot_id_sha256_prefixed(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a"}) data = _ls_json(repo) assert data["snapshot_id"].startswith("sha256:") # --------------------------------------------------------------------------- # II Error payload shape # --------------------------------------------------------------------------- class TestErrorPayloadShape: def test_error_keys_present(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) # no commits result = _ls(repo, "--json") assert result.exit_code != 0 data = json.loads(result.output.strip()) assert _REQUIRED_ERROR_KEYS.issubset(set(data.keys())) def test_error_status_is_error(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ls(repo, "--json") assert result.exit_code != 0 data = json.loads(result.output.strip()) assert data["status"] == "error" def test_error_message_nonempty(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ls(repo, "--json") data = json.loads(result.output.strip()) assert isinstance(data["error"], str) and len(data["error"]) > 0 def test_error_exit_code_nonzero(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ls(repo, "--json") data = json.loads(result.output.strip()) assert isinstance(data["exit_code"], int) and data["exit_code"] != 0 def test_invalid_commit_error_payload(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ls(repo, "--json", "--commit", "not-valid") assert result.exit_code != 0 data = json.loads(result.output.strip()) assert data["status"] == "error" assert "exit_code" in data def test_nonexistent_commit_error_payload(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ls(repo, "--json", "--commit", long_id("f" * 64)) assert result.exit_code != 0 data = json.loads(result.output.strip()) assert data["status"] == "error" # --------------------------------------------------------------------------- # III branch field # --------------------------------------------------------------------------- class TestBranchField: def test_branch_is_main_when_on_main(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path, branch="main") _add_commit(repo, {"a.py": b"a"}, branch="main") data = _ls_json(repo) assert data["branch"] == "main" def test_branch_is_dev_when_on_dev(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path, branch="dev") _add_commit(repo, {"a.py": b"a"}, branch="dev") data = _ls_json(repo) assert data["branch"] == "dev" def test_branch_is_none_when_explicit_commit_given( self, tmp_path: pathlib.Path ) -> None: """When --commit is given explicitly, no branch resolution occurs.""" repo = _make_repo(tmp_path) cid = _add_commit(repo, {"a.py": b"a"}) data = _ls_json(repo, "--commit", cid) # branch is null when commit was specified directly, not via HEAD assert data["branch"] is None # --------------------------------------------------------------------------- # IV path_prefix echoed # --------------------------------------------------------------------------- class TestPathPrefixEchoed: def test_path_prefix_null_without_filter(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a"}) data = _ls_json(repo) assert data["path_prefix"] is None def test_path_prefix_echoed_src(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"src/a.py": b"a"}) data = _ls_json(repo, "--path-prefix", "src/") assert data["path_prefix"] == "src/" def test_path_prefix_echoed_nested(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a/b/c.py": b"c"}) data = _ls_json(repo, "--path-prefix", "a/b/") assert data["path_prefix"] == "a/b/" # --------------------------------------------------------------------------- # V TypedDicts # --------------------------------------------------------------------------- class TestTypedDicts: def test_ls_files_json_typed_dict_exists(self) -> None: from muse.cli.commands.ls_files import _LsFilesJson # type: ignore[attr-defined] assert _LsFilesJson is not None def test_ls_files_error_json_typed_dict_exists(self) -> None: from muse.cli.commands.ls_files import _LsFilesErrorJson # type: ignore[attr-defined] assert _LsFilesErrorJson is not None def test_ls_files_json_has_all_annotations(self) -> None: from muse.cli.commands.ls_files import _LsFilesJson # type: ignore[attr-defined] hints = _LsFilesJson.__annotations__ required = {"status", "error", "commit_id", "snapshot_id", "branch", "path_prefix", "file_count", "files", "duration_ms", "exit_code"} assert not (required - set(hints)), f"Missing: {required - set(hints)}" def test_ls_files_error_json_has_all_annotations(self) -> None: from muse.cli.commands.ls_files import _LsFilesErrorJson # type: ignore[attr-defined] hints = _LsFilesErrorJson.__annotations__ assert not ({"status", "error", "exit_code"} - set(hints)) # --------------------------------------------------------------------------- # VI Docstring # --------------------------------------------------------------------------- class TestDocstring: def test_docstring_documents_status(self) -> None: import muse.cli.commands.ls_files as m assert '"status"' in (m.__doc__ or "") def test_docstring_documents_branch(self) -> None: import muse.cli.commands.ls_files as m assert '"branch"' in (m.__doc__ or "") def test_docstring_documents_path_prefix(self) -> None: import muse.cli.commands.ls_files as m assert '"path_prefix"' in (m.__doc__ or "") def test_docstring_documents_duration_ms(self) -> None: import muse.cli.commands.ls_files as m assert "duration_ms" in (m.__doc__ or "") def test_docstring_documents_exit_code(self) -> None: import muse.cli.commands.ls_files as m assert "exit_code" in (m.__doc__ or "") def test_docstring_documents_error(self) -> None: import muse.cli.commands.ls_files as m assert '"error"' in (m.__doc__ or "") # --------------------------------------------------------------------------- # VII Data integrity # --------------------------------------------------------------------------- class TestDataIntegrity: def test_object_ids_sha256_prefixed_in_json(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"content", "b.py": b"more"}) data = _ls_json(repo) for f in data["files"]: assert f["object_id"].startswith("sha256:"), ( f"object_id not sha256:-prefixed: {f['object_id']!r}" ) def test_object_ids_sha256_prefixed_in_text(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"content"}) # Default (no --json) emits text: \t per line result = _ls(repo) assert result.exit_code == 0 for line in result.output.strip().splitlines(): oid = line.split("\t")[0] assert oid.startswith("sha256:"), f"text OID not prefixed: {oid!r}" def test_commit_id_matches_stored(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _add_commit(repo, {"a.py": b"a"}) data = _ls_json(repo) assert data["commit_id"] == cid def test_files_sorted_alphabetically(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"z.py": b"z", "a.py": b"a", "m.py": b"m"}) data = _ls_json(repo) paths = [f["path"] for f in data["files"]] assert paths == sorted(paths) def test_path_prefix_file_count_consistent(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"src/a.py": b"a", "src/b.py": b"b", "tests/c.py": b"c"}) data = _ls_json(repo, "--path-prefix", "src/") assert data["file_count"] == len(data["files"]) == 2 # --------------------------------------------------------------------------- # VIII No prose pollution in JSON mode # --------------------------------------------------------------------------- class TestNoProsePollution: def test_success_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a"}) result = _ls(repo, "--json") json.loads(result.output.strip()) # must not raise def test_no_emoji_in_json_success_output(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _add_commit(repo, {"a.py": b"a"}) result = _ls(repo, "--json") assert "❌" not in result.output assert "✅" not in result.output def test_no_emoji_in_json_error_output(self, tmp_path: pathlib.Path) -> None: """Errors in --json mode must not emit prose emoji to stdout.""" repo = _make_repo(tmp_path) # no commits result = _ls(repo, "--json") assert "❌" not in result.output data = json.loads(result.output.strip()) assert data["status"] == "error" def test_error_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ls(repo, "--json") json.loads(result.output.strip()) # must not raise class TestRegisterFlags: def test_json_short_flag(self) -> None: import argparse from muse.cli.commands.ls_files import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) args = p.parse_args(["ls-files", "-j"]) assert args.json_out is True def test_json_long_flag(self) -> None: import argparse from muse.cli.commands.ls_files import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) args = p.parse_args(["ls-files", "--json"]) assert args.json_out is True def test_default_no_json(self) -> None: import argparse from muse.cli.commands.ls_files import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) # Command-specific required args may differ; just check dest exists when possible try: args = p.parse_args(["ls-files"]) assert args.json_out is False except SystemExit: pass # required positional args missing — flag default still correct