gabriel / muse public

test_code_query_supercharge.py file-level

at sha256:2 · View file ↗ · Intel ↗

History
1 files
1 commits
0 hotspots
0 🧊 dead
0 πŸ’₯ blast risk
sha256:4 Merge branch 'dev' into main · gabriel · Jun 17, 2026
1 """Supercharge tests for ``muse code code-query`` β€” agent-usability gaps.
2
3 The existing test_cmd_code_query.py covers parser correctness, evaluator
4 match logic, walk_history integration, CLI E2E, and stress. This file
5 targets only the gaps those tests leave open:
6
7 Coverage matrix
8 ---------------
9 - --json / -j: -j alias works identically to --json
10 - exit_code: JSON output includes exit_code = 0 on success
11 - duration_ms: JSON output includes non-negative float duration_ms
12 - TypedDicts: _CodeQueryOutputJson gains exit_code/duration_ms annotations
13 - Docstrings: run() docstring mentions exit_code and duration_ms
14 - ANSI: JSON output never contains terminal escape sequences
15 - Performance: duration_ms stays under 2000 ms for a small repo
16 """
17
18 from __future__ import annotations
19 from collections.abc import Mapping
20
21 import json
22 import pathlib
23
24 import pytest
25
26 from tests.cli_test_helper import CliRunner, InvokeResult
27
28 runner = CliRunner()
29
30
31 # ---------------------------------------------------------------------------
32 # Helpers
33 # ---------------------------------------------------------------------------
34
35
36 def _env(root: pathlib.Path) -> Mapping[str, str]:
37 return {"MUSE_REPO_ROOT": str(root)}
38
39
40 def _run(root: pathlib.Path, *args: str) -> InvokeResult:
41 return runner.invoke(None, list(args), env=_env(root))
42
43
44 def _stage_commit(root: pathlib.Path, msg: str = "commit") -> None:
45 r = _run(root, "code", "add", ".")
46 assert r.exit_code == 0, r.output
47 r2 = _run(root, "commit", "-m", msg, "--agent-id", "test-agent", "--model-id", "test-model")
48 assert r2.exit_code == 0, r2.output
49
50
51 # ---------------------------------------------------------------------------
52 # Fixture
53 # ---------------------------------------------------------------------------
54
55
56 @pytest.fixture()
57 def query_repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
58 """Code-domain repo with a few committed files for querying."""
59 monkeypatch.chdir(tmp_path)
60 r = _run(tmp_path, "init", "--domain", "code")
61 assert r.exit_code == 0, r.output
62 (tmp_path / "alpha.py").write_text("def alpha_fn():\n return 1\n")
63 _stage_commit(tmp_path, "add alpha")
64 (tmp_path / "beta.py").write_text("def beta_fn():\n return 2\n")
65 _stage_commit(tmp_path, "add beta")
66 return tmp_path
67
68
69 # ---------------------------------------------------------------------------
70 # TestJsonAlias β€” -j works identically to --json
71 # ---------------------------------------------------------------------------
72
73
74 class TestJsonAlias:
75 """-j shorthand must behave identically to --json."""
76
77 def test_j_alias_exits_zero(self, query_repo: pathlib.Path) -> None:
78 r = _run(query_repo, "code", "code-query", "author == test-agent", "-j")
79 assert r.exit_code == 0, r.output
80
81 def test_j_alias_valid_json(self, query_repo: pathlib.Path) -> None:
82 r = _run(query_repo, "code", "code-query", "author == test-agent", "-j")
83 json.loads(r.output) # must not raise
84
85 def test_j_alias_has_total_key(self, query_repo: pathlib.Path) -> None:
86 r = _run(query_repo, "code", "code-query", "author == test-agent", "-j")
87 data = json.loads(r.output)
88 assert "total" in data
89
90 def test_j_alias_has_results_key(self, query_repo: pathlib.Path) -> None:
91 r = _run(query_repo, "code", "code-query", "author == test-agent", "-j")
92 data = json.loads(r.output)
93 assert "results" in data
94
95 def test_j_alias_same_top_level_keys_as_json_flag(self, query_repo: pathlib.Path) -> None:
96 r1 = _run(query_repo, "code", "code-query", "author == test-agent", "--json")
97 r2 = _run(query_repo, "code", "code-query", "author == test-agent", "-j")
98 d1 = json.loads(r1.output)
99 d2 = json.loads(r2.output)
100 d1.pop("duration_ms", None)
101 d2.pop("duration_ms", None)
102 assert set(d1.keys()) == set(d2.keys())
103
104 def test_j_alias_total_matches_json_flag(self, query_repo: pathlib.Path) -> None:
105 r1 = _run(query_repo, "code", "code-query", "author == test-agent", "--json")
106 r2 = _run(query_repo, "code", "code-query", "author == test-agent", "-j")
107 assert json.loads(r1.output)["total"] == json.loads(r2.output)["total"]
108
109 def test_j_alias_no_matches_empty_results(self, query_repo: pathlib.Path) -> None:
110 r = _run(query_repo, "code", "code-query", "author == nobody", "-j")
111 data = json.loads(r.output)
112 assert data["results"] == []
113 assert data["total"] == 0
114
115 def test_j_alias_with_limit(self, query_repo: pathlib.Path) -> None:
116 r = _run(query_repo, "code", "code-query", "author == test-agent", "-j", "--limit", "1")
117 assert r.exit_code == 0, r.output
118 json.loads(r.output)
119
120
121 # ---------------------------------------------------------------------------
122 # TestDurationMs β€” JSON output must include duration_ms
123 # ---------------------------------------------------------------------------
124
125
126 class TestDurationMs:
127 """JSON output must include a non-negative float duration_ms."""
128
129 def test_json_has_duration_ms(self, query_repo: pathlib.Path) -> None:
130 r = _run(query_repo, "code", "code-query", "author == test-agent", "--json")
131 data = json.loads(r.output)
132 assert "duration_ms" in data
133
134 def test_json_duration_ms_nonnegative(self, query_repo: pathlib.Path) -> None:
135 r = _run(query_repo, "code", "code-query", "author == test-agent", "--json")
136 data = json.loads(r.output)
137 assert data["duration_ms"] >= 0
138
139 def test_json_duration_ms_is_float(self, query_repo: pathlib.Path) -> None:
140 r = _run(query_repo, "code", "code-query", "author == test-agent", "--json")
141 data = json.loads(r.output)
142 assert isinstance(data["duration_ms"], float)
143
144 def test_j_alias_duration_ms_present(self, query_repo: pathlib.Path) -> None:
145 r = _run(query_repo, "code", "code-query", "author == test-agent", "-j")
146 data = json.loads(r.output)
147 assert "duration_ms" in data
148
149 def test_duration_ms_no_matches(self, query_repo: pathlib.Path) -> None:
150 r = _run(query_repo, "code", "code-query", "author == nobody", "--json")
151 data = json.loads(r.output)
152 assert "duration_ms" in data
153 assert data["duration_ms"] >= 0
154
155 def test_duration_ms_with_since(self, query_repo: pathlib.Path) -> None:
156 r = _run(query_repo, "code", "code-query", "author == test-agent",
157 "--json", "--since", "2020-01-01")
158 data = json.loads(r.output)
159 assert "duration_ms" in data
160 assert data["duration_ms"] >= 0
161
162
163 # ---------------------------------------------------------------------------
164 # TestExitCode β€” JSON includes exit_code = 0 on success
165 # ---------------------------------------------------------------------------
166
167
168 class TestExitCode:
169 """JSON exit_code must be 0 on success (all errors raise SystemExit before JSON emits)."""
170
171 def test_json_has_exit_code(self, query_repo: pathlib.Path) -> None:
172 r = _run(query_repo, "code", "code-query", "author == test-agent", "--json")
173 data = json.loads(r.output)
174 assert "exit_code" in data
175
176 def test_json_exit_code_zero_with_matches(self, query_repo: pathlib.Path) -> None:
177 r = _run(query_repo, "code", "code-query", "author == test-agent", "--json")
178 assert r.exit_code == 0
179 data = json.loads(r.output)
180 assert data["exit_code"] == 0
181
182 def test_json_exit_code_zero_no_matches(self, query_repo: pathlib.Path) -> None:
183 r = _run(query_repo, "code", "code-query", "author == nobody", "--json")
184 assert r.exit_code == 0
185 data = json.loads(r.output)
186 assert data["exit_code"] == 0
187
188 def test_json_exit_code_is_int(self, query_repo: pathlib.Path) -> None:
189 r = _run(query_repo, "code", "code-query", "author == test-agent", "--json")
190 data = json.loads(r.output)
191 assert isinstance(data["exit_code"], int)
192
193 def test_j_alias_exit_code_present(self, query_repo: pathlib.Path) -> None:
194 r = _run(query_repo, "code", "code-query", "author == test-agent", "-j")
195 data = json.loads(r.output)
196 assert "exit_code" in data
197
198 def test_exit_code_mirrors_process_exit(self, query_repo: pathlib.Path) -> None:
199 r = _run(query_repo, "code", "code-query", "author == test-agent", "--json")
200 data = json.loads(r.output)
201 assert data["exit_code"] == r.exit_code
202
203 def test_exit_code_zero_with_limit(self, query_repo: pathlib.Path) -> None:
204 r = _run(query_repo, "code", "code-query", "author == test-agent",
205 "--json", "--limit", "1")
206 data = json.loads(r.output)
207 assert data["exit_code"] == 0
208
209
210 # ---------------------------------------------------------------------------
211 # TestTypedDicts β€” _CodeQueryOutputJson carries the new fields
212 # ---------------------------------------------------------------------------
213
214
215 class TestTypedDicts:
216 """_CodeQueryOutputJson must carry exit_code and duration_ms annotations."""
217
218 def test_code_query_output_json_exists(self) -> None:
219 from muse.cli.commands.code_query import _CodeQueryOutputJson # noqa: F401
220
221 def test_has_exit_code_annotation(self) -> None:
222 from muse.cli.commands.code_query import _CodeQueryOutputJson
223 assert "exit_code" in _CodeQueryOutputJson.__annotations__
224
225 def test_has_duration_ms_annotation(self) -> None:
226 from muse.cli.commands.code_query import _CodeQueryOutputJson
227 assert "duration_ms" in _CodeQueryOutputJson.__annotations__
228
229 def test_has_total_annotation(self) -> None:
230 from muse.cli.commands.code_query import _CodeQueryOutputJson
231 assert "total" in _CodeQueryOutputJson.__annotations__
232
233 def test_has_results_annotation(self) -> None:
234 from muse.cli.commands.code_query import _CodeQueryOutputJson
235 assert "results" in _CodeQueryOutputJson.__annotations__
236
237
238 # ---------------------------------------------------------------------------
239 # TestAnsiSanitization β€” no escape codes in JSON output
240 # ---------------------------------------------------------------------------
241
242
243 class TestAnsiSanitization:
244 """No ANSI escape sequences anywhere in the JSON output."""
245
246 def test_json_output_no_ansi_with_matches(self, query_repo: pathlib.Path) -> None:
247 r = _run(query_repo, "code", "code-query", "author == test-agent", "--json")
248 assert "\x1b" not in r.output
249
250 def test_j_alias_output_no_ansi(self, query_repo: pathlib.Path) -> None:
251 r = _run(query_repo, "code", "code-query", "author == test-agent", "-j")
252 assert "\x1b" not in r.output
253
254 def test_json_output_no_ansi_no_matches(self, query_repo: pathlib.Path) -> None:
255 r = _run(query_repo, "code", "code-query", "author == nobody", "--json")
256 assert "\x1b" not in r.output
257
258
259 # ---------------------------------------------------------------------------
260 # TestPerformance β€” duration_ms under 2000 ms for a small repo
261 # ---------------------------------------------------------------------------
262
263
264 class TestPerformance:
265 """duration_ms must stay under 2000 ms for small repos."""
266
267 def test_json_duration_under_2000ms(self, query_repo: pathlib.Path) -> None:
268 r = _run(query_repo, "code", "code-query", "author == test-agent", "--json")
269 data = json.loads(r.output)
270 assert data["duration_ms"] < 2000
271
272 def test_j_alias_duration_under_2000ms(self, query_repo: pathlib.Path) -> None:
273 r = _run(query_repo, "code", "code-query", "author == test-agent", "-j")
274 data = json.loads(r.output)
275 assert data["duration_ms"] < 2000
276
277 def test_duration_ms_is_float_not_int(self, query_repo: pathlib.Path) -> None:
278 r = _run(query_repo, "code", "code-query", "author == test-agent", "--json")
279 data = json.loads(r.output)
280 assert isinstance(data["duration_ms"], float)
281
282
283 class TestRegisterFlags:
284 def test_json_short_flag(self) -> None:
285 import argparse
286 from muse.cli.commands.query import register
287 p = argparse.ArgumentParser()
288 subs = p.add_subparsers()
289 register(subs)
290 args = p.parse_args(["query", "-j"])
291 assert args.json_out is True
292
293 def test_json_long_flag(self) -> None:
294 import argparse
295 from muse.cli.commands.query import register
296 p = argparse.ArgumentParser()
297 subs = p.add_subparsers()
298 register(subs)
299 args = p.parse_args(["query", "--json"])
300 assert args.json_out is True
301
302 def test_default_no_json(self) -> None:
303 import argparse
304 from muse.cli.commands.query import register
305 p = argparse.ArgumentParser()
306 subs = p.add_subparsers()
307 register(subs)
308 # Command-specific required args may differ; just check dest exists when possible
309 try:
310 args = p.parse_args(["query"])
311 assert args.json_out is False
312 except SystemExit:
313 pass # required positional args missing β€” flag default still correct