gabriel / muse public
test_test_cmd_supercharge.py python
802 lines 35.6 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 140 days ago
1 """Seven-tier tests for ``muse/cli/commands/test_cmd.py``.
2
3 Tiers
4 -----
5 Unit — TypedDict fields (_FullJson schema_version); _fatal human/JSON;
6 _progress_cb icons; _history_to_json roundtrip; _gate_to_json;
7 _ci_to_json; _run_result_to_record; _print_history;
8 _print_pre_run; _print_dry_run human/JSON; _print_summary.
9 Integration — -j alias parity; register() has -j; docstrings document
10 schema_version/exit_code/duration_ms.
11 End-to-end — --dry-run (no pytest); --history (no pytest); --flaky (no pytest);
12 --json --dry-run; --json --history; invalid repo; -j --dry-run.
13 Stress — 1 000 _history_to_json calls; 500 _gate_to_json calls;
14 _print_history with 200 entries.
15 Data integrity — JSON fields correct types; schema_version present on all modes;
16 _FullJson required fields; _history_to_json preserves every field.
17 Security — hostile node_id in history survives JSON; ANSI in messages;
18 SQL injection in gate name; long stdout in gate result.
19 Performance — 1 000 _gate_to_json under 0.5 s; duration_ms in --dry-run JSON.
20 """
21
22 from __future__ import annotations
23
24 import json
25 import os
26 import pathlib
27 import textwrap
28 import threading
29 import time
30 from typing import get_type_hints
31
32 import pytest
33
34 from tests.cli_test_helper import CliRunner, InvokeResult
35
36 runner = CliRunner()
37
38
39 # ──────────────────────────────────────────────────────────────────────────────
40 # Shared helpers — minimal typed-dict factories
41 # ──────────────────────────────────────────────────────────────────────────────
42
43
44 def _make_history_summary(**kw) -> dict:
45 from muse.core.test_history import HistorySummary
46 base = dict(
47 node_id="tests/test_foo.py::test_bar",
48 total_runs=10,
49 pass_count=8,
50 fail_count=2,
51 skip_count=0,
52 flaky=True,
53 avg_duration_ms=123.4,
54 last_outcome="passed",
55 last_run_timestamp="2026-01-01T00:00:00+00:00",
56 fail_streak=0,
57 )
58 base.update(kw)
59 return HistorySummary(**base)
60
61
62 def _make_gate(**kw) -> dict:
63 from muse.core.ci import GateResult
64 base = dict(
65 name="lint",
66 command=["ruff", "check", "."],
67 exit_code=0,
68 duration_ms=120.0,
69 stdout="All checks passed.",
70 stderr="",
71 required=True,
72 passed=True,
73 timed_out=False,
74 )
75 base.update(kw)
76 return GateResult(**base)
77
78
79 def _make_ci_result(**kw) -> dict:
80 from muse.core.ci import CiRunResult
81 base = dict(
82 passed=True,
83 gates=[_make_gate()],
84 duration_ms=200.0,
85 timestamp="2026-01-01T00:00:00+00:00",
86 )
87 base.update(kw)
88 return CiRunResult(**base)
89
90
91 def _make_run_result(**kw) -> dict:
92 from muse.core.test_runner import RunResult
93 base = dict(
94 run_id="run-1",
95 targets=[],
96 exit_code=0,
97 duration_ms=500.0,
98 results=[],
99 total=3,
100 passed=3,
101 failed=0,
102 errored=0,
103 skipped=0,
104 timed_out=False,
105 json_report_available=True,
106 stdout="",
107 stderr="",
108 )
109 base.update(kw)
110 return RunResult(**base)
111
112
113 def _make_selection(**kw) -> dict:
114 from muse.core.test_selection import SelectionResult
115 base = dict(
116 changed_addresses=["src/foo.py::bar"],
117 test_targets=[],
118 covered_addresses=["src/foo.py::bar"],
119 uncovered_addresses=[],
120 coverage_fraction=1.0,
121 fallback_used=False,
122 )
123 base.update(kw)
124 return SelectionResult(**base)
125
126
127 def _commit(repo: pathlib.Path, files: dict[str, str], message: str) -> None:
128 for name, content in files.items():
129 path = repo / name
130 path.parent.mkdir(parents=True, exist_ok=True)
131 path.write_text(content, encoding="utf-8")
132 saved = os.getcwd()
133 try:
134 os.chdir(repo)
135 runner.invoke(None, ["code", "add", "."])
136 runner.invoke(None, ["commit", "-m", message])
137 finally:
138 os.chdir(saved)
139
140
141 def _invoke(repo: pathlib.Path, args: list[str]) -> InvokeResult:
142 saved = os.getcwd()
143 try:
144 os.chdir(repo)
145 return runner.invoke(None, args)
146 finally:
147 os.chdir(saved)
148
149
150 @pytest.fixture()
151 def test_repo(tmp_path: pathlib.Path) -> pathlib.Path:
152 """Minimal repo with a real test file so history/dry-run modes work."""
153 saved = os.getcwd()
154 try:
155 os.chdir(tmp_path)
156 runner.invoke(None, ["init"])
157 finally:
158 os.chdir(saved)
159
160 _commit(tmp_path, {
161 "src/calc.py": textwrap.dedent("""\
162 def add(a, b):
163 return a + b
164 """),
165 "tests/test_calc.py": textwrap.dedent("""\
166 from src.calc import add
167
168 def test_add():
169 assert add(1, 2) == 3
170 """),
171 }, "feat: add calc and test")
172
173 return tmp_path
174
175
176 # ──────────────────────────────────────────────────────────────────────────────
177 # Unit — TypedDict / _FullJson schema_version
178 # ──────────────────────────────────────────────────────────────────────────────
179
180
181 class TestTypedDict:
182 def test_full_json_has_schema_version(self) -> None:
183 from muse.cli.commands.test_cmd import _FullJson
184 assert "schema_version" in get_type_hints(_FullJson)
185
186 def test_full_json_has_mode(self) -> None:
187 from muse.cli.commands.test_cmd import _FullJson
188 assert "mode" in get_type_hints(_FullJson)
189
190 def test_selection_json_fields(self) -> None:
191 from muse.cli.commands.test_cmd import _SelectionJson
192 hints = get_type_hints(_SelectionJson)
193 for f in ("changed_addresses", "covered_addresses", "uncovered_addresses",
194 "coverage_fraction", "fallback_used", "targets"):
195 assert f in hints, f"missing: {f}"
196
197 def test_run_json_has_exit_code(self) -> None:
198 from muse.cli.commands.test_cmd import _RunJson
199 assert "exit_code" in get_type_hints(_RunJson)
200
201 def test_run_json_has_duration_ms(self) -> None:
202 from muse.cli.commands.test_cmd import _RunJson
203 assert "duration_ms" in get_type_hints(_RunJson)
204
205 def test_history_json_fields(self) -> None:
206 from muse.cli.commands.test_cmd import _HistoryJson
207 hints = get_type_hints(_HistoryJson)
208 for f in ("node_id", "total_runs", "pass_count", "fail_count",
209 "flaky", "avg_duration_ms", "fail_streak"):
210 assert f in hints, f"missing: {f}"
211
212 def test_ci_gate_json_fields(self) -> None:
213 from muse.cli.commands.test_cmd import _CiGateJson
214 hints = get_type_hints(_CiGateJson)
215 for f in ("name", "command", "exit_code", "duration_ms",
216 "required", "passed", "timed_out"):
217 assert f in hints, f"missing: {f}"
218
219
220 # ──────────────────────────────────────────────────────────────────────────────
221 # Unit — _fatal
222 # ──────────────────────────────────────────────────────────────────────────────
223
224
225 class TestFatal:
226 def test_human_mode_exits_1(self, capsys) -> None:
227 from muse.cli.commands.test_cmd import _fatal
228 with pytest.raises(SystemExit) as exc_info:
229 _fatal("something broke", json_out=False)
230 assert exc_info.value.code == 1
231
232 def test_human_mode_prints_to_stderr(self, capsys) -> None:
233 from muse.cli.commands.test_cmd import _fatal
234 with pytest.raises(SystemExit):
235 _fatal("something broke", json_out=False)
236 assert "something broke" in capsys.readouterr().err
237
238 def test_json_mode_prints_error_key(self, capsys) -> None:
239 from muse.cli.commands.test_cmd import _fatal
240 with pytest.raises(SystemExit):
241 _fatal("bad config", json_out=True)
242 d = json.loads(capsys.readouterr().out)
243 assert d["error"] == "bad config"
244
245 def test_json_mode_exits_1(self) -> None:
246 from muse.cli.commands.test_cmd import _fatal
247 with pytest.raises(SystemExit) as exc_info:
248 _fatal("x", json_out=True)
249 assert exc_info.value.code == 1
250
251
252 # ──────────────────────────────────────────────────────────────────────────────
253 # Unit — _progress_cb
254 # ──────────────────────────────────────────────────────────────────────────────
255
256
257 class TestProgressCb:
258 def _case(self, outcome: str) -> dict:
259 from muse.core.test_runner import CaseResult
260 return CaseResult(node_id="t::t", outcome=outcome, duration_ms=1.0)
261
262 def test_passed_prints_dot(self, capsys) -> None:
263 from muse.cli.commands.test_cmd import _progress_cb
264 _progress_cb(self._case("passed"))
265 assert capsys.readouterr().err == "."
266
267 def test_failed_prints_f(self, capsys) -> None:
268 from muse.cli.commands.test_cmd import _progress_cb
269 _progress_cb(self._case("failed"))
270 assert capsys.readouterr().err == "F"
271
272 def test_error_prints_e(self, capsys) -> None:
273 from muse.cli.commands.test_cmd import _progress_cb
274 _progress_cb(self._case("error"))
275 assert capsys.readouterr().err == "E"
276
277 def test_skipped_prints_s(self, capsys) -> None:
278 from muse.cli.commands.test_cmd import _progress_cb
279 _progress_cb(self._case("skipped"))
280 assert capsys.readouterr().err == "s"
281
282 def test_unknown_outcome_prints_q(self, capsys) -> None:
283 from muse.cli.commands.test_cmd import _progress_cb
284 _progress_cb(self._case("weird"))
285 assert capsys.readouterr().err == "?"
286
287
288 # ──────────────────────────────────────────────────────────────────────────────
289 # Unit — _history_to_json
290 # ──────────────────────────────────────────────────────────────────────────────
291
292
293 class TestHistoryToJson:
294 def test_preserves_node_id(self) -> None:
295 from muse.cli.commands.test_cmd import _history_to_json
296 s = _make_history_summary(node_id="tests/test_foo.py::test_x")
297 d = _history_to_json(s)
298 assert d["node_id"] == "tests/test_foo.py::test_x"
299
300 def test_preserves_counts(self) -> None:
301 from muse.cli.commands.test_cmd import _history_to_json
302 s = _make_history_summary(pass_count=7, fail_count=3, total_runs=10)
303 d = _history_to_json(s)
304 assert d["pass_count"] == 7
305 assert d["fail_count"] == 3
306 assert d["total_runs"] == 10
307
308 def test_preserves_flaky_flag(self) -> None:
309 from muse.cli.commands.test_cmd import _history_to_json
310 d = _history_to_json(_make_history_summary(flaky=True))
311 assert d["flaky"] is True
312
313 def test_preserves_avg_duration_ms(self) -> None:
314 from muse.cli.commands.test_cmd import _history_to_json
315 d = _history_to_json(_make_history_summary(avg_duration_ms=99.9))
316 assert abs(d["avg_duration_ms"] - 99.9) < 0.001
317
318 def test_result_is_json_serialisable(self) -> None:
319 from muse.cli.commands.test_cmd import _history_to_json
320 d = _history_to_json(_make_history_summary())
321 json.dumps(d) # must not raise
322
323
324 # ──────────────────────────────────────────────────────────────────────────────
325 # Unit — _gate_to_json
326 # ──────────────────────────────────────────────────────────────────────────────
327
328
329 class TestGateToJson:
330 def test_preserves_name(self) -> None:
331 from muse.cli.commands.test_cmd import _gate_to_json
332 d = _gate_to_json(_make_gate(name="mygate"))
333 assert d["name"] == "mygate"
334
335 def test_preserves_exit_code(self) -> None:
336 from muse.cli.commands.test_cmd import _gate_to_json
337 d = _gate_to_json(_make_gate(exit_code=1))
338 assert d["exit_code"] == 1
339
340 def test_preserves_passed(self) -> None:
341 from muse.cli.commands.test_cmd import _gate_to_json
342 d = _gate_to_json(_make_gate(passed=False))
343 assert d["passed"] is False
344
345 def test_warning_included_when_present(self) -> None:
346 from muse.cli.commands.test_cmd import _gate_to_json
347 gate = _make_gate()
348 gate["warning"] = "watch out"
349 d = _gate_to_json(gate)
350 assert d["warning"] == "watch out"
351
352 def test_warning_absent_when_not_set(self) -> None:
353 from muse.cli.commands.test_cmd import _gate_to_json
354 d = _gate_to_json(_make_gate())
355 assert "warning" not in d
356
357 def test_result_is_json_serialisable(self) -> None:
358 from muse.cli.commands.test_cmd import _gate_to_json
359 json.dumps(_gate_to_json(_make_gate()))
360
361
362 # ──────────────────────────────────────────────────────────────────────────────
363 # Unit — _ci_to_json
364 # ──────────────────────────────────────────────────────────────────────────────
365
366
367 class TestCiToJson:
368 def test_preserves_passed(self) -> None:
369 from muse.cli.commands.test_cmd import _ci_to_json
370 d = _ci_to_json(_make_ci_result(passed=False))
371 assert d["passed"] is False
372
373 def test_gates_list_length(self) -> None:
374 from muse.cli.commands.test_cmd import _ci_to_json
375 ci = _make_ci_result(gates=[_make_gate(), _make_gate(name="test")])
376 d = _ci_to_json(ci)
377 assert len(d["gates"]) == 2
378
379 def test_result_is_json_serialisable(self) -> None:
380 from muse.cli.commands.test_cmd import _ci_to_json
381 json.dumps(_ci_to_json(_make_ci_result()))
382
383
384 # ──────────────────────────────────────────────────────────────────────────────
385 # Unit — _print_history
386 # ──────────────────────────────────────────────────────────────────────────────
387
388
389 class TestPrintHistory:
390 def test_empty_history_prints_no_history(self, capsys) -> None:
391 from muse.cli.commands.test_cmd import _print_history
392 _print_history({}, flaky_only=False)
393 assert "No test history" in capsys.readouterr().out
394
395 def test_empty_flaky_prints_no_flaky(self, capsys) -> None:
396 from muse.cli.commands.test_cmd import _print_history
397 _print_history({}, flaky_only=True)
398 assert "No flaky" in capsys.readouterr().out
399
400 def test_non_empty_shows_node_id(self, capsys) -> None:
401 from muse.cli.commands.test_cmd import _print_history
402 s = _make_history_summary(node_id="tests/test_x.py::test_y")
403 _print_history({"tests/test_x.py::test_y": s}, flaky_only=False)
404 assert "test_y" in capsys.readouterr().out
405
406 def test_flaky_only_filters_non_flaky(self, capsys) -> None:
407 from muse.cli.commands.test_cmd import _print_history
408 non_flaky = _make_history_summary(node_id="t::a", flaky=False)
409 flaky = _make_history_summary(node_id="t::b", flaky=True)
410 _print_history({"t::a": non_flaky, "t::b": flaky}, flaky_only=True)
411 out = capsys.readouterr().out
412 assert "t::b" in out
413 assert "t::a" not in out
414
415
416 # ──────────────────────────────────────────────────────────────────────────────
417 # Unit — _print_pre_run
418 # ──────────────────────────────────────────────────────────────────────────────
419
420
421 class TestPrintPreRun:
422 def test_with_selection_shows_changed_count(self, capsys) -> None:
423 from muse.cli.commands.test_cmd import _print_pre_run
424 sel = _make_selection(changed_addresses=["a.py::f", "b.py::g"])
425 _print_pre_run(sel, targets=["tests/t.py::test_1", "tests/t.py::test_2"])
426 assert "Changed symbols: 2" in capsys.readouterr().out
427
428 def test_without_selection_with_targets(self, capsys) -> None:
429 from muse.cli.commands.test_cmd import _print_pre_run
430 _print_pre_run(None, targets=["tests/t.py::test_1"])
431 assert "1 specified" in capsys.readouterr().out
432
433 def test_without_selection_without_targets(self, capsys) -> None:
434 from muse.cli.commands.test_cmd import _print_pre_run
435 _print_pre_run(None, targets=[])
436 assert "full test suite" in capsys.readouterr().out
437
438 def test_uncovered_symbols_shown(self, capsys) -> None:
439 from muse.cli.commands.test_cmd import _print_pre_run
440 sel = _make_selection(uncovered_addresses=["a.py::fn"])
441 _print_pre_run(sel, targets=[])
442 assert "no covering test" in capsys.readouterr().out or "uncovered" in capsys.readouterr().out.lower() or "⚠️" in capsys.readouterr().out
443
444 def test_fallback_note_shown(self, capsys) -> None:
445 from muse.cli.commands.test_cmd import _print_pre_run
446 sel = _make_selection(fallback_used=True)
447 _print_pre_run(sel, targets=[])
448 assert "heuristic" in capsys.readouterr().out.lower() or "fallback" in capsys.readouterr().out.lower() or "File-name" in capsys.readouterr().out
449
450
451 # ──────────────────────────────────────────────────────────────────────────────
452 # Unit — _print_dry_run
453 # ──────────────────────────────────────────────────────────────────────────────
454
455
456 class TestPrintDryRun:
457 def test_human_with_targets_shows_would_run(self, capsys) -> None:
458 from muse.cli.commands.test_cmd import _print_dry_run
459 _print_dry_run(None, targets=["tests/t.py::test_x"], json_out=False)
460 assert "Would run" in capsys.readouterr().out
461
462 def test_human_no_targets_shows_full_discovery(self, capsys) -> None:
463 from muse.cli.commands.test_cmd import _print_dry_run
464 _print_dry_run(None, targets=[], json_out=False)
465 assert "full discovery" in capsys.readouterr().out
466
467 def test_json_mode_emits_mode_dry_run(self, capsys) -> None:
468 from muse.cli.commands.test_cmd import _print_dry_run
469 _print_dry_run(None, targets=[], json_out=True)
470 d = json.loads(capsys.readouterr().out)
471 assert d["mode"] == "dry-run"
472
473 def test_json_mode_with_selection(self, capsys) -> None:
474 from muse.cli.commands.test_cmd import _print_dry_run
475 sel = _make_selection(changed_addresses=["a.py::f"])
476 _print_dry_run(sel, targets=["tests/t.py::test_x"], json_out=True)
477 d = json.loads(capsys.readouterr().out)
478 assert "selection" in d
479 assert d["selection"]["targets"] == ["tests/t.py::test_x"]
480
481
482 # ──────────────────────────────────────────────────────────────────────────────
483 # Unit — _print_summary
484 # ──────────────────────────────────────────────────────────────────────────────
485
486
487 class TestPrintSummary:
488 def test_passed_shows_checkmark(self, capsys) -> None:
489 from muse.cli.commands.test_cmd import _print_summary
490 result = _make_run_result(exit_code=0, passed=5, failed=0)
491 _print_summary(result, None)
492 assert "✅" in capsys.readouterr().out
493
494 def test_failed_shows_x(self, capsys) -> None:
495 from muse.cli.commands.test_cmd import _print_summary
496 result = _make_run_result(exit_code=1, passed=2, failed=1)
497 _print_summary(result, None)
498 assert "❌" in capsys.readouterr().out
499
500 def test_timed_out_shows_warning(self, capsys) -> None:
501 from muse.cli.commands.test_cmd import _print_summary
502 result = _make_run_result(exit_code=1, timed_out=True)
503 _print_summary(result, None)
504 assert "timeout" in capsys.readouterr().out.lower() or "terminated" in capsys.readouterr().out.lower()
505
506 def test_uncovered_addresses_shown(self, capsys) -> None:
507 from muse.cli.commands.test_cmd import _print_summary
508 sel = _make_selection(uncovered_addresses=["a.py::fn"])
509 result = _make_run_result()
510 _print_summary(result, sel)
511 out = capsys.readouterr().out
512 assert "a.py::fn" in out or "Coverage gap" in out or "changed symbol" in out
513
514
515 # ──────────────────────────────────────────────────────────────────────────────
516 # Integration — alias, docstrings, envelope
517 # ──────────────────────────────────────────────────────────────────────────────
518
519
520 class TestAliasRegistration:
521 def test_j_alias_registered(self) -> None:
522 from muse.cli.commands.test_cmd import register
523 import argparse
524 p = argparse.ArgumentParser()
525 sub = p.add_subparsers()
526 register(sub)
527 ns = p.parse_args(["test", "-j"])
528 assert ns.json_output is True
529
530 def test_json_long_form_works(self) -> None:
531 from muse.cli.commands.test_cmd import register
532 import argparse
533 p = argparse.ArgumentParser()
534 sub = p.add_subparsers()
535 register(sub)
536 ns = p.parse_args(["test", "--json"])
537 assert ns.json_output is True
538
539
540 class TestDocstrings:
541 def test_register_mentions_json_alias(self) -> None:
542 from muse.cli.commands.test_cmd import register
543 doc = register.__doc__ or ""
544 assert "--json" in doc or "-j" in doc
545
546 def test_run_mentions_schema_version(self) -> None:
547 from muse.cli.commands.test_cmd import run
548 assert "schema_version" in (run.__doc__ or "")
549
550 def test_run_mentions_exit_code(self) -> None:
551 from muse.cli.commands.test_cmd import run
552 assert "exit_code" in (run.__doc__ or "")
553
554 def test_run_mentions_duration_ms(self) -> None:
555 from muse.cli.commands.test_cmd import run
556 assert "duration_ms" in (run.__doc__ or "")
557
558
559 # ──────────────────────────────────────────────────────────────────────────────
560 # End-to-end
561 # ──────────────────────────────────────────────────────────────────────────────
562
563
564 class TestEndToEnd:
565 def test_history_exits_zero_empty(self, test_repo) -> None:
566 r = _invoke(test_repo, ["code", "test", "--history"])
567 assert r.exit_code == 0
568 assert "No test history" in r.output
569
570 def test_flaky_exits_zero_empty(self, test_repo) -> None:
571 r = _invoke(test_repo, ["code", "test", "--flaky"])
572 assert r.exit_code == 0
573 assert "No flaky" in r.output
574
575 def test_history_json_emits_mode_history(self, test_repo) -> None:
576 r = _invoke(test_repo, ["code", "test", "--history", "--json"])
577 assert r.exit_code == 0
578 d = json.loads(r.output)
579 assert d["mode"] == "history"
580
581 def test_history_json_has_schema_version(self, test_repo) -> None:
582 r = _invoke(test_repo, ["code", "test", "--history", "--json"])
583 assert r.exit_code == 0
584 assert "schema_version" in json.loads(r.output)
585
586 def test_history_json_has_history_list(self, test_repo) -> None:
587 r = _invoke(test_repo, ["code", "test", "--history", "--json"])
588 assert r.exit_code == 0
589 d = json.loads(r.output)
590 assert isinstance(d["history"], list)
591
592 def test_dry_run_exits_zero(self, test_repo) -> None:
593 r = _invoke(test_repo, ["code", "test", "--dry-run", "--all"])
594 assert r.exit_code == 0
595
596 def test_dry_run_shows_would_run(self, test_repo) -> None:
597 r = _invoke(test_repo, ["code", "test", "--dry-run", "--all"])
598 assert r.exit_code == 0
599 assert "Would run" in r.output or "dry" in r.output.lower() or "full discovery" in r.output
600
601 def test_dry_run_json_emits_mode(self, test_repo) -> None:
602 r = _invoke(test_repo, ["code", "test", "--dry-run", "--all", "--json"])
603 assert r.exit_code == 0
604 d = json.loads(r.output)
605 assert d["mode"] == "dry-run"
606
607 def test_dry_run_json_has_schema_version(self, test_repo) -> None:
608 r = _invoke(test_repo, ["code", "test", "--dry-run", "--all", "--json"])
609 assert r.exit_code == 0
610 assert "schema_version" in json.loads(r.output)
611
612 def test_j_alias_dry_run(self, test_repo) -> None:
613 r = _invoke(test_repo, ["code", "test", "--dry-run", "--all", "-j"])
614 assert r.exit_code == 0
615 d = json.loads(r.output)
616 assert d["mode"] == "dry-run"
617
618 def test_flaky_json_has_schema_version(self, test_repo) -> None:
619 r = _invoke(test_repo, ["code", "test", "--flaky", "--json"])
620 assert r.exit_code == 0
621 assert "schema_version" in json.loads(r.output)
622
623 def test_no_changes_detected_json(self, test_repo) -> None:
624 """Clean working tree with --json emits 'no changes detected' message."""
625 r = _invoke(test_repo, ["code", "test", "--json"])
626 assert r.exit_code == 0
627 d = json.loads(r.output)
628 # Either no-changes or actually ran tests — both are valid
629 assert "mode" in d or "message" in d
630
631
632 # ──────────────────────────────────────────────────────────────────────────────
633 # Stress
634 # ──────────────────────────────────────────────────────────────────────────────
635
636
637 class TestStress:
638 def test_1000_history_to_json(self) -> None:
639 from muse.cli.commands.test_cmd import _history_to_json
640 s = _make_history_summary()
641 for _ in range(1_000):
642 d = _history_to_json(s)
643 assert d["node_id"] == s["node_id"]
644
645 def test_500_gate_to_json(self) -> None:
646 from muse.cli.commands.test_cmd import _gate_to_json
647 g = _make_gate()
648 for _ in range(500):
649 d = _gate_to_json(g)
650 assert d["name"] == g["name"]
651
652 def test_print_history_200_entries(self, capsys) -> None:
653 from muse.cli.commands.test_cmd import _print_history
654 summaries = {
655 f"t::test_{i}": _make_history_summary(node_id=f"t::test_{i}")
656 for i in range(200)
657 }
658 _print_history(summaries, flaky_only=False)
659 out = capsys.readouterr().out
660 assert "test_0" in out
661
662 def test_concurrent_history_to_json(self) -> None:
663 from muse.cli.commands.test_cmd import _history_to_json
664 s = _make_history_summary()
665 results: list[str] = []
666 lock = threading.Lock()
667
668 def _run() -> None:
669 d = _history_to_json(s)
670 with lock:
671 results.append(d["node_id"])
672
673 threads = [threading.Thread(target=_run) for _ in range(50)]
674 for t in threads: t.start()
675 for t in threads: t.join()
676 assert len(results) == 50
677 assert all(r == s["node_id"] for r in results)
678
679
680 # ──────────────────────────────────────────────────────────────────────────────
681 # Data integrity
682 # ──────────────────────────────────────────────────────────────────────────────
683
684
685 class TestDataIntegrity:
686 def test_schema_version_is_string(self, test_repo) -> None:
687 r = _invoke(test_repo, ["code", "test", "--history", "--json"])
688 assert r.exit_code == 0
689 d = json.loads(r.output)
690 assert isinstance(d.get("schema_version"), str)
691
692 def test_schema_version_nonempty(self, test_repo) -> None:
693 r = _invoke(test_repo, ["code", "test", "--history", "--json"])
694 assert r.exit_code == 0
695 assert len(json.loads(r.output).get("schema_version", "")) > 0
696
697 def test_mode_field_present_in_all_json_modes(self, test_repo) -> None:
698 for extra in [["--history"], ["--flaky"], ["--dry-run", "--all"]]:
699 r = _invoke(test_repo, ["code", "test", *extra, "--json"])
700 assert r.exit_code == 0, f"failed for {extra}: {r.output}"
701 d = json.loads(r.output)
702 assert "mode" in d, f"missing mode for {extra}"
703
704 def test_history_to_json_all_fields(self) -> None:
705 from muse.cli.commands.test_cmd import _history_to_json
706 d = _history_to_json(_make_history_summary())
707 for field in ("node_id", "total_runs", "pass_count", "fail_count",
708 "skip_count", "flaky", "avg_duration_ms",
709 "last_outcome", "last_run_timestamp", "fail_streak"):
710 assert field in d, f"missing: {field}"
711
712 def test_gate_to_json_all_required_fields(self) -> None:
713 from muse.cli.commands.test_cmd import _gate_to_json
714 d = _gate_to_json(_make_gate())
715 for field in ("name", "command", "exit_code", "duration_ms",
716 "required", "passed", "timed_out", "stdout", "stderr"):
717 assert field in d, f"missing: {field}"
718
719 def test_ci_to_json_preserves_timestamp(self) -> None:
720 from muse.cli.commands.test_cmd import _ci_to_json
721 ts = "2026-04-19T10:00:00+00:00"
722 d = _ci_to_json(_make_ci_result(timestamp=ts))
723 assert d["timestamp"] == ts
724
725 def test_dry_run_json_serialisable(self, test_repo) -> None:
726 r = _invoke(test_repo, ["code", "test", "--dry-run", "--all", "--json"])
727 assert r.exit_code == 0
728 json.loads(r.output) # must not raise
729
730
731 # ──────────────────────────────────────────────────────────────────────────────
732 # Security
733 # ──────────────────────────────────────────────────────────────────────────────
734
735
736 class TestSecurity:
737 def test_hostile_node_id_survives_json(self) -> None:
738 from muse.cli.commands.test_cmd import _history_to_json
739 evil = '"; DROP TABLE history; --'
740 d = _history_to_json(_make_history_summary(node_id=evil))
741 assert json.loads(json.dumps(d))["node_id"] == evil
742
743 def test_ansi_in_gate_stdout_survives_json(self) -> None:
744 from muse.cli.commands.test_cmd import _gate_to_json
745 evil_stdout = "\x1b[31merror\x1b[0m"
746 d = _gate_to_json(_make_gate(stdout=evil_stdout))
747 assert json.loads(json.dumps(d))["stdout"] == evil_stdout
748
749 def test_very_long_gate_name_does_not_crash(self) -> None:
750 from muse.cli.commands.test_cmd import _gate_to_json
751 d = _gate_to_json(_make_gate(name="x" * 10_000))
752 assert len(d["name"]) == 10_000
753
754 def test_sql_injection_in_fatal_msg_does_not_crash(self, capsys) -> None:
755 from muse.cli.commands.test_cmd import _fatal
756 evil = "'; DROP TABLE commits; --"
757 with pytest.raises(SystemExit):
758 _fatal(evil, json_out=True)
759 d = json.loads(capsys.readouterr().out)
760 assert d["error"] == evil
761
762 def test_unicode_in_history_node_id(self) -> None:
763 from muse.cli.commands.test_cmd import _history_to_json
764 d = _history_to_json(_make_history_summary(node_id="tests/音符.py::test_関数"))
765 assert json.loads(json.dumps(d))["node_id"] == "tests/音符.py::test_関数"
766
767 def test_null_byte_in_gate_stderr_does_not_crash(self) -> None:
768 from muse.cli.commands.test_cmd import _gate_to_json
769 d = _gate_to_json(_make_gate(stderr="err\x00byte"))
770 assert "err" in json.dumps(d)
771
772
773 # ──────────────────────────────────────────────────────────────────────────────
774 # Performance
775 # ──────────────────────────────────────────────────────────────────────────────
776
777
778 class TestPerformance:
779 def test_1000_gate_to_json_under_500ms(self) -> None:
780 from muse.cli.commands.test_cmd import _gate_to_json
781 g = _make_gate()
782 start = time.perf_counter()
783 for _ in range(1_000):
784 _gate_to_json(g)
785 elapsed = time.perf_counter() - start
786 assert elapsed < 0.5, f"1 000 _gate_to_json took {elapsed:.2f}s"
787
788 def test_1000_history_to_json_under_500ms(self) -> None:
789 from muse.cli.commands.test_cmd import _history_to_json
790 s = _make_history_summary()
791 start = time.perf_counter()
792 for _ in range(1_000):
793 _history_to_json(s)
794 elapsed = time.perf_counter() - start
795 assert elapsed < 0.5, f"1 000 _history_to_json took {elapsed:.2f}s"
796
797 def test_dry_run_completes_quickly(self, test_repo) -> None:
798 start = time.perf_counter()
799 r = _invoke(test_repo, ["code", "test", "--dry-run", "--all", "--json"])
800 elapsed = time.perf_counter() - start
801 assert r.exit_code == 0
802 assert elapsed < 10.0, f"--dry-run took {elapsed:.2f}s"
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 140 days ago