gabriel / muse public
test_query_supercharge.py python
562 lines 22.8 KB
Raw
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 21 days ago
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
File History 4 commits
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 21 days ago
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e fix: rename objects→blobs in push client and all stale test… Sonnet 4.6 patch 23 days ago
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 29 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor ⚠ 29 days ago