"""Supercharge tests for ``muse code index`` — agent-usability gaps. No prior supercharge tests existed for ``muse code index``. This file covers all three sub-commands (status, rebuild, purge). Coverage matrix --------------- - -j alias: already present on all three sub-commands (no change needed) - exit_code: JSON output includes exit_code = 0 on success (all three) - duration_ms: JSON output includes non-negative float duration_ms (all three) - status shape: status --json now wraps indexes in an object (schema change) - TypedDicts: _RebuildResult, _StatusResult, _PurgeResult all carry exit_code and duration_ms annotations - Docstrings: run_status, run_rebuild, run_purge docstrings mention exit_code and duration_ms - ANSI: JSON output never contains terminal escape sequences - Performance: duration_ms present and is a float """ from __future__ import annotations from collections.abc import Mapping import json import pathlib import textwrap import pytest from tests.cli_test_helper import CliRunner, InvokeResult runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _env(root: pathlib.Path) -> Mapping[str, str]: return {"MUSE_REPO_ROOT": str(root)} def _run(root: pathlib.Path, *args: str) -> InvokeResult: return runner.invoke(None, list(args), env=_env(root)) # --------------------------------------------------------------------------- # Fixture — minimal repo # --------------------------------------------------------------------------- @pytest.fixture() def index_repo( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> pathlib.Path: """Minimal repo with one Python file — enough to build real indexes.""" monkeypatch.chdir(tmp_path) r = _run(tmp_path, "init", "--domain", "code") assert r.exit_code == 0, r.output (tmp_path / "core.py").write_text(textwrap.dedent("""\ def compute(x): return x * 2 def validate(x): return x > 0 """)) r = _run(tmp_path, "code", "add", ".") assert r.exit_code == 0, r.output r = _run(tmp_path, "commit", "-m", "seed index repo") assert r.exit_code == 0, r.output return tmp_path # --------------------------------------------------------------------------- # TestStatusJson — status --json envelope shape # --------------------------------------------------------------------------- class TestStatusJson: """status --json must emit a wrapper object with indexes, exit_code, duration_ms.""" def test_status_j_exits_zero(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "status", "-j") assert r.exit_code == 0, r.output def test_status_j_valid_json(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "status", "-j") json.loads(r.output) # must not raise def test_status_j_has_indexes_key(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "status", "-j") assert "indexes" in json.loads(r.output) def test_status_j_indexes_is_list(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "status", "-j") assert isinstance(json.loads(r.output)["indexes"], list) def test_status_j_has_exit_code(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "status", "-j") assert "exit_code" in json.loads(r.output) def test_status_j_exit_code_zero(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "status", "-j") assert json.loads(r.output)["exit_code"] == 0 def test_status_j_has_duration_ms(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "status", "-j") assert "duration_ms" in json.loads(r.output) def test_status_j_duration_ms_is_float(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "status", "-j") assert isinstance(json.loads(r.output)["duration_ms"], float) def test_status_j_duration_ms_nonnegative(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "status", "-j") assert json.loads(r.output)["duration_ms"] >= 0 def test_status_j_index_entries_have_name(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "status", "-j") for entry in json.loads(r.output)["indexes"]: assert "name" in entry def test_status_j_index_entries_have_status(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "status", "-j") for entry in json.loads(r.output)["indexes"]: assert "status" in entry def test_status_j_no_ansi(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "status", "-j") assert "\x1b" not in r.output def test_status_json_flag_same_as_j(self, index_repo: pathlib.Path) -> None: r1 = _run(index_repo, "code", "index", "status", "--json") r2 = _run(index_repo, "code", "index", "status", "-j") d1 = json.loads(r1.output) d2 = json.loads(r2.output) d1.pop("duration_ms", None) d2.pop("duration_ms", None) d1.pop("timestamp", None) d2.pop("timestamp", None) assert d1 == d2 # --------------------------------------------------------------------------- # TestRebuildJson — rebuild --json envelope # --------------------------------------------------------------------------- class TestRebuildJson: """rebuild --json must include exit_code and duration_ms.""" def test_rebuild_j_exits_zero(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "rebuild", "-j") assert r.exit_code == 0, r.output def test_rebuild_j_valid_json(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "rebuild", "-j") json.loads(r.output) # must not raise def test_rebuild_j_has_exit_code(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "rebuild", "-j") assert "exit_code" in json.loads(r.output) def test_rebuild_j_exit_code_zero(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "rebuild", "-j") assert json.loads(r.output)["exit_code"] == 0 def test_rebuild_j_has_duration_ms(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "rebuild", "-j") assert "duration_ms" in json.loads(r.output) def test_rebuild_j_duration_ms_is_float(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "rebuild", "-j") assert isinstance(json.loads(r.output)["duration_ms"], float) def test_rebuild_j_has_rebuilt_key(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "rebuild", "-j") assert "rebuilt" in json.loads(r.output) def test_rebuild_dry_run_j_exit_code_zero(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "rebuild", "--dry-run", "-j") assert r.exit_code == 0, r.output assert json.loads(r.output)["exit_code"] == 0 def test_rebuild_dry_run_j_duration_ms(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "rebuild", "--dry-run", "-j") data = json.loads(r.output) assert "duration_ms" in data assert isinstance(data["duration_ms"], float) def test_rebuild_j_no_ansi(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "rebuild", "-j") assert "\x1b" not in r.output # --------------------------------------------------------------------------- # TestPurgeJson — purge --json envelope # --------------------------------------------------------------------------- class TestPurgeJson: """purge --json must include exit_code and duration_ms.""" def test_purge_j_exits_zero(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "purge", "-j") assert r.exit_code == 0, r.output def test_purge_j_valid_json(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "purge", "-j") json.loads(r.output) # must not raise def test_purge_j_has_exit_code(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "purge", "-j") assert "exit_code" in json.loads(r.output) def test_purge_j_exit_code_zero(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "purge", "-j") assert json.loads(r.output)["exit_code"] == 0 def test_purge_j_has_duration_ms(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "purge", "-j") assert "duration_ms" in json.loads(r.output) def test_purge_j_duration_ms_is_float(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "purge", "-j") assert isinstance(json.loads(r.output)["duration_ms"], float) def test_purge_j_has_purged_key(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "purge", "-j") assert "purged" in json.loads(r.output) def test_purge_j_has_skipped_key(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "purge", "-j") assert "skipped" in json.loads(r.output) def test_purge_j_no_ansi(self, index_repo: pathlib.Path) -> None: r = _run(index_repo, "code", "index", "purge", "-j") assert "\x1b" not in r.output def test_purge_absent_indexes_skipped_not_error(self, index_repo: pathlib.Path) -> None: """Purging an already-absent index exits 0 and lists it as skipped.""" # Purge twice — second time all indexes are absent _run(index_repo, "code", "index", "purge", "-j") r = _run(index_repo, "code", "index", "purge", "-j") assert r.exit_code == 0 data = json.loads(r.output) assert data["exit_code"] == 0 assert data["purged"] == [] assert len(data["skipped"]) > 0 # --------------------------------------------------------------------------- # TestTypedDicts — TypedDicts carry exit_code and duration_ms # --------------------------------------------------------------------------- class TestTypedDicts: """All three TypedDicts must carry exit_code and duration_ms annotations.""" def test_rebuild_result_typeddict_exists(self) -> None: from muse.cli.commands.index_rebuild import _RebuildResult # noqa: F401 def test_rebuild_result_has_exit_code(self) -> None: from muse.cli.commands.index_rebuild import _RebuildResult assert "exit_code" in _RebuildResult.__annotations__ def test_rebuild_result_has_duration_ms(self) -> None: from muse.cli.commands.index_rebuild import _RebuildResult assert "duration_ms" in _RebuildResult.__annotations__ def test_status_result_typeddict_exists(self) -> None: from muse.cli.commands.index_rebuild import _StatusResult # noqa: F401 def test_status_result_has_indexes(self) -> None: from muse.cli.commands.index_rebuild import _StatusResult assert "indexes" in _StatusResult.__annotations__ def test_status_result_has_exit_code(self) -> None: from muse.cli.commands.index_rebuild import _StatusResult assert "exit_code" in _StatusResult.__annotations__ def test_status_result_has_duration_ms(self) -> None: from muse.cli.commands.index_rebuild import _StatusResult assert "duration_ms" in _StatusResult.__annotations__ def test_purge_result_typeddict_exists(self) -> None: from muse.cli.commands.index_rebuild import _PurgeResult # noqa: F401 def test_purge_result_has_exit_code(self) -> None: from muse.cli.commands.index_rebuild import _PurgeResult assert "exit_code" in _PurgeResult.__annotations__ def test_purge_result_has_duration_ms(self) -> None: from muse.cli.commands.index_rebuild import _PurgeResult assert "duration_ms" in _PurgeResult.__annotations__ # --------------------------------------------------------------------------- # TestDocstrings — run_* functions document exit_code and duration_ms # --------------------------------------------------------------------------- class TestDocstrings: """All three run_* functions must mention exit_code and duration_ms.""" def test_run_status_mentions_exit_code(self) -> None: from muse.cli.commands.index_rebuild import run_status assert run_status.__doc__ is not None assert "exit_code" in run_status.__doc__ def test_run_status_mentions_duration_ms(self) -> None: from muse.cli.commands.index_rebuild import run_status assert "duration_ms" in run_status.__doc__ def test_run_rebuild_mentions_exit_code(self) -> None: from muse.cli.commands.index_rebuild import run_rebuild assert run_rebuild.__doc__ is not None assert "exit_code" in run_rebuild.__doc__ def test_run_rebuild_mentions_duration_ms(self) -> None: from muse.cli.commands.index_rebuild import run_rebuild assert "duration_ms" in run_rebuild.__doc__ def test_run_purge_mentions_exit_code(self) -> None: from muse.cli.commands.index_rebuild import run_purge assert run_purge.__doc__ is not None assert "exit_code" in run_purge.__doc__ def test_run_purge_mentions_duration_ms(self) -> None: from muse.cli.commands.index_rebuild import run_purge assert "duration_ms" in run_purge.__doc__ class TestRegisterFlags: def test_json_short_flag(self) -> None: import argparse from muse.cli.commands.index_rebuild import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) args = p.parse_args(["index", "rebuild", "-j"]) assert args.json_out is True def test_json_long_flag(self) -> None: import argparse from muse.cli.commands.index_rebuild import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) args = p.parse_args(["index", "rebuild", "--json"]) assert args.json_out is True def test_default_no_json(self) -> None: import argparse from muse.cli.commands.index_rebuild import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) args = p.parse_args(["index", "rebuild"]) assert args.json_out is False