"""Supercharge tests for ``muse code stable``. Tiers ----- Unit — TypedDict shape, alias registration, docstring completeness. Integration — -j alias, exit_code/duration_ms/schema_version in envelope, filter flags, --since, truncation flag, empty-repo edge case. End-to-end — full CLI invocation; --json vs -j parity; --kind/--language filters. Stress — many commits; concurrent invocations on separate repos. Data integrity — stability counts correct; since_start_of_range semantics; ranked order is descending. Security — ANSI/null in --kind, --language, --since args. Performance — duration_ms present and reasonable. """ from __future__ import annotations from collections.abc import Mapping import json import os import pathlib import textwrap import threading import pytest from tests.cli_test_helper import CliRunner, InvokeResult runner = CliRunner() # ────────────────────────────────────────────────────────────────────────────── # Helpers # ────────────────────────────────────────────────────────────────────────────── def _invoke(repo: pathlib.Path, args: list[str]) -> InvokeResult: saved = os.getcwd() try: os.chdir(repo) return runner.invoke(None, args) finally: os.chdir(saved) def _stable(repo: pathlib.Path, *args: str) -> InvokeResult: return _invoke(repo, ["code", "stable", *args]) def _commit(repo: pathlib.Path, files: Mapping[str, str], message: str) -> None: for name, content in files.items(): path = repo / name path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") saved = os.getcwd() try: os.chdir(repo) runner.invoke(None, ["code", "add", "."]) runner.invoke(None, ["commit", "-m", message]) finally: os.chdir(saved) @pytest.fixture() def stable_repo(tmp_path: pathlib.Path) -> pathlib.Path: """Repo with two commits. Commit 1: a.py (stable — never modified again) + b.py Commit 2: b.py modified (hot), a.py untouched (stable) """ saved = os.getcwd() try: os.chdir(tmp_path) runner.invoke(None, ["init"]) finally: os.chdir(saved) _commit(tmp_path, { "a.py": textwrap.dedent("""\ def stable_fn(): return 42 """), "b.py": textwrap.dedent("""\ def hot_fn(): return 1 """), }, "initial") _commit(tmp_path, { "b.py": textwrap.dedent("""\ def hot_fn(): return 2 """), }, "modify b") return tmp_path # ────────────────────────────────────────────────────────────────────────────── # Unit — TypedDict # ────────────────────────────────────────────────────────────────────────────── class TestTypedDict: def test_stable_json_typed_dict_exists(self) -> None: from muse.cli.commands.stable import _StableJson # noqa: F401 def test_has_exit_code(self) -> None: import typing from muse.cli.commands.stable import _StableJson assert "exit_code" in typing.get_type_hints(_StableJson) def test_has_duration_ms(self) -> None: import typing from muse.cli.commands.stable import _StableJson assert "duration_ms" in typing.get_type_hints(_StableJson) def test_has_schema(self) -> None: import typing from muse.cli.commands.stable import _StableJson assert "schema" in typing.get_type_hints(_StableJson) def test_retains_core_fields(self) -> None: import typing from muse.cli.commands.stable import _StableJson hints = typing.get_type_hints(_StableJson) required = {"from_ref", "to_ref", "commits_analysed", "truncated", "filters", "stable"} assert required <= set(hints) # ────────────────────────────────────────────────────────────────────────────── # Unit — alias registration # ────────────────────────────────────────────────────────────────────────────── class TestAliasRegistration: def _parser(self) -> "argparse.ArgumentParser": import argparse from muse.cli.commands.stable import register p = argparse.ArgumentParser() sub = p.add_subparsers() register(sub) return p def test_j_alias_sets_json_out(self) -> None: ns = self._parser().parse_args(["stable", "-j"]) assert ns.json_out is True def test_j_alias_with_other_flags(self) -> None: ns = self._parser().parse_args(["stable", "-j", "--top", "5"]) assert ns.json_out is True assert ns.top == 5 # ────────────────────────────────────────────────────────────────────────────── # Unit — docstrings # ────────────────────────────────────────────────────────────────────────────── class TestDocstrings: def test_register_docstring_lists_flags(self) -> None: from muse.cli.commands.stable import register doc = register.__doc__ or "" assert "--json" in doc or "-j" in doc def test_run_docstring_mentions_schema(self) -> None: from muse.cli.commands.stable import run assert "schema" in (run.__doc__ or "") # ────────────────────────────────────────────────────────────────────────────── # Integration — JSON envelope fields # ────────────────────────────────────────────────────────────────────────────── class TestJsonEnvelope: def test_has_exit_code(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert "exit_code" in data def test_exit_code_zero(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert data["exit_code"] == 0 def test_exit_code_is_int(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert isinstance(data["exit_code"], int) def test_has_duration_ms(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert "duration_ms" in data def test_duration_ms_nonnegative_float(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert isinstance(data["duration_ms"], float) assert data["duration_ms"] >= 0.0 def test_has_schema(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert "schema" in data def test_schema_is_int(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert isinstance(data["schema"], int) # ────────────────────────────────────────────────────────────────────────────── # Integration — -j alias parity # ────────────────────────────────────────────────────────────────────────────── class TestJsonAlias: def test_j_alias_exit_code_zero(self, stable_repo: pathlib.Path) -> None: assert _stable(stable_repo, "-j").exit_code == 0 def test_j_alias_valid_json(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "-j").output) assert isinstance(data, dict) def test_j_alias_same_top_level_keys(self, stable_repo: pathlib.Path) -> None: keys_json = set(json.loads(_stable(stable_repo, "--json").output)) keys_j = set(json.loads(_stable(stable_repo, "-j").output)) assert keys_json == keys_j def test_j_alias_stable_list_matches(self, stable_repo: pathlib.Path) -> None: d1 = json.loads(_stable(stable_repo, "--json").output) d2 = json.loads(_stable(stable_repo, "-j").output) assert d1["stable"] == d2["stable"] # ────────────────────────────────────────────────────────────────────────────── # End-to-end — filters, output shape # ────────────────────────────────────────────────────────────────────────────── class TestEndToEnd: def test_default_text_output_exits_zero(self, stable_repo: pathlib.Path) -> None: assert _stable(stable_repo).exit_code == 0 def test_default_text_mentions_bedrock(self, stable_repo: pathlib.Path) -> None: assert "bedrock" in _stable(stable_repo).output.lower() def test_json_stable_list_is_list(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert isinstance(data["stable"], list) def test_json_stable_entry_has_address(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert data["stable"] assert "address" in data["stable"][0] def test_json_stable_entry_has_unchanged_for(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert "unchanged_for" in data["stable"][0] def test_json_stable_entry_has_since_start_of_range(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert "since_start_of_range" in data["stable"][0] def test_top_flag_limits_results(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json", "--top", "1").output) assert len(data["stable"]) <= 1 def test_kind_filter_restricts_results(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json", "--kind", "class").output) for entry in data["stable"]: # addresses filtered to class symbols — spot-check via filter echoed pass # no crash and valid JSON is the assertion assert "kind" in data["filters"] def test_commits_analysed_positive(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert data["commits_analysed"] > 0 def test_truncated_flag_present(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert "truncated" in data assert isinstance(data["truncated"], bool) def test_filters_dict_present(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert isinstance(data["filters"], dict) def test_from_ref_and_to_ref_present(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert "from_ref" in data assert "to_ref" in data def test_invalid_since_ref_exits_nonzero(self, stable_repo: pathlib.Path) -> None: result = _stable(stable_repo, "--since", "nonexistent-ref-xyz") assert result.exit_code != 0 # ────────────────────────────────────────────────────────────────────────────── # Stress # ────────────────────────────────────────────────────────────────────────────── class TestStress: def test_many_commits_does_not_crash(self, tmp_path: pathlib.Path) -> None: saved = os.getcwd() try: os.chdir(tmp_path) runner.invoke(None, ["init"]) finally: os.chdir(saved) for i in range(30): _commit(tmp_path, {"f.py": f"def fn(): return {i}\n"}, f"commit {i}") result = _stable(tmp_path, "--json") assert result.exit_code == 0 data = json.loads(result.output) assert data["commits_analysed"] >= 1 def test_concurrent_stable_separate_repos(self, tmp_path: pathlib.Path) -> None: repos = [] for i in range(4): r = tmp_path / f"repo{i}" r.mkdir() saved = os.getcwd() try: os.chdir(r) runner.invoke(None, ["init"]) finally: os.chdir(saved) _commit(r, {"x.py": f"def f(): return {i}\n"}, "init") repos.append(r) results: list[int] = [] lock = threading.Lock() def _run(repo: pathlib.Path) -> None: rc = _stable(repo, "--json").exit_code with lock: results.append(rc) threads = [threading.Thread(target=_run, args=(r,)) for r in repos] for t in threads: t.start() for t in threads: t.join() assert all(rc == 0 for rc in results) # ────────────────────────────────────────────────────────────────────────────── # Data integrity # ────────────────────────────────────────────────────────────────────────────── class TestDataIntegrity: def test_stable_list_sorted_descending(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) counts = [e["unchanged_for"] for e in data["stable"]] assert counts == sorted(counts, reverse=True) def test_unchanged_for_is_nonnegative_int(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) for entry in data["stable"]: assert isinstance(entry["unchanged_for"], int) assert entry["unchanged_for"] >= 0 def test_since_start_of_range_is_bool(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) for entry in data["stable"]: assert isinstance(entry["since_start_of_range"], bool) def test_stable_fn_has_higher_stability_than_hot_fn( self, stable_repo: pathlib.Path ) -> None: """stable_fn was never modified; hot_fn was — stable_fn must rank higher.""" data = json.loads(_stable(stable_repo, "--json").output) stable_counts = { e["address"].split("::")[-1]: e["unchanged_for"] for e in data["stable"] } if "stable_fn" in stable_counts and "hot_fn" in stable_counts: assert stable_counts["stable_fn"] >= stable_counts["hot_fn"] def test_filters_reflect_input_flags(self, stable_repo: pathlib.Path) -> None: data = json.loads( _stable(stable_repo, "--json", "--top", "5", "--kind", "function").output ) assert data["filters"]["top"] == 5 assert data["filters"]["kind"] == "function" def test_max_commits_cap_respected(self, stable_repo: pathlib.Path) -> None: data = json.loads( _stable(stable_repo, "--json", "--max-commits", "1").output ) assert data["commits_analysed"] <= 1 # ────────────────────────────────────────────────────────────────────────────── # Security # ────────────────────────────────────────────────────────────────────────────── class TestSecurity: def test_ansi_in_kind_filter_not_echoed_raw(self, stable_repo: pathlib.Path) -> None: result = _stable(stable_repo, "--kind", "\x1b[31mfunc\x1b[0m") combined = result.output + (result.stderr or "") assert "\x1b[31m" not in combined def test_ansi_in_language_filter_not_echoed_raw( self, stable_repo: pathlib.Path ) -> None: result = _stable(stable_repo, "--language", "\x1b[31mpython\x1b[0m") combined = result.output + (result.stderr or "") assert "\x1b[31m" not in combined def test_ansi_in_since_ref_not_echoed_raw(self, stable_repo: pathlib.Path) -> None: result = _stable(stable_repo, "--since", "\x1b[31mmalicious\x1b[0m") combined = result.output + (result.stderr or "") assert "\x1b[31m" not in combined def test_null_byte_in_kind_does_not_crash(self, stable_repo: pathlib.Path) -> None: result = _stable(stable_repo, "--kind", "func\x00malicious") assert result.exit_code in (0, 1) # ────────────────────────────────────────────────────────────────────────────── # Performance # ────────────────────────────────────────────────────────────────────────────── class TestPerformance: def test_duration_ms_under_10000(self, stable_repo: pathlib.Path) -> None: data = json.loads(_stable(stable_repo, "--json").output) assert data["duration_ms"] < 10_000.0 # --------------------------------------------------------------------------- # Flag registration tests # --------------------------------------------------------------------------- class TestRegisterFlags: def _parser(self) -> "argparse.ArgumentParser": import argparse from muse.cli.commands.stable import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) return p def test_default_json_out_is_false(self) -> None: args = self._parser().parse_args(["stable"]) assert args.json_out is False def test_json_flag_sets_json_out(self) -> None: args = self._parser().parse_args(["stable", "--json"]) assert args.json_out is True def test_j_shorthand_sets_json_out(self) -> None: args = self._parser().parse_args(["stable", "-j"]) assert args.json_out is True