"""TDD supercharge tests for ``muse code query`` (symbol graph v2 predicate query). Gaps being closed ----------------- - ``-j`` alias for ``--json`` - ``exit_code`` and ``duration_ms`` in single-snapshot JSON envelope - ``exit_code`` and ``duration_ms`` in ``--all-commits`` JSON envelope - ``_QueryJson`` TypedDict — formal schema for single-snapshot JSON output - ``_AllCommitsJson`` TypedDict — formal schema for all-commits JSON output - Docstring coverage for ``run()`` and ``register()`` - CLI-level integration tests (zero existed before this file) Test classes ------------ TestJsonAlias -j alias works identically to --json TestSingleSnapshotJson exit_code, duration_ms, schema in snapshot mode TestAllCommitsJson exit_code, duration_ms, schema in --all-commits mode TestTypedDicts _QueryJson, _AllCommitsJson importable + field coverage TestCLIBasic basic CLI round-trips (results, no-match, count) TestCLIFilters --sort, --limit, --count, --unique-bodies, --hashes TestCLIAllCommits --all-commits basic and with date filters TestCLISecurity control chars in predicates don't corrupt output TestDocstrings run() / register() document JSON envelope fields """ from __future__ import annotations import json import pathlib import textwrap import typing import pytest from tests.cli_test_helper import CliRunner cli = None runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _run(root: pathlib.Path, *args: str) -> "InvokeResult": return runner.invoke(cli, list(args), env={"MUSE_REPO_ROOT": str(root)}) def _commit(root: pathlib.Path, msg: str = "commit") -> None: r = _run(root, "code", "add", ".") assert r.exit_code == 0, r.output r2 = _run(root, "commit", "-m", msg) assert r2.exit_code == 0, r2.output # --------------------------------------------------------------------------- # Fixture # --------------------------------------------------------------------------- @pytest.fixture def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Code-domain repo with two Python files for symbol-graph querying.""" monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) r = _run(tmp_path, "init", "--domain", "code") assert r.exit_code == 0, r.output (tmp_path / "billing.py").write_text(textwrap.dedent("""\ class Invoice: def compute_total(self, items: list[int]) -> int: return sum(items) def apply_discount(self, total: float, pct: float) -> float: return total * (1 - pct) def validate_amount(amount: float) -> bool: return amount > 0 """)) (tmp_path / "utils.py").write_text(textwrap.dedent("""\ def format_currency(amount: float) -> str: return f"${amount:.2f}" def parse_amount(raw: str) -> float: return float(raw.strip("$")) """)) _commit(tmp_path, "initial billing + utils") return tmp_path # --------------------------------------------------------------------------- # 1. -j alias # --------------------------------------------------------------------------- class TestJsonAlias: def test_j_alias_exits_zero(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "-j") assert r.exit_code == 0, r.output def test_j_alias_emits_valid_json(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "-j") assert r.exit_code == 0, r.output data = json.loads(r.output.strip()) assert isinstance(data, dict) def test_j_alias_has_results(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "-j") data = json.loads(r.output) assert "results" in data def test_j_alias_same_keys_as_json_flag(self, repo: pathlib.Path) -> None: r1 = _run(repo, "code", "query", "kind=function", "--json") r2 = _run(repo, "code", "query", "kind=function", "-j") d1 = json.loads(r1.output) d2 = json.loads(r2.output) d1.pop("duration_ms", None) d2.pop("duration_ms", None) assert set(d1.keys()) == set(d2.keys()) def test_j_alias_result_count_matches_json_flag(self, repo: pathlib.Path) -> None: r1 = _run(repo, "code", "query", "kind=function", "--json") r2 = _run(repo, "code", "query", "kind=function", "-j") assert len(json.loads(r1.output)["results"]) == len(json.loads(r2.output)["results"]) def test_j_alias_no_match_empty_results(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "name=zzz_nonexistent", "-j") data = json.loads(r.output) assert data["results"] == [] def test_j_alias_with_count(self, repo: pathlib.Path) -> None: # --count takes precedence over JSON even with -j r = _run(repo, "code", "query", "kind=function", "-j", "--count") assert r.exit_code == 0, r.output # --------------------------------------------------------------------------- # 2. Single-snapshot JSON schema # --------------------------------------------------------------------------- class TestSingleSnapshotJson: def test_json_exits_zero(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json") assert r.exit_code == 0, r.output def test_json_has_results(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json") data = json.loads(r.output) assert "results" in data assert isinstance(data["results"], list) def test_json_has_schema_version(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json") data = json.loads(r.output) assert "schema" in data def test_json_has_commit(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json") data = json.loads(r.output) assert "commit" in data def test_json_has_truncated(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json") data = json.loads(r.output) assert "truncated" in data assert isinstance(data["truncated"], bool) def test_json_has_exit_code(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json") data = json.loads(r.output) assert "exit_code" in data def test_json_exit_code_is_zero(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json") data = json.loads(r.output) assert data["exit_code"] == 0 def test_json_has_duration_ms(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json") data = json.loads(r.output) assert "duration_ms" in data def test_json_duration_ms_positive(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json") data = json.loads(r.output) assert isinstance(data["duration_ms"], float) assert data["duration_ms"] > 0 def test_json_result_record_schema(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json") data = json.loads(r.output) assert len(data["results"]) > 0 rec = data["results"][0] required = {"address", "kind", "name", "qualified_name", "file", "lineno", "end_lineno", "size", "language", "content_id", "body_hash", "signature_id"} assert required <= set(rec.keys()) def test_json_no_match_exit_code_zero(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "name=zzz_nonexistent", "--json") assert r.exit_code == 0 data = json.loads(r.output) assert data["exit_code"] == 0 def test_json_no_match_duration_ms_present(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "name=zzz_nonexistent", "--json") data = json.loads(r.output) assert "duration_ms" in data # --------------------------------------------------------------------------- # 3. --all-commits JSON schema # --------------------------------------------------------------------------- class TestAllCommitsJson: def test_all_commits_json_exits_zero(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") assert r.exit_code == 0, r.output def test_all_commits_json_has_results(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") data = json.loads(r.output) assert "results" in data def test_all_commits_json_has_mode(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") data = json.loads(r.output) assert "mode" in data assert data["mode"] == "all-commits" def test_all_commits_json_has_truncated(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") data = json.loads(r.output) assert "truncated" in data assert isinstance(data["truncated"], bool) def test_all_commits_json_has_exit_code(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") data = json.loads(r.output) assert "exit_code" in data def test_all_commits_json_exit_code_zero(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") data = json.loads(r.output) assert data["exit_code"] == 0 def test_all_commits_json_has_duration_ms(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") data = json.loads(r.output) assert "duration_ms" in data def test_all_commits_json_duration_ms_positive(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") data = json.loads(r.output) assert isinstance(data["duration_ms"], float) assert data["duration_ms"] > 0 def test_all_commits_j_alias_works(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "-j") assert r.exit_code == 0, r.output data = json.loads(r.output) assert "exit_code" in data assert "duration_ms" in data def test_all_commits_result_record_schema(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") data = json.loads(r.output) if not data["results"]: pytest.skip("no results") rec = data["results"][0] required = {"address", "kind", "name", "content_id", "first_seen", "commit_id", "commit_message", "committed_at", "branch"} assert required <= set(rec.keys()) # --------------------------------------------------------------------------- # 4. TypedDicts # --------------------------------------------------------------------------- class TestTypedDicts: def test_query_json_importable(self) -> None: from muse.cli.commands.query import _QueryJson assert _QueryJson is not None def test_query_json_has_schema_version(self) -> None: from muse.cli.commands.query import _QueryJson hints = typing.get_type_hints(_QueryJson) assert "schema" in hints def test_query_json_has_commit(self) -> None: from muse.cli.commands.query import _QueryJson hints = typing.get_type_hints(_QueryJson) assert "commit" in hints def test_query_json_has_results(self) -> None: from muse.cli.commands.query import _QueryJson hints = typing.get_type_hints(_QueryJson) assert "results" in hints def test_query_json_has_truncated(self) -> None: from muse.cli.commands.query import _QueryJson hints = typing.get_type_hints(_QueryJson) assert "truncated" in hints def test_query_json_has_exit_code(self) -> None: from muse.cli.commands.query import _QueryJson hints = typing.get_type_hints(_QueryJson) assert "exit_code" in hints def test_query_json_has_duration_ms(self) -> None: from muse.cli.commands.query import _QueryJson hints = typing.get_type_hints(_QueryJson) assert "duration_ms" in hints def test_all_commits_json_importable(self) -> None: from muse.cli.commands.query import _AllCommitsJson assert _AllCommitsJson is not None def test_all_commits_json_has_mode(self) -> None: from muse.cli.commands.query import _AllCommitsJson hints = typing.get_type_hints(_AllCommitsJson) assert "mode" in hints def test_all_commits_json_has_results(self) -> None: from muse.cli.commands.query import _AllCommitsJson hints = typing.get_type_hints(_AllCommitsJson) assert "results" in hints def test_all_commits_json_has_exit_code(self) -> None: from muse.cli.commands.query import _AllCommitsJson hints = typing.get_type_hints(_AllCommitsJson) assert "exit_code" in hints def test_all_commits_json_has_duration_ms(self) -> None: from muse.cli.commands.query import _AllCommitsJson hints = typing.get_type_hints(_AllCommitsJson) assert "duration_ms" in hints # --------------------------------------------------------------------------- # 5. Basic CLI round-trips # --------------------------------------------------------------------------- class TestCLIBasic: def test_query_functions_returns_results(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function") assert r.exit_code == 0, r.output assert "match" in r.output.lower() or "fn" in r.output def test_query_class_returns_class(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=class", "--json") data = json.loads(r.output) kinds = {rec["kind"] for rec in data["results"]} assert "class" in kinds def test_query_no_match_exits_zero(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "name=zzz_nonexistent") assert r.exit_code == 0 def test_query_name_contains(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "name~=compute", "--json") data = json.loads(r.output) assert any("compute" in rec["name"].lower() for rec in data["results"]) def test_query_file_filter(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "file~=billing", "--json") data = json.loads(r.output) assert all("billing" in rec["file"] for rec in data["results"]) def test_query_count_only(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--count") assert r.exit_code == 0 count = int(r.output.strip()) assert count > 0 def test_query_bad_predicate_exits_one(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "nonexistent_field=value") assert r.exit_code == 1 def test_query_missing_predicate_exits_one(self, repo: pathlib.Path) -> None: # No predicates at all — should fail with USER_ERROR r = _run(repo, "code", "query") assert r.exit_code != 0 # --------------------------------------------------------------------------- # 6. Filters # --------------------------------------------------------------------------- class TestCLIFilters: def test_limit_caps_results(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json", "--limit", "1") data = json.loads(r.output) assert len(data["results"]) <= 1 def test_limit_sets_truncated_flag(self, repo: pathlib.Path) -> None: # If there are more results than the limit, truncated should be True. r_all = _run(repo, "code", "query", "kind=function", "--json") total = len(json.loads(r_all.output)["results"]) if total <= 1: pytest.skip("not enough results to test truncation") r = _run(repo, "code", "query", "kind=function", "--json", "--limit", "1") data = json.loads(r.output) assert data["truncated"] is True def test_sort_by_name(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json", "--sort", "name") data = json.loads(r.output) names = [rec["name"] for rec in data["results"]] assert names == sorted(names, key=str.lower) def test_sort_by_size_largest_first(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json", "--sort", "size") data = json.loads(r.output) sizes = [rec["size"] for rec in data["results"]] assert sizes == sorted(sizes, reverse=True) def test_unique_bodies_deduplicates(self, repo: pathlib.Path) -> None: # Write two files with identical function bodies. (repo / "dup_a.py").write_text("def same_body():\n return 42\n") (repo / "dup_b.py").write_text("def same_body():\n return 42\n") _commit(repo, "add duplicates") r_all = _run(repo, "code", "query", "name=same_body", "--json") r_uniq = _run(repo, "code", "query", "name=same_body", "--json", "--unique-bodies") all_count = len(json.loads(r_all.output)["results"]) uniq_count = len(json.loads(r_uniq.output)["results"]) assert uniq_count <= all_count def test_hashes_adds_content_id(self, repo: pathlib.Path) -> None: # --hashes in human mode adds hash info to output r = _run(repo, "code", "query", "kind=function", "--hashes") assert r.exit_code == 0 # JSON mode always includes content_id regardless of --hashes r2 = _run(repo, "code", "query", "kind=function", "--json") data = json.loads(r2.output) for rec in data["results"]: assert "content_id" in rec def test_or_predicate(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "(kind=function OR kind=class)", "--json") data = json.loads(r.output) kinds = {rec["kind"] for rec in data["results"]} assert kinds <= {"function", "class", "method", "async_function"} def test_not_predicate(self, repo: pathlib.Path) -> None: r_all = _run(repo, "code", "query", "kind=function", "--json") r_not = _run(repo, "code", "query", "NOT kind=import", "kind=function", "--json") # NOT kind=import AND kind=function should only include functions data = json.loads(r_not.output) for rec in data["results"]: assert rec["kind"] != "import" # --------------------------------------------------------------------------- # 7. --all-commits # --------------------------------------------------------------------------- class TestCLIAllCommits: def test_all_commits_exits_zero(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits") assert r.exit_code == 0, r.output def test_all_commits_finds_symbols(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "name~=compute", "--all-commits", "--json") data = json.loads(r.output) assert len(data["results"]) > 0 def test_all_commits_mutual_exclusive_with_commit(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "--commit", "HEAD") assert r.exit_code == 1 def test_all_commits_since_filter(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "--since", "2099-01-01", "--json") data = json.loads(r.output) # Far-future date → no results assert data["results"] == [] def test_all_commits_since_requires_all_commits(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--since", "2026-01-01") assert r.exit_code == 1 def test_all_commits_count(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "--count") assert r.exit_code == 0 assert int(r.output.strip()) >= 0 # --------------------------------------------------------------------------- # 8. Security # --------------------------------------------------------------------------- class TestCLISecurity: def test_control_chars_in_predicate_value_rejected(self, repo: pathlib.Path) -> None: # A predicate with embedded control chars should fail gracefully. r = _run(repo, "code", "query", "name=\x00malicious") # Either fails to parse (exit 1) or matches nothing safely (exit 0). assert "\x00" not in r.output def test_ansi_not_in_json_output(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--json") assert "\x1b" not in r.output def test_ansi_not_in_all_commits_json_output(self, repo: pathlib.Path) -> None: r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") assert "\x1b" not in r.output # --------------------------------------------------------------------------- # 9. Docstrings # --------------------------------------------------------------------------- class TestDocstrings: def test_run_docstring_exists(self) -> None: from muse.cli.commands.query import run assert run.__doc__ is not None assert len(run.__doc__) > 50 def test_run_docstring_mentions_json(self) -> None: from muse.cli.commands.query import run assert "json" in (run.__doc__ or "").lower() def test_register_docstring_exists(self) -> None: from muse.cli.commands.query import register assert register.__doc__ is not None class TestRegisterFlags: def _make_parser(self) -> "argparse.ArgumentParser": import argparse from muse.cli.commands.query import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) return p def test_default_json_out_is_false(self) -> None: args = self._make_parser().parse_args(["query", "kind=function"]) assert args.json_out is False def test_json_flag_sets_json_out(self) -> None: args = self._make_parser().parse_args(["query", "kind=function", "--json"]) assert args.json_out is True def test_j_shorthand_sets_json_out(self) -> None: args = self._make_parser().parse_args(["query", "kind=function", "-j"]) assert args.json_out is True