test_query_supercharge.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """TDD supercharge tests for ``muse code query`` (symbol graph v2 predicate query). |
| 2 | |
| 3 | Gaps being closed |
| 4 | ----------------- |
| 5 | - ``-j`` alias for ``--json`` |
| 6 | - ``exit_code`` and ``duration_ms`` in single-snapshot JSON envelope |
| 7 | - ``exit_code`` and ``duration_ms`` in ``--all-commits`` JSON envelope |
| 8 | - ``_QueryJson`` TypedDict — formal schema for single-snapshot JSON output |
| 9 | - ``_AllCommitsJson`` TypedDict — formal schema for all-commits JSON output |
| 10 | - Docstring coverage for ``run()`` and ``register()`` |
| 11 | - CLI-level integration tests (zero existed before this file) |
| 12 | |
| 13 | Test classes |
| 14 | ------------ |
| 15 | TestJsonAlias -j alias works identically to --json |
| 16 | TestSingleSnapshotJson exit_code, duration_ms, schema in snapshot mode |
| 17 | TestAllCommitsJson exit_code, duration_ms, schema in --all-commits mode |
| 18 | TestTypedDicts _QueryJson, _AllCommitsJson importable + field coverage |
| 19 | TestCLIBasic basic CLI round-trips (results, no-match, count) |
| 20 | TestCLIFilters --sort, --limit, --count, --unique-bodies, --hashes |
| 21 | TestCLIAllCommits --all-commits basic and with date filters |
| 22 | TestCLISecurity control chars in predicates don't corrupt output |
| 23 | TestDocstrings run() / register() document JSON envelope fields |
| 24 | """ |
| 25 | |
| 26 | from __future__ import annotations |
| 27 | |
| 28 | import json |
| 29 | import pathlib |
| 30 | import textwrap |
| 31 | import typing |
| 32 | |
| 33 | import pytest |
| 34 | |
| 35 | from tests.cli_test_helper import CliRunner |
| 36 | |
| 37 | cli = None |
| 38 | runner = CliRunner() |
| 39 | |
| 40 | |
| 41 | # --------------------------------------------------------------------------- |
| 42 | # Helpers |
| 43 | # --------------------------------------------------------------------------- |
| 44 | |
| 45 | |
| 46 | def _run(root: pathlib.Path, *args: str) -> "InvokeResult": |
| 47 | return runner.invoke(cli, list(args), env={"MUSE_REPO_ROOT": str(root)}) |
| 48 | |
| 49 | |
| 50 | def _commit(root: pathlib.Path, msg: str = "commit") -> None: |
| 51 | r = _run(root, "code", "add", ".") |
| 52 | assert r.exit_code == 0, r.output |
| 53 | r2 = _run(root, "commit", "-m", msg) |
| 54 | assert r2.exit_code == 0, r2.output |
| 55 | |
| 56 | |
| 57 | # --------------------------------------------------------------------------- |
| 58 | # Fixture |
| 59 | # --------------------------------------------------------------------------- |
| 60 | |
| 61 | |
| 62 | @pytest.fixture |
| 63 | def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: |
| 64 | """Code-domain repo with two Python files for symbol-graph querying.""" |
| 65 | monkeypatch.chdir(tmp_path) |
| 66 | monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) |
| 67 | r = _run(tmp_path, "init", "--domain", "code") |
| 68 | assert r.exit_code == 0, r.output |
| 69 | |
| 70 | (tmp_path / "billing.py").write_text(textwrap.dedent("""\ |
| 71 | class Invoice: |
| 72 | def compute_total(self, items: list[int]) -> int: |
| 73 | return sum(items) |
| 74 | |
| 75 | def apply_discount(self, total: float, pct: float) -> float: |
| 76 | return total * (1 - pct) |
| 77 | |
| 78 | def validate_amount(amount: float) -> bool: |
| 79 | return amount > 0 |
| 80 | """)) |
| 81 | |
| 82 | (tmp_path / "utils.py").write_text(textwrap.dedent("""\ |
| 83 | def format_currency(amount: float) -> str: |
| 84 | return f"${amount:.2f}" |
| 85 | |
| 86 | def parse_amount(raw: str) -> float: |
| 87 | return float(raw.strip("$")) |
| 88 | """)) |
| 89 | |
| 90 | _commit(tmp_path, "initial billing + utils") |
| 91 | return tmp_path |
| 92 | |
| 93 | |
| 94 | # --------------------------------------------------------------------------- |
| 95 | # 1. -j alias |
| 96 | # --------------------------------------------------------------------------- |
| 97 | |
| 98 | |
| 99 | class TestJsonAlias: |
| 100 | def test_j_alias_exits_zero(self, repo: pathlib.Path) -> None: |
| 101 | r = _run(repo, "code", "query", "kind=function", "-j") |
| 102 | assert r.exit_code == 0, r.output |
| 103 | |
| 104 | def test_j_alias_emits_valid_json(self, repo: pathlib.Path) -> None: |
| 105 | r = _run(repo, "code", "query", "kind=function", "-j") |
| 106 | assert r.exit_code == 0, r.output |
| 107 | data = json.loads(r.output.strip()) |
| 108 | assert isinstance(data, dict) |
| 109 | |
| 110 | def test_j_alias_has_results(self, repo: pathlib.Path) -> None: |
| 111 | r = _run(repo, "code", "query", "kind=function", "-j") |
| 112 | data = json.loads(r.output) |
| 113 | assert "results" in data |
| 114 | |
| 115 | def test_j_alias_same_keys_as_json_flag(self, repo: pathlib.Path) -> None: |
| 116 | r1 = _run(repo, "code", "query", "kind=function", "--json") |
| 117 | r2 = _run(repo, "code", "query", "kind=function", "-j") |
| 118 | d1 = json.loads(r1.output) |
| 119 | d2 = json.loads(r2.output) |
| 120 | d1.pop("duration_ms", None) |
| 121 | d2.pop("duration_ms", None) |
| 122 | assert set(d1.keys()) == set(d2.keys()) |
| 123 | |
| 124 | def test_j_alias_result_count_matches_json_flag(self, repo: pathlib.Path) -> None: |
| 125 | r1 = _run(repo, "code", "query", "kind=function", "--json") |
| 126 | r2 = _run(repo, "code", "query", "kind=function", "-j") |
| 127 | assert len(json.loads(r1.output)["results"]) == len(json.loads(r2.output)["results"]) |
| 128 | |
| 129 | def test_j_alias_no_match_empty_results(self, repo: pathlib.Path) -> None: |
| 130 | r = _run(repo, "code", "query", "name=zzz_nonexistent", "-j") |
| 131 | data = json.loads(r.output) |
| 132 | assert data["results"] == [] |
| 133 | |
| 134 | def test_j_alias_with_count(self, repo: pathlib.Path) -> None: |
| 135 | # --count takes precedence over JSON even with -j |
| 136 | r = _run(repo, "code", "query", "kind=function", "-j", "--count") |
| 137 | assert r.exit_code == 0, r.output |
| 138 | |
| 139 | |
| 140 | # --------------------------------------------------------------------------- |
| 141 | # 2. Single-snapshot JSON schema |
| 142 | # --------------------------------------------------------------------------- |
| 143 | |
| 144 | |
| 145 | class TestSingleSnapshotJson: |
| 146 | def test_json_exits_zero(self, repo: pathlib.Path) -> None: |
| 147 | r = _run(repo, "code", "query", "kind=function", "--json") |
| 148 | assert r.exit_code == 0, r.output |
| 149 | |
| 150 | def test_json_has_results(self, repo: pathlib.Path) -> None: |
| 151 | r = _run(repo, "code", "query", "kind=function", "--json") |
| 152 | data = json.loads(r.output) |
| 153 | assert "results" in data |
| 154 | assert isinstance(data["results"], list) |
| 155 | |
| 156 | def test_json_has_schema_version(self, repo: pathlib.Path) -> None: |
| 157 | r = _run(repo, "code", "query", "kind=function", "--json") |
| 158 | data = json.loads(r.output) |
| 159 | assert "schema" in data |
| 160 | |
| 161 | def test_json_has_commit(self, repo: pathlib.Path) -> None: |
| 162 | r = _run(repo, "code", "query", "kind=function", "--json") |
| 163 | data = json.loads(r.output) |
| 164 | assert "commit" in data |
| 165 | |
| 166 | def test_json_has_truncated(self, repo: pathlib.Path) -> None: |
| 167 | r = _run(repo, "code", "query", "kind=function", "--json") |
| 168 | data = json.loads(r.output) |
| 169 | assert "truncated" in data |
| 170 | assert isinstance(data["truncated"], bool) |
| 171 | |
| 172 | def test_json_has_exit_code(self, repo: pathlib.Path) -> None: |
| 173 | r = _run(repo, "code", "query", "kind=function", "--json") |
| 174 | data = json.loads(r.output) |
| 175 | assert "exit_code" in data |
| 176 | |
| 177 | def test_json_exit_code_is_zero(self, repo: pathlib.Path) -> None: |
| 178 | r = _run(repo, "code", "query", "kind=function", "--json") |
| 179 | data = json.loads(r.output) |
| 180 | assert data["exit_code"] == 0 |
| 181 | |
| 182 | def test_json_has_duration_ms(self, repo: pathlib.Path) -> None: |
| 183 | r = _run(repo, "code", "query", "kind=function", "--json") |
| 184 | data = json.loads(r.output) |
| 185 | assert "duration_ms" in data |
| 186 | |
| 187 | def test_json_duration_ms_positive(self, repo: pathlib.Path) -> None: |
| 188 | r = _run(repo, "code", "query", "kind=function", "--json") |
| 189 | data = json.loads(r.output) |
| 190 | assert isinstance(data["duration_ms"], float) |
| 191 | assert data["duration_ms"] > 0 |
| 192 | |
| 193 | def test_json_result_record_schema(self, repo: pathlib.Path) -> None: |
| 194 | r = _run(repo, "code", "query", "kind=function", "--json") |
| 195 | data = json.loads(r.output) |
| 196 | assert len(data["results"]) > 0 |
| 197 | rec = data["results"][0] |
| 198 | required = {"address", "kind", "name", "qualified_name", "file", |
| 199 | "lineno", "end_lineno", "size", "language", |
| 200 | "content_id", "body_hash", "signature_id"} |
| 201 | assert required <= set(rec.keys()) |
| 202 | |
| 203 | def test_json_no_match_exit_code_zero(self, repo: pathlib.Path) -> None: |
| 204 | r = _run(repo, "code", "query", "name=zzz_nonexistent", "--json") |
| 205 | assert r.exit_code == 0 |
| 206 | data = json.loads(r.output) |
| 207 | assert data["exit_code"] == 0 |
| 208 | |
| 209 | def test_json_no_match_duration_ms_present(self, repo: pathlib.Path) -> None: |
| 210 | r = _run(repo, "code", "query", "name=zzz_nonexistent", "--json") |
| 211 | data = json.loads(r.output) |
| 212 | assert "duration_ms" in data |
| 213 | |
| 214 | |
| 215 | # --------------------------------------------------------------------------- |
| 216 | # 3. --all-commits JSON schema |
| 217 | # --------------------------------------------------------------------------- |
| 218 | |
| 219 | |
| 220 | class TestAllCommitsJson: |
| 221 | def test_all_commits_json_exits_zero(self, repo: pathlib.Path) -> None: |
| 222 | r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") |
| 223 | assert r.exit_code == 0, r.output |
| 224 | |
| 225 | def test_all_commits_json_has_results(self, repo: pathlib.Path) -> None: |
| 226 | r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") |
| 227 | data = json.loads(r.output) |
| 228 | assert "results" in data |
| 229 | |
| 230 | def test_all_commits_json_has_mode(self, repo: pathlib.Path) -> None: |
| 231 | r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") |
| 232 | data = json.loads(r.output) |
| 233 | assert "mode" in data |
| 234 | assert data["mode"] == "all-commits" |
| 235 | |
| 236 | def test_all_commits_json_has_truncated(self, repo: pathlib.Path) -> None: |
| 237 | r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") |
| 238 | data = json.loads(r.output) |
| 239 | assert "truncated" in data |
| 240 | assert isinstance(data["truncated"], bool) |
| 241 | |
| 242 | def test_all_commits_json_has_exit_code(self, repo: pathlib.Path) -> None: |
| 243 | r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") |
| 244 | data = json.loads(r.output) |
| 245 | assert "exit_code" in data |
| 246 | |
| 247 | def test_all_commits_json_exit_code_zero(self, repo: pathlib.Path) -> None: |
| 248 | r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") |
| 249 | data = json.loads(r.output) |
| 250 | assert data["exit_code"] == 0 |
| 251 | |
| 252 | def test_all_commits_json_has_duration_ms(self, repo: pathlib.Path) -> None: |
| 253 | r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") |
| 254 | data = json.loads(r.output) |
| 255 | assert "duration_ms" in data |
| 256 | |
| 257 | def test_all_commits_json_duration_ms_positive(self, repo: pathlib.Path) -> None: |
| 258 | r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") |
| 259 | data = json.loads(r.output) |
| 260 | assert isinstance(data["duration_ms"], float) |
| 261 | assert data["duration_ms"] > 0 |
| 262 | |
| 263 | def test_all_commits_j_alias_works(self, repo: pathlib.Path) -> None: |
| 264 | r = _run(repo, "code", "query", "kind=function", "--all-commits", "-j") |
| 265 | assert r.exit_code == 0, r.output |
| 266 | data = json.loads(r.output) |
| 267 | assert "exit_code" in data |
| 268 | assert "duration_ms" in data |
| 269 | |
| 270 | def test_all_commits_result_record_schema(self, repo: pathlib.Path) -> None: |
| 271 | r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") |
| 272 | data = json.loads(r.output) |
| 273 | if not data["results"]: |
| 274 | pytest.skip("no results") |
| 275 | rec = data["results"][0] |
| 276 | required = {"address", "kind", "name", "content_id", |
| 277 | "first_seen", "commit_id", "commit_message", "committed_at", "branch"} |
| 278 | assert required <= set(rec.keys()) |
| 279 | |
| 280 | |
| 281 | # --------------------------------------------------------------------------- |
| 282 | # 4. TypedDicts |
| 283 | # --------------------------------------------------------------------------- |
| 284 | |
| 285 | |
| 286 | class TestTypedDicts: |
| 287 | def test_query_json_importable(self) -> None: |
| 288 | from muse.cli.commands.query import _QueryJson |
| 289 | assert _QueryJson is not None |
| 290 | |
| 291 | def test_query_json_has_schema_version(self) -> None: |
| 292 | from muse.cli.commands.query import _QueryJson |
| 293 | hints = typing.get_type_hints(_QueryJson) |
| 294 | assert "schema" in hints |
| 295 | |
| 296 | def test_query_json_has_commit(self) -> None: |
| 297 | from muse.cli.commands.query import _QueryJson |
| 298 | hints = typing.get_type_hints(_QueryJson) |
| 299 | assert "commit" in hints |
| 300 | |
| 301 | def test_query_json_has_results(self) -> None: |
| 302 | from muse.cli.commands.query import _QueryJson |
| 303 | hints = typing.get_type_hints(_QueryJson) |
| 304 | assert "results" in hints |
| 305 | |
| 306 | def test_query_json_has_truncated(self) -> None: |
| 307 | from muse.cli.commands.query import _QueryJson |
| 308 | hints = typing.get_type_hints(_QueryJson) |
| 309 | assert "truncated" in hints |
| 310 | |
| 311 | def test_query_json_has_exit_code(self) -> None: |
| 312 | from muse.cli.commands.query import _QueryJson |
| 313 | hints = typing.get_type_hints(_QueryJson) |
| 314 | assert "exit_code" in hints |
| 315 | |
| 316 | def test_query_json_has_duration_ms(self) -> None: |
| 317 | from muse.cli.commands.query import _QueryJson |
| 318 | hints = typing.get_type_hints(_QueryJson) |
| 319 | assert "duration_ms" in hints |
| 320 | |
| 321 | def test_all_commits_json_importable(self) -> None: |
| 322 | from muse.cli.commands.query import _AllCommitsJson |
| 323 | assert _AllCommitsJson is not None |
| 324 | |
| 325 | def test_all_commits_json_has_mode(self) -> None: |
| 326 | from muse.cli.commands.query import _AllCommitsJson |
| 327 | hints = typing.get_type_hints(_AllCommitsJson) |
| 328 | assert "mode" in hints |
| 329 | |
| 330 | def test_all_commits_json_has_results(self) -> None: |
| 331 | from muse.cli.commands.query import _AllCommitsJson |
| 332 | hints = typing.get_type_hints(_AllCommitsJson) |
| 333 | assert "results" in hints |
| 334 | |
| 335 | def test_all_commits_json_has_exit_code(self) -> None: |
| 336 | from muse.cli.commands.query import _AllCommitsJson |
| 337 | hints = typing.get_type_hints(_AllCommitsJson) |
| 338 | assert "exit_code" in hints |
| 339 | |
| 340 | def test_all_commits_json_has_duration_ms(self) -> None: |
| 341 | from muse.cli.commands.query import _AllCommitsJson |
| 342 | hints = typing.get_type_hints(_AllCommitsJson) |
| 343 | assert "duration_ms" in hints |
| 344 | |
| 345 | |
| 346 | # --------------------------------------------------------------------------- |
| 347 | # 5. Basic CLI round-trips |
| 348 | # --------------------------------------------------------------------------- |
| 349 | |
| 350 | |
| 351 | class TestCLIBasic: |
| 352 | def test_query_functions_returns_results(self, repo: pathlib.Path) -> None: |
| 353 | r = _run(repo, "code", "query", "kind=function") |
| 354 | assert r.exit_code == 0, r.output |
| 355 | assert "match" in r.output.lower() or "fn" in r.output |
| 356 | |
| 357 | def test_query_class_returns_class(self, repo: pathlib.Path) -> None: |
| 358 | r = _run(repo, "code", "query", "kind=class", "--json") |
| 359 | data = json.loads(r.output) |
| 360 | kinds = {rec["kind"] for rec in data["results"]} |
| 361 | assert "class" in kinds |
| 362 | |
| 363 | def test_query_no_match_exits_zero(self, repo: pathlib.Path) -> None: |
| 364 | r = _run(repo, "code", "query", "name=zzz_nonexistent") |
| 365 | assert r.exit_code == 0 |
| 366 | |
| 367 | def test_query_name_contains(self, repo: pathlib.Path) -> None: |
| 368 | r = _run(repo, "code", "query", "name~=compute", "--json") |
| 369 | data = json.loads(r.output) |
| 370 | assert any("compute" in rec["name"].lower() for rec in data["results"]) |
| 371 | |
| 372 | def test_query_file_filter(self, repo: pathlib.Path) -> None: |
| 373 | r = _run(repo, "code", "query", "file~=billing", "--json") |
| 374 | data = json.loads(r.output) |
| 375 | assert all("billing" in rec["file"] for rec in data["results"]) |
| 376 | |
| 377 | def test_query_count_only(self, repo: pathlib.Path) -> None: |
| 378 | r = _run(repo, "code", "query", "kind=function", "--count") |
| 379 | assert r.exit_code == 0 |
| 380 | count = int(r.output.strip()) |
| 381 | assert count > 0 |
| 382 | |
| 383 | def test_query_bad_predicate_exits_one(self, repo: pathlib.Path) -> None: |
| 384 | r = _run(repo, "code", "query", "nonexistent_field=value") |
| 385 | assert r.exit_code == 1 |
| 386 | |
| 387 | def test_query_missing_predicate_exits_one(self, repo: pathlib.Path) -> None: |
| 388 | # No predicates at all — should fail with USER_ERROR |
| 389 | r = _run(repo, "code", "query") |
| 390 | assert r.exit_code != 0 |
| 391 | |
| 392 | |
| 393 | # --------------------------------------------------------------------------- |
| 394 | # 6. Filters |
| 395 | # --------------------------------------------------------------------------- |
| 396 | |
| 397 | |
| 398 | class TestCLIFilters: |
| 399 | def test_limit_caps_results(self, repo: pathlib.Path) -> None: |
| 400 | r = _run(repo, "code", "query", "kind=function", "--json", "--limit", "1") |
| 401 | data = json.loads(r.output) |
| 402 | assert len(data["results"]) <= 1 |
| 403 | |
| 404 | def test_limit_sets_truncated_flag(self, repo: pathlib.Path) -> None: |
| 405 | # If there are more results than the limit, truncated should be True. |
| 406 | r_all = _run(repo, "code", "query", "kind=function", "--json") |
| 407 | total = len(json.loads(r_all.output)["results"]) |
| 408 | if total <= 1: |
| 409 | pytest.skip("not enough results to test truncation") |
| 410 | r = _run(repo, "code", "query", "kind=function", "--json", "--limit", "1") |
| 411 | data = json.loads(r.output) |
| 412 | assert data["truncated"] is True |
| 413 | |
| 414 | def test_sort_by_name(self, repo: pathlib.Path) -> None: |
| 415 | r = _run(repo, "code", "query", "kind=function", "--json", "--sort", "name") |
| 416 | data = json.loads(r.output) |
| 417 | names = [rec["name"] for rec in data["results"]] |
| 418 | assert names == sorted(names, key=str.lower) |
| 419 | |
| 420 | def test_sort_by_size_largest_first(self, repo: pathlib.Path) -> None: |
| 421 | r = _run(repo, "code", "query", "kind=function", "--json", "--sort", "size") |
| 422 | data = json.loads(r.output) |
| 423 | sizes = [rec["size"] for rec in data["results"]] |
| 424 | assert sizes == sorted(sizes, reverse=True) |
| 425 | |
| 426 | def test_unique_bodies_deduplicates(self, repo: pathlib.Path) -> None: |
| 427 | # Write two files with identical function bodies. |
| 428 | (repo / "dup_a.py").write_text("def same_body():\n return 42\n") |
| 429 | (repo / "dup_b.py").write_text("def same_body():\n return 42\n") |
| 430 | _commit(repo, "add duplicates") |
| 431 | |
| 432 | r_all = _run(repo, "code", "query", "name=same_body", "--json") |
| 433 | r_uniq = _run(repo, "code", "query", "name=same_body", "--json", "--unique-bodies") |
| 434 | all_count = len(json.loads(r_all.output)["results"]) |
| 435 | uniq_count = len(json.loads(r_uniq.output)["results"]) |
| 436 | assert uniq_count <= all_count |
| 437 | |
| 438 | def test_hashes_adds_content_id(self, repo: pathlib.Path) -> None: |
| 439 | # --hashes in human mode adds hash info to output |
| 440 | r = _run(repo, "code", "query", "kind=function", "--hashes") |
| 441 | assert r.exit_code == 0 |
| 442 | # JSON mode always includes content_id regardless of --hashes |
| 443 | r2 = _run(repo, "code", "query", "kind=function", "--json") |
| 444 | data = json.loads(r2.output) |
| 445 | for rec in data["results"]: |
| 446 | assert "content_id" in rec |
| 447 | |
| 448 | def test_or_predicate(self, repo: pathlib.Path) -> None: |
| 449 | r = _run(repo, "code", "query", "(kind=function OR kind=class)", "--json") |
| 450 | data = json.loads(r.output) |
| 451 | kinds = {rec["kind"] for rec in data["results"]} |
| 452 | assert kinds <= {"function", "class", "method", "async_function"} |
| 453 | |
| 454 | def test_not_predicate(self, repo: pathlib.Path) -> None: |
| 455 | r_all = _run(repo, "code", "query", "kind=function", "--json") |
| 456 | r_not = _run(repo, "code", "query", "NOT kind=import", "kind=function", "--json") |
| 457 | # NOT kind=import AND kind=function should only include functions |
| 458 | data = json.loads(r_not.output) |
| 459 | for rec in data["results"]: |
| 460 | assert rec["kind"] != "import" |
| 461 | |
| 462 | |
| 463 | # --------------------------------------------------------------------------- |
| 464 | # 7. --all-commits |
| 465 | # --------------------------------------------------------------------------- |
| 466 | |
| 467 | |
| 468 | class TestCLIAllCommits: |
| 469 | def test_all_commits_exits_zero(self, repo: pathlib.Path) -> None: |
| 470 | r = _run(repo, "code", "query", "kind=function", "--all-commits") |
| 471 | assert r.exit_code == 0, r.output |
| 472 | |
| 473 | def test_all_commits_finds_symbols(self, repo: pathlib.Path) -> None: |
| 474 | r = _run(repo, "code", "query", "name~=compute", "--all-commits", "--json") |
| 475 | data = json.loads(r.output) |
| 476 | assert len(data["results"]) > 0 |
| 477 | |
| 478 | def test_all_commits_mutual_exclusive_with_commit(self, repo: pathlib.Path) -> None: |
| 479 | r = _run(repo, "code", "query", "kind=function", |
| 480 | "--all-commits", "--commit", "HEAD") |
| 481 | assert r.exit_code == 1 |
| 482 | |
| 483 | def test_all_commits_since_filter(self, repo: pathlib.Path) -> None: |
| 484 | r = _run(repo, "code", "query", "kind=function", |
| 485 | "--all-commits", "--since", "2099-01-01", "--json") |
| 486 | data = json.loads(r.output) |
| 487 | # Far-future date → no results |
| 488 | assert data["results"] == [] |
| 489 | |
| 490 | def test_all_commits_since_requires_all_commits(self, repo: pathlib.Path) -> None: |
| 491 | r = _run(repo, "code", "query", "kind=function", "--since", "2026-01-01") |
| 492 | assert r.exit_code == 1 |
| 493 | |
| 494 | def test_all_commits_count(self, repo: pathlib.Path) -> None: |
| 495 | r = _run(repo, "code", "query", "kind=function", "--all-commits", "--count") |
| 496 | assert r.exit_code == 0 |
| 497 | assert int(r.output.strip()) >= 0 |
| 498 | |
| 499 | |
| 500 | # --------------------------------------------------------------------------- |
| 501 | # 8. Security |
| 502 | # --------------------------------------------------------------------------- |
| 503 | |
| 504 | |
| 505 | class TestCLISecurity: |
| 506 | def test_control_chars_in_predicate_value_rejected(self, repo: pathlib.Path) -> None: |
| 507 | # A predicate with embedded control chars should fail gracefully. |
| 508 | r = _run(repo, "code", "query", "name=\x00malicious") |
| 509 | # Either fails to parse (exit 1) or matches nothing safely (exit 0). |
| 510 | assert "\x00" not in r.output |
| 511 | |
| 512 | def test_ansi_not_in_json_output(self, repo: pathlib.Path) -> None: |
| 513 | r = _run(repo, "code", "query", "kind=function", "--json") |
| 514 | assert "\x1b" not in r.output |
| 515 | |
| 516 | def test_ansi_not_in_all_commits_json_output(self, repo: pathlib.Path) -> None: |
| 517 | r = _run(repo, "code", "query", "kind=function", "--all-commits", "--json") |
| 518 | assert "\x1b" not in r.output |
| 519 | |
| 520 | |
| 521 | # --------------------------------------------------------------------------- |
| 522 | # 9. Docstrings |
| 523 | # --------------------------------------------------------------------------- |
| 524 | |
| 525 | |
| 526 | class TestDocstrings: |
| 527 | def test_run_docstring_exists(self) -> None: |
| 528 | from muse.cli.commands.query import run |
| 529 | assert run.__doc__ is not None |
| 530 | assert len(run.__doc__) > 50 |
| 531 | |
| 532 | def test_run_docstring_mentions_json(self) -> None: |
| 533 | from muse.cli.commands.query import run |
| 534 | assert "json" in (run.__doc__ or "").lower() |
| 535 | |
| 536 | |
| 537 | |
| 538 | def test_register_docstring_exists(self) -> None: |
| 539 | from muse.cli.commands.query import register |
| 540 | assert register.__doc__ is not None |
| 541 | |
| 542 | |
| 543 | class TestRegisterFlags: |
| 544 | def _make_parser(self) -> "argparse.ArgumentParser": |
| 545 | import argparse |
| 546 | from muse.cli.commands.query import register |
| 547 | p = argparse.ArgumentParser() |
| 548 | subs = p.add_subparsers() |
| 549 | register(subs) |
| 550 | return p |
| 551 | |
| 552 | def test_default_json_out_is_false(self) -> None: |
| 553 | args = self._make_parser().parse_args(["query", "kind=function"]) |
| 554 | assert args.json_out is False |
| 555 | |
| 556 | def test_json_flag_sets_json_out(self) -> None: |
| 557 | args = self._make_parser().parse_args(["query", "kind=function", "--json"]) |
| 558 | assert args.json_out is True |
| 559 | |
| 560 | def test_j_shorthand_sets_json_out(self) -> None: |
| 561 | args = self._make_parser().parse_args(["query", "kind=function", "-j"]) |
| 562 | assert args.json_out is True |