"""Supercharge tests for ``muse code add`` / ``muse code reset`` — agent-usability gaps. Coverage matrix --------------- - --json / -j: alias works identically to --format json for both add and reset - exit_code: every JSON output path includes it (success and error branches) - duration_ms: every JSON output path includes it; non-negative float - TypedDicts: _CodeAddJson, _CodeResetJson annotations exist and carry new fields - Docstrings: run_add, run_reset docstrings mention exit_code and duration_ms - Error JSON: error branches include exit_code and duration_ms - ANSI: file paths in JSON output never contain escape sequences - Performance: duration_ms stays < 1000 ms for normal operations """ from __future__ import annotations from collections.abc import Mapping import json import pathlib import pytest from tests.cli_test_helper import CliRunner 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)) @pytest.fixture() def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Fresh code-domain repo with one committed file (main.py = 'x = 1').""" monkeypatch.chdir(tmp_path) r = runner.invoke(None, ["init", "--domain", "code"], env=_env(tmp_path)) assert r.exit_code == 0, r.output (tmp_path / "main.py").write_text("x = 1\n") r2 = runner.invoke(None, ["commit", "--allow-empty", "-m", "init"], env=_env(tmp_path)) assert r2.exit_code == 0, r2.output return tmp_path # --------------------------------------------------------------------------- # --json / -j alias # --------------------------------------------------------------------------- class TestJsonAlias: """--json and -j are accepted on both add and reset, mirror --format json.""" def test_add_json_flag_accepted(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 2\n") r = _run(repo, "code", "add", "--json", "main.py") assert r.exit_code == 0 json.loads(r.output.strip()) # must be valid JSON def test_add_j_flag_accepted(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 3\n") r = _run(repo, "code", "add", "-j", "main.py") assert r.exit_code == 0 json.loads(r.output.strip()) def test_reset_json_flag_accepted(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 4\n") _run(repo, "code", "add", "main.py") r = _run(repo, "code", "reset", "--json", "main.py") assert r.exit_code == 0 json.loads(r.output.strip()) def test_reset_j_flag_accepted(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 5\n") _run(repo, "code", "add", "main.py") r = _run(repo, "code", "reset", "-j", "main.py") assert r.exit_code == 0 json.loads(r.output.strip()) def test_add_json_and_j_produce_same_keys(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 10\n") r1 = _run(repo, "code", "add", "--json", "main.py") (repo / "main.py").write_text("x = 11\n") r2 = _run(repo, "code", "add", "-j", "main.py") d1 = json.loads(r1.output.strip()) d2 = json.loads(r2.output.strip()) d1.pop("duration_ms", None); d1.pop("timestamp", None) d2.pop("duration_ms", None); d2.pop("timestamp", None) assert set(d1.keys()) == set(d2.keys()) def test_add_j_alias_produces_same_keys(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 20\n") r1 = _run(repo, "code", "add", "--json", "main.py") (repo / "main.py").write_text("x = 21\n") r2 = _run(repo, "code", "add", "-j", "main.py") d1 = json.loads(r1.output.strip()) d2 = json.loads(r2.output.strip()) d1.pop("duration_ms", None); d1.pop("timestamp", None) d2.pop("duration_ms", None); d2.pop("timestamp", None) assert set(d1.keys()) == set(d2.keys()) def test_reset_json_and_j_produce_same_keys(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 30\n") _run(repo, "code", "add", "main.py") r1 = _run(repo, "code", "reset", "--json", "main.py") (repo / "main.py").write_text("x = 31\n") _run(repo, "code", "add", "main.py") r2 = _run(repo, "code", "reset", "-j", "main.py") d1 = json.loads(r1.output.strip()) d2 = json.loads(r2.output.strip()) d1.pop("duration_ms", None); d1.pop("timestamp", None) d2.pop("duration_ms", None); d2.pop("timestamp", None) assert set(d1.keys()) == set(d2.keys()) def test_add_dot_with_json_flag(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 99\n") r = _run(repo, "code", "add", "--json", ".") assert r.exit_code == 0 d = json.loads(r.output.strip()) assert "staged" in d def test_add_update_flag_with_json(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 77\n") r = _run(repo, "code", "add", "-u", "--json") assert r.exit_code == 0 d = json.loads(r.output.strip()) assert "staged" in d def test_add_all_flag_with_json(self, repo: pathlib.Path) -> None: (repo / "new.py").write_text("y = 0\n") r = _run(repo, "code", "add", "-A", "--json") assert r.exit_code == 0 d = json.loads(r.output.strip()) assert "staged" in d # --------------------------------------------------------------------------- # duration_ms # --------------------------------------------------------------------------- class TestDurationMs: """Every JSON output path from add and reset must include duration_ms.""" def test_add_success_has_duration_ms(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 2\n") r = _run(repo, "code", "add", "--json", "main.py") assert "duration_ms" in json.loads(r.output.strip()) def test_add_nothing_to_stage_has_duration_ms(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "add", "--json", ".") assert "duration_ms" in json.loads(r.output.strip()) def test_add_dry_run_has_duration_ms(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 5\n") r = _run(repo, "code", "add", "--dry-run", "--json", "main.py") assert "duration_ms" in json.loads(r.output.strip()) def test_add_all_has_duration_ms(self, repo: pathlib.Path) -> None: (repo / "new.py").write_text("y = 0\n") r = _run(repo, "code", "add", "-A", "--json") assert "duration_ms" in json.loads(r.output.strip()) def test_add_update_has_duration_ms(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 6\n") r = _run(repo, "code", "add", "-u", "--json") assert "duration_ms" in json.loads(r.output.strip()) def test_reset_success_has_duration_ms(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 7\n") _run(repo, "code", "add", "main.py") r = _run(repo, "code", "reset", "--json") assert "duration_ms" in json.loads(r.output.strip()) def test_reset_nothing_staged_has_duration_ms(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "reset", "--json") assert "duration_ms" in json.loads(r.output.strip()) def test_reset_dry_run_has_duration_ms(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 8\n") _run(repo, "code", "add", "main.py") r = _run(repo, "code", "reset", "--dry-run", "--json") assert "duration_ms" in json.loads(r.output.strip()) def test_duration_ms_is_non_negative(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 9\n") r = _run(repo, "code", "add", "--json", "main.py") assert json.loads(r.output.strip())["duration_ms"] >= 0 def test_duration_ms_is_numeric(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 10\n") r = _run(repo, "code", "add", "--json", "main.py") val = json.loads(r.output.strip())["duration_ms"] assert isinstance(val, (int, float)) # --------------------------------------------------------------------------- # exit_code # --------------------------------------------------------------------------- class TestExitCode: """Every JSON output path must include exit_code mirroring process exit.""" def test_add_success_has_exit_code(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 2\n") r = _run(repo, "code", "add", "--json", "main.py") assert "exit_code" in json.loads(r.output.strip()) def test_add_nothing_has_exit_code(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "add", "--json", ".") assert "exit_code" in json.loads(r.output.strip()) def test_add_dry_run_has_exit_code(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 5\n") r = _run(repo, "code", "add", "--dry-run", "--json", "main.py") assert "exit_code" in json.loads(r.output.strip()) def test_reset_success_has_exit_code(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 3\n") _run(repo, "code", "add", "main.py") r = _run(repo, "code", "reset", "--json") assert "exit_code" in json.loads(r.output.strip()) def test_reset_nothing_has_exit_code(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "reset", "--json") assert "exit_code" in json.loads(r.output.strip()) def test_add_exit_code_zero_on_success(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 4\n") r = _run(repo, "code", "add", "--json", "main.py") assert json.loads(r.output.strip())["exit_code"] == 0 def test_reset_exit_code_zero_on_success(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "reset", "--json") assert json.loads(r.output.strip())["exit_code"] == 0 def test_add_exit_code_mirrors_process_exit(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 5\n") r = _run(repo, "code", "add", "--json", "main.py") d = json.loads(r.output.strip()) assert d["exit_code"] == r.exit_code def test_reset_exit_code_mirrors_process_exit(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 6\n") _run(repo, "code", "add", "main.py") r = _run(repo, "code", "reset", "--json", "main.py") d = json.loads(r.output.strip()) assert d["exit_code"] == r.exit_code def test_exit_code_is_int(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "reset", "--json") assert isinstance(json.loads(r.output.strip())["exit_code"], int) def test_add_all_exit_code_zero(self, repo: pathlib.Path) -> None: (repo / "new.py").write_text("n = 0\n") r = _run(repo, "code", "add", "-A", "--json") assert json.loads(r.output.strip())["exit_code"] == 0 # --------------------------------------------------------------------------- # TypedDicts # --------------------------------------------------------------------------- class TestTypedDicts: """_CodeAddJson and _CodeResetJson TypedDicts must exist with required fields.""" def test_code_add_json_typeddict_exists(self) -> None: from muse.cli.commands.code_stage import _CodeAddJson assert "staged" in _CodeAddJson.__annotations__ assert "exit_code" in _CodeAddJson.__annotations__ assert "duration_ms" in _CodeAddJson.__annotations__ def test_code_add_json_has_files_annotation(self) -> None: from muse.cli.commands.code_stage import _CodeAddJson assert "files" in _CodeAddJson.__annotations__ def test_code_add_json_has_added_modified_deleted(self) -> None: from muse.cli.commands.code_stage import _CodeAddJson assert "added" in _CodeAddJson.__annotations__ assert "modified" in _CodeAddJson.__annotations__ assert "deleted" in _CodeAddJson.__annotations__ def test_code_add_json_has_dry_run(self) -> None: from muse.cli.commands.code_stage import _CodeAddJson assert "dry_run" in _CodeAddJson.__annotations__ def test_code_reset_json_typeddict_exists(self) -> None: from muse.cli.commands.code_stage import _CodeResetJson assert "unstaged" in _CodeResetJson.__annotations__ assert "exit_code" in _CodeResetJson.__annotations__ assert "duration_ms" in _CodeResetJson.__annotations__ def test_code_reset_json_has_files_annotation(self) -> None: from muse.cli.commands.code_stage import _CodeResetJson assert "files" in _CodeResetJson.__annotations__ def test_code_reset_json_has_not_staged(self) -> None: from muse.cli.commands.code_stage import _CodeResetJson assert "not_staged" in _CodeResetJson.__annotations__ def test_code_reset_json_has_dry_run(self) -> None: from muse.cli.commands.code_stage import _CodeResetJson assert "dry_run" in _CodeResetJson.__annotations__ # --------------------------------------------------------------------------- # Docstrings # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Error JSON paths # --------------------------------------------------------------------------- class TestErrorJson: """Error branches emit valid JSON with exit_code and duration_ms when --json.""" def test_add_no_matching_files_json_has_exit_code(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "add", "--json", "ghost.py") assert r.exit_code != 0 d = json.loads(r.output.strip()) assert "exit_code" in d def test_add_no_matching_files_json_has_duration_ms(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "add", "--json", "ghost.py") assert r.exit_code != 0 d = json.loads(r.output.strip()) assert "duration_ms" in d def test_add_error_exit_code_nonzero_in_json(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "add", "--json", "ghost.py") assert r.exit_code != 0 d = json.loads(r.output.strip()) assert d["exit_code"] != 0 def test_add_error_exit_code_mirrors_process(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "add", "--json", "ghost.py") d = json.loads(r.output.strip()) assert d["exit_code"] == r.exit_code def test_add_error_json_has_staged_zero(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "add", "--json", "ghost.py") d = json.loads(r.output.strip()) assert d.get("staged") == 0 or "error" in d # --------------------------------------------------------------------------- # ANSI sanitization in JSON output # --------------------------------------------------------------------------- class TestAnsiSanitizationJson: """File paths emitted in JSON must never contain ANSI escape sequences.""" def test_add_json_file_path_has_no_ansi(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 2\n") r = _run(repo, "code", "add", "--json", "main.py") assert r.exit_code == 0 assert "\x1b" not in r.output def test_add_json_multiple_files_no_ansi(self, repo: pathlib.Path) -> None: for i in range(3): (repo / f"f{i}.py").write_text(f"v = {i}\n") r = _run(repo, "code", "add", "-A", "--json") assert r.exit_code == 0 assert "\x1b" not in r.output def test_reset_json_file_path_has_no_ansi(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 3\n") _run(repo, "code", "add", "main.py") r = _run(repo, "code", "reset", "--json", "main.py") assert r.exit_code == 0 assert "\x1b" not in r.output def test_add_json_dry_run_no_ansi(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 4\n") r = _run(repo, "code", "add", "--dry-run", "--json", ".") assert "\x1b" not in r.output # --------------------------------------------------------------------------- # Performance # --------------------------------------------------------------------------- class TestPerformance: """duration_ms stays well within reason for normal operations.""" def test_add_single_file_duration_under_1000ms(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 2\n") r = _run(repo, "code", "add", "--json", "main.py") assert r.exit_code == 0 assert json.loads(r.output.strip())["duration_ms"] < 1000 def test_add_all_duration_under_1000ms(self, repo: pathlib.Path) -> None: for i in range(20): (repo / f"f{i}.py").write_text(f"v = {i}\n") r = _run(repo, "code", "add", "-A", "--json") assert r.exit_code == 0 assert json.loads(r.output.strip())["duration_ms"] < 1000 def test_reset_duration_under_1000ms(self, repo: pathlib.Path) -> None: (repo / "main.py").write_text("x = 5\n") _run(repo, "code", "add", "main.py") r = _run(repo, "code", "reset", "--json") assert r.exit_code == 0 assert json.loads(r.output.strip())["duration_ms"] < 1000