gabriel / muse public
test_stress_code_analytics.py python
661 lines 26.3 KB
Raw
sha256:74b5023693ac2ab80e3b89fddc66e0d60d7d931a1266d3f9294f645c3102fe76 tests/test_lineage_supercharge.py, tests/test_narrative_sup… Human 4 days ago
1 """Stress tests for all history-walking ``muse code`` analytics commands.
2
3 The mega-suite (test_code_commands.py) has comprehensive CLI integration
4 coverage for each command. This file adds the missing stress layer:
5 a large repo with many commits and symbols to verify that each command
6 completes in a reasonable time, emits valid JSON, and handles filters
7 correctly under load.
8
9 Stress budget per command: < 15 s on a standard developer machine.
10
11 Commands covered
12 ----------------
13 muse code hotspots — symbol churn leaderboard
14 muse code stable — symbol stability leaderboard
15 muse code coupling — file-level co-change
16 muse code compare — symbol diff between refs
17 muse code blast-risk — impact risk score
18 muse code age — symbol age / staleness
19 muse code velocity — commit velocity by module
20 muse code entangle — cross-module coupling
21 muse code narrative — prose timeline
22 muse code gravity — call-graph gravity
23 muse code contract — type/call contract inference
24 muse code predict — change prediction
25 muse code semantic-test-coverage — test coverage mapping
26 """
27
28 from __future__ import annotations
29
30 import json
31 import pathlib
32 import textwrap
33 import time
34 from collections.abc import Generator
35
36 import pytest
37
38 from tests.cli_test_helper import CliRunner
39
40 cli = None
41 runner = CliRunner()
42
43 # ── Time budget (generous to avoid flakiness on CI) ─────────────────────────
44
45 _TIME_BUDGET_S: float = 15.0
46
47
48 # ---------------------------------------------------------------------------
49 # Shared large repo fixture — built once per module
50 # ---------------------------------------------------------------------------
51
52
53 @pytest.fixture(scope="module")
54 def large_repo(tmp_path_factory: pytest.TempPathFactory) -> Generator[pathlib.Path, None, None]:
55 """Repo with 30 commits across 6 Python modules (~180 total symbol events).
56
57 Module relationships:
58 billing.py imports from utils.py
59 auth.py imports from utils.py
60 report.py imports from billing.py and auth.py
61 api.py imports from report.py
62 tests/test_billing.py imports from billing.py
63 """
64 tmp_path = tmp_path_factory.mktemp("large_analytics_repo")
65
66 import os
67 _prev_root = os.environ.get("MUSE_REPO_ROOT")
68 _prev_cwd = os.getcwd()
69 os.environ["MUSE_REPO_ROOT"] = str(tmp_path)
70 os.chdir(tmp_path)
71
72 r = runner.invoke(cli, ["init", "--domain", "code"])
73 assert r.exit_code == 0, r.output
74
75 (tmp_path / "tests").mkdir()
76
77 # Commit 0 — scaffold all modules.
78 (tmp_path / "utils.py").write_text(textwrap.dedent("""\
79 def validate(amount: float) -> bool:
80 return amount > 0
81
82 def format_currency(amount: float) -> str:
83 return f"${amount:.2f}"
84 """))
85 (tmp_path / "billing.py").write_text(textwrap.dedent("""\
86 from utils import validate, format_currency
87
88 class Invoice:
89 def compute_total(self, items: list[float]) -> float:
90 return sum(items)
91
92 def apply_discount(self, total: float, rate: float) -> float:
93 return total * (1 - rate)
94
95 def process(invoice: Invoice, items: list[float]) -> str:
96 total = invoice.compute_total(items)
97 return format_currency(total)
98 """))
99 (tmp_path / "auth.py").write_text(textwrap.dedent("""\
100 from utils import validate
101
102 def authenticate(token: str) -> bool:
103 return validate(len(token))
104
105 def authorise(user: str, action: str) -> bool:
106 return bool(user) and bool(action)
107 """))
108 (tmp_path / "report.py").write_text(textwrap.dedent("""\
109 from billing import Invoice, process
110 from auth import authenticate
111
112 def generate_report(user: str, items: list[float]) -> dict[str, str | float]:
113 if not authenticate(user):
114 return {}
115 inv = Invoice()
116 return {"total": inv.compute_total(items), "result": process(inv, items)}
117 """))
118 (tmp_path / "api.py").write_text(textwrap.dedent("""\
119 from report import generate_report
120
121 def handle_request(user: str, payload: list[float]) -> dict[str, str | float]:
122 return generate_report(user, payload)
123
124 def health_check() -> bool:
125 return True
126 """))
127 (tmp_path / "tests" / "test_billing.py").write_text(textwrap.dedent("""\
128 from billing import Invoice, process
129
130 def test_compute_total() -> None:
131 inv = Invoice()
132 assert inv.compute_total([1.0, 2.0]) == 3.0
133
134 def test_apply_discount() -> None:
135 inv = Invoice()
136 assert inv.apply_discount(100.0, 0.1) == 90.0
137
138 def test_process() -> None:
139 inv = Invoice()
140 assert "$" in process(inv, [10.0])
141 """))
142 runner.invoke(cli, ["code", "add", "."])
143 r = runner.invoke(cli, ["commit", "-m", "scaffold all modules"])
144 assert r.exit_code == 0, r.output
145
146 # Commits 1–9 — iterative modifications to billing.py
147 discounts = [0.05, 0.10, 0.15, 0.20, 0.25, 0.30, 0.35, 0.40, 0.45]
148 for i, rate in enumerate(discounts):
149 (tmp_path / "billing.py").write_text(textwrap.dedent(f"""\
150 from utils import validate, format_currency
151
152 class Invoice:
153 def compute_total(self, items: list[float]) -> float:
154 return sum(items)
155
156 def apply_discount(self, total: float, rate: float = {rate}) -> float:
157 return total * (1 - rate)
158
159 def process(invoice: Invoice, items: list[float]) -> str:
160 total = invoice.compute_total(items)
161 if not validate(total):
162 raise ValueError("invalid total")
163 return format_currency(total)
164 """))
165 runner.invoke(cli, ["code", "add", "billing.py"])
166 r = runner.invoke(cli, ["commit", "-m", f"billing: update discount default to {rate}"])
167 assert r.exit_code == 0, r.output
168
169 # Commits 10–14 — iterative modifications to auth.py
170 for i in range(5):
171 (tmp_path / "auth.py").write_text(textwrap.dedent(f"""\
172 from utils import validate
173
174 MIN_TOKEN_LEN: int = {8 + i}
175
176 def authenticate(token: str) -> bool:
177 return validate(len(token) - MIN_TOKEN_LEN)
178
179 def authorise(user: str, action: str) -> bool:
180 return bool(user) and bool(action) and len(user) > {i}
181 """))
182 runner.invoke(cli, ["code", "add", "auth.py"])
183 r = runner.invoke(cli, ["commit", "-m", f"auth: tighten token min length to {8 + i}"])
184 assert r.exit_code == 0, r.output
185
186 # Commits 15–19 — add helpers to utils.py
187 helpers = ["clamp", "round_up", "is_positive", "is_negative", "safe_divide"]
188 utils_lines = [
189 "def validate(amount: float) -> bool:\n return amount > 0\n",
190 "def format_currency(amount: float) -> str:\n return f'${amount:.2f}'\n",
191 ]
192 for i, helper in enumerate(helpers):
193 utils_lines.append(f"def {helper}(x: float) -> float:\n return float(x)\n")
194 (tmp_path / "utils.py").write_text("\n".join(utils_lines))
195 runner.invoke(cli, ["code", "add", "utils.py"])
196 r = runner.invoke(cli, ["commit", "-m", f"utils: add {helper}"])
197 assert r.exit_code == 0, r.output
198
199 # Commits 20–24 — add API endpoints to api.py
200 for i in range(5):
201 (tmp_path / "api.py").write_text(textwrap.dedent(f"""\
202 from report import generate_report
203
204 def handle_request(user: str, payload: list[float]) -> dict[str, str | float]:
205 return generate_report(user, payload)
206
207 def health_check() -> bool:
208 return True
209
210 def endpoint_{i:02d}(x: float) -> float:
211 return float(x)
212 """))
213 runner.invoke(cli, ["code", "add", "api.py"])
214 r = runner.invoke(cli, ["commit", "-m", f"api: add endpoint_{i:02d}"])
215 assert r.exit_code == 0, r.output
216
217 # Commits 25–29 — modify report.py
218 for i in range(5):
219 (tmp_path / "report.py").write_text(textwrap.dedent(f"""\
220 from billing import Invoice, process
221 from auth import authenticate
222
223 REPORT_VERSION: int = {i + 1}
224
225 def generate_report(user: str, items: list[float]) -> dict[str, str | float | int]:
226 if not authenticate(user):
227 return {{}}
228 inv = Invoice()
229 return {{
230 "version": REPORT_VERSION,
231 "total": inv.compute_total(items),
232 "result": process(inv, items),
233 }}
234 """))
235 runner.invoke(cli, ["code", "add", "report.py"])
236 r = runner.invoke(cli, ["commit", "-m", f"report: version {i + 1}"])
237 assert r.exit_code == 0, r.output
238
239 yield tmp_path
240
241 # Restore environment so subsequent test modules see a clean state.
242 os.chdir(_prev_cwd)
243 if _prev_root is None:
244 os.environ.pop("MUSE_REPO_ROOT", None)
245 else:
246 os.environ["MUSE_REPO_ROOT"] = _prev_root
247
248
249 # ---------------------------------------------------------------------------
250 # Helper
251 # ---------------------------------------------------------------------------
252
253
254 def _assert_under_budget(elapsed: float, cmd: str) -> None:
255 assert elapsed < _TIME_BUDGET_S, (
256 f"'{cmd}' took {elapsed:.1f}s — exceeds {_TIME_BUDGET_S}s budget"
257 )
258
259
260 # ---------------------------------------------------------------------------
261 # muse code hotspots — stress
262 # ---------------------------------------------------------------------------
263
264
265 class TestHotspotsStress:
266 def test_exits_zero_on_large_repo(self, large_repo: pathlib.Path) -> None:
267 result = runner.invoke(cli, ["code", "hotspots"])
268 assert result.exit_code == 0, result.output
269
270 def test_json_schema_correct(self, large_repo: pathlib.Path) -> None:
271 result = runner.invoke(cli, ["code", "hotspots", "--json"])
272 assert result.exit_code == 0, result.output
273 data = json.loads(result.output)
274 assert "hotspots" in data
275 assert "commits_analysed" in data
276
277 def test_completes_under_budget(self, large_repo: pathlib.Path) -> None:
278 start = time.monotonic()
279 runner.invoke(cli, ["code", "hotspots", "--json"])
280 _assert_under_budget(time.monotonic() - start, "hotspots --json")
281
282 def test_top_n_respected(self, large_repo: pathlib.Path) -> None:
283 result = runner.invoke(cli, ["code", "hotspots", "--top", "5", "--json"])
284 assert result.exit_code == 0
285 data = json.loads(result.output)
286 assert len(data["hotspots"]) <= 5
287
288 def test_kind_filter(self, large_repo: pathlib.Path) -> None:
289 result = runner.invoke(cli, ["code", "hotspots", "--kind", "function", "--json"])
290 assert result.exit_code == 0
291 data = json.loads(result.output)
292 for h in data["hotspots"]:
293 assert h.get("kind") in (None, "function", "method")
294
295 def test_min_changes_filter(self, large_repo: pathlib.Path) -> None:
296 result = runner.invoke(cli, ["code", "hotspots", "--min", "5", "--json"])
297 assert result.exit_code == 0
298
299 def test_from_to_range(self, large_repo: pathlib.Path) -> None:
300 result = runner.invoke(cli, ["code", "hotspots", "--from", "HEAD~10", "--to", "HEAD", "--json"])
301 assert result.exit_code == 0
302
303
304 # ---------------------------------------------------------------------------
305 # muse code stable — stress
306 # ---------------------------------------------------------------------------
307
308
309 class TestStableStress:
310 def test_exits_zero_on_large_repo(self, large_repo: pathlib.Path) -> None:
311 result = runner.invoke(cli, ["code", "stable"])
312 assert result.exit_code == 0, result.output
313
314 def test_json_schema_correct(self, large_repo: pathlib.Path) -> None:
315 result = runner.invoke(cli, ["code", "stable", "--json"])
316 assert result.exit_code == 0, result.output
317 data = json.loads(result.output)
318 assert "stable" in data
319 assert "commits_analysed" in data
320
321 def test_completes_under_budget(self, large_repo: pathlib.Path) -> None:
322 start = time.monotonic()
323 runner.invoke(cli, ["code", "stable", "--json"])
324 _assert_under_budget(time.monotonic() - start, "stable --json")
325
326 def test_top_n_respected(self, large_repo: pathlib.Path) -> None:
327 result = runner.invoke(cli, ["code", "stable", "--top", "3", "--json"])
328 assert result.exit_code == 0
329 data = json.loads(result.output)
330 assert len(data["stable"]) <= 3
331
332 def test_kind_filter_function(self, large_repo: pathlib.Path) -> None:
333 result = runner.invoke(cli, ["code", "stable", "--kind", "function", "--json"])
334 assert result.exit_code == 0
335
336
337 # ---------------------------------------------------------------------------
338 # muse code coupling — stress
339 # ---------------------------------------------------------------------------
340
341
342 class TestCouplingStress:
343 def test_exits_zero_on_large_repo(self, large_repo: pathlib.Path) -> None:
344 result = runner.invoke(cli, ["code", "coupling"])
345 assert result.exit_code == 0, result.output
346
347 def test_json_schema_correct(self, large_repo: pathlib.Path) -> None:
348 result = runner.invoke(cli, ["code", "coupling", "--json"])
349 assert result.exit_code == 0
350 data = json.loads(result.output)
351 assert "pairs" in data
352
353 def test_completes_under_budget(self, large_repo: pathlib.Path) -> None:
354 start = time.monotonic()
355 runner.invoke(cli, ["code", "coupling", "--json"])
356 _assert_under_budget(time.monotonic() - start, "coupling --json")
357
358 def test_top_n_respected(self, large_repo: pathlib.Path) -> None:
359 result = runner.invoke(cli, ["code", "coupling", "--top", "5", "--json"])
360 assert result.exit_code == 0
361 data = json.loads(result.output)
362 assert len(data["pairs"]) <= 5
363
364 def test_min_filter(self, large_repo: pathlib.Path) -> None:
365 result = runner.invoke(cli, ["code", "coupling", "--min", "2", "--json"])
366 assert result.exit_code == 0
367 data = json.loads(result.output)
368 for c in data["pairs"]:
369 assert c["count"] >= 2
370
371
372 # ---------------------------------------------------------------------------
373 # muse code compare — stress
374 # ---------------------------------------------------------------------------
375
376
377 class TestCompareStress:
378 def test_exits_zero_on_large_repo(self, large_repo: pathlib.Path) -> None:
379 result = runner.invoke(cli, ["code", "compare", "HEAD~5", "HEAD"])
380 assert result.exit_code == 0, result.output
381
382 def test_json_schema_correct(self, large_repo: pathlib.Path) -> None:
383 result = runner.invoke(cli, ["code", "compare", "HEAD~5", "HEAD", "--json"])
384 assert result.exit_code == 0
385 data = json.loads(result.output)
386 # compare returns from/to refs and an ops list (or stat breakdown).
387 assert "from" in data or "ops" in data or "stat" in data
388
389 def test_completes_under_budget(self, large_repo: pathlib.Path) -> None:
390 start = time.monotonic()
391 runner.invoke(cli, ["code", "compare", "HEAD~10", "HEAD", "--json"])
392 _assert_under_budget(time.monotonic() - start, "compare HEAD~10 HEAD")
393
394 def test_kind_filter_function(self, large_repo: pathlib.Path) -> None:
395 result = runner.invoke(cli, [
396 "code", "compare", "HEAD~5", "HEAD", "--kind", "function", "--json",
397 ])
398 assert result.exit_code == 0
399
400 def test_same_refs_empty_diff(self, large_repo: pathlib.Path) -> None:
401 result = runner.invoke(cli, ["code", "compare", "HEAD", "HEAD", "--json"])
402 assert result.exit_code == 0
403 data = json.loads(result.output)
404 # With identical refs, ops list should be empty and stat should show 0 changes.
405 ops = data.get("ops", [])
406 assert len(ops) == 0
407
408
409 # ---------------------------------------------------------------------------
410 # muse code blast-risk — stress
411 # ---------------------------------------------------------------------------
412
413
414 class TestBlastRiskStress:
415 def test_exits_zero_on_large_repo(self, large_repo: pathlib.Path) -> None:
416 result = runner.invoke(cli, ["code", "blast-risk"])
417 assert result.exit_code == 0, result.output
418
419 def test_json_schema_correct(self, large_repo: pathlib.Path) -> None:
420 result = runner.invoke(cli, ["code", "blast-risk", "--json"])
421 assert result.exit_code == 0
422 data = json.loads(result.output)
423 assert "symbols" in data
424
425 def test_completes_under_budget(self, large_repo: pathlib.Path) -> None:
426 start = time.monotonic()
427 runner.invoke(cli, ["code", "blast-risk", "--json"])
428 _assert_under_budget(time.monotonic() - start, "blast-risk --json")
429
430 def test_top_n_respected(self, large_repo: pathlib.Path) -> None:
431 result = runner.invoke(cli, ["code", "blast-risk", "--top", "5", "--json"])
432 assert result.exit_code == 0
433 data = json.loads(result.output)
434 assert len(data["symbols"]) <= 5
435
436
437 # ---------------------------------------------------------------------------
438 # muse code age — stress
439 # ---------------------------------------------------------------------------
440
441
442 class TestAgeStress:
443 def test_exits_zero_on_large_repo(self, large_repo: pathlib.Path) -> None:
444 result = runner.invoke(cli, ["code", "age"])
445 assert result.exit_code == 0, result.output
446
447 def test_json_schema_correct(self, large_repo: pathlib.Path) -> None:
448 result = runner.invoke(cli, ["code", "age", "--json"])
449 assert result.exit_code == 0
450 data = json.loads(result.output)
451 assert "symbols" in data
452
453 def test_completes_under_budget(self, large_repo: pathlib.Path) -> None:
454 start = time.monotonic()
455 runner.invoke(cli, ["code", "age", "--json"])
456 _assert_under_budget(time.monotonic() - start, "age --json")
457
458 def test_top_n_respected(self, large_repo: pathlib.Path) -> None:
459 result = runner.invoke(cli, ["code", "age", "--top", "5", "--json"])
460 assert result.exit_code == 0
461 data = json.loads(result.output)
462 assert len(data["symbols"]) <= 5
463
464 def test_kind_filter(self, large_repo: pathlib.Path) -> None:
465 result = runner.invoke(cli, ["code", "age", "--kind", "function", "--json"])
466 assert result.exit_code == 0
467
468
469 # ---------------------------------------------------------------------------
470 # muse code velocity — stress
471 # ---------------------------------------------------------------------------
472
473
474 class TestVelocityStress:
475 def test_exits_zero_on_large_repo(self, large_repo: pathlib.Path) -> None:
476 result = runner.invoke(cli, ["code", "velocity"])
477 assert result.exit_code == 0, result.output
478
479 def test_json_schema_correct(self, large_repo: pathlib.Path) -> None:
480 result = runner.invoke(cli, ["code", "velocity", "--json"])
481 assert result.exit_code == 0
482 data = json.loads(result.output)
483 assert "modules" in data
484
485 def test_completes_under_budget(self, large_repo: pathlib.Path) -> None:
486 start = time.monotonic()
487 runner.invoke(cli, ["code", "velocity", "--json"])
488 _assert_under_budget(time.monotonic() - start, "velocity --json")
489
490 def test_window_flag(self, large_repo: pathlib.Path) -> None:
491 result = runner.invoke(cli, ["code", "velocity", "--window", "10", "--json"])
492 assert result.exit_code == 0
493
494
495 # ---------------------------------------------------------------------------
496 # muse code entangle — stress
497 # ---------------------------------------------------------------------------
498
499
500 class TestEntangleStress:
501 def test_exits_zero_on_large_repo(self, large_repo: pathlib.Path) -> None:
502 result = runner.invoke(cli, ["code", "entangle"])
503 assert result.exit_code == 0, result.output
504
505 def test_json_schema_correct(self, large_repo: pathlib.Path) -> None:
506 result = runner.invoke(cli, ["code", "entangle", "--json"])
507 assert result.exit_code == 0
508 data = json.loads(result.output)
509 assert "pairs" in data
510
511 def test_completes_under_budget(self, large_repo: pathlib.Path) -> None:
512 start = time.monotonic()
513 runner.invoke(cli, ["code", "entangle", "--json"])
514 _assert_under_budget(time.monotonic() - start, "entangle --json")
515
516 def test_top_n_respected(self, large_repo: pathlib.Path) -> None:
517 result = runner.invoke(cli, ["code", "entangle", "--top", "5", "--json"])
518 assert result.exit_code == 0
519 data = json.loads(result.output)
520 assert len(data["pairs"]) <= 5
521
522
523 # ---------------------------------------------------------------------------
524 # muse code narrative — stress
525 # ---------------------------------------------------------------------------
526
527
528 class TestNarrativeStress:
529 def test_exits_zero_on_large_repo(self, large_repo: pathlib.Path) -> None:
530 result = runner.invoke(cli, ["code", "narrative", "billing.py::Invoice.apply_discount"])
531 assert result.exit_code == 0, result.output
532
533 def test_json_schema_correct(self, large_repo: pathlib.Path) -> None:
534 result = runner.invoke(cli, [
535 "code", "narrative", "billing.py::Invoice.apply_discount", "--json",
536 ])
537 assert result.exit_code == 0
538 data = json.loads(result.output)
539 assert "address" in data
540 assert "events" in data
541
542 def test_completes_under_budget(self, large_repo: pathlib.Path) -> None:
543 start = time.monotonic()
544 runner.invoke(cli, [
545 "code", "narrative", "billing.py::Invoice.apply_discount", "--json",
546 ])
547 _assert_under_budget(time.monotonic() - start, "narrative billing.py::... --json")
548
549 def test_format_prose_mode(self, large_repo: pathlib.Path) -> None:
550 result = runner.invoke(cli, [
551 "code", "narrative", "billing.py::Invoice.apply_discount", "--format", "prose",
552 ])
553 assert result.exit_code == 0
554
555
556 # ---------------------------------------------------------------------------
557 # muse code gravity — stress
558 # ---------------------------------------------------------------------------
559
560
561 class TestGravityStress:
562 def test_exits_zero_on_large_repo(self, large_repo: pathlib.Path) -> None:
563 result = runner.invoke(cli, ["code", "gravity"])
564 assert result.exit_code == 0, result.output
565
566 def test_json_schema_correct(self, large_repo: pathlib.Path) -> None:
567 result = runner.invoke(cli, ["code", "gravity", "--json"])
568 assert result.exit_code == 0
569 data = json.loads(result.output)
570 assert "symbols" in data
571
572 def test_completes_under_budget(self, large_repo: pathlib.Path) -> None:
573 start = time.monotonic()
574 runner.invoke(cli, ["code", "gravity", "--json"])
575 _assert_under_budget(time.monotonic() - start, "gravity --json")
576
577 def test_top_n_respected(self, large_repo: pathlib.Path) -> None:
578 result = runner.invoke(cli, ["code", "gravity", "--top", "5", "--json"])
579 assert result.exit_code == 0
580 data = json.loads(result.output)
581 assert len(data["symbols"]) <= 5
582
583
584 # ---------------------------------------------------------------------------
585 # muse code contract — stress
586 # ---------------------------------------------------------------------------
587
588
589 class TestContractStress:
590 def test_exits_zero_on_large_repo(self, large_repo: pathlib.Path) -> None:
591 result = runner.invoke(cli, ["code", "contract", "billing.py::Invoice.compute_total"])
592 assert result.exit_code == 0, result.output
593
594 def test_json_schema_correct(self, large_repo: pathlib.Path) -> None:
595 result = runner.invoke(cli, [
596 "code", "contract", "billing.py::Invoice.compute_total", "--json",
597 ])
598 assert result.exit_code == 0
599 data = json.loads(result.output)
600 assert "address" in data
601
602 def test_completes_under_budget(self, large_repo: pathlib.Path) -> None:
603 start = time.monotonic()
604 runner.invoke(cli, [
605 "code", "contract", "billing.py::Invoice.compute_total", "--json",
606 ])
607 _assert_under_budget(time.monotonic() - start, "contract billing.py::... --json")
608
609
610 # ---------------------------------------------------------------------------
611 # muse code predict — stress
612 # ---------------------------------------------------------------------------
613
614
615 class TestPredictStress:
616 def test_exits_zero_on_large_repo(self, large_repo: pathlib.Path) -> None:
617 result = runner.invoke(cli, ["code", "predict"])
618 assert result.exit_code == 0, result.output
619
620 def test_json_schema_correct(self, large_repo: pathlib.Path) -> None:
621 result = runner.invoke(cli, ["code", "predict", "--json"])
622 assert result.exit_code == 0
623 data = json.loads(result.output)
624 assert "predictions" in data
625
626 def test_completes_under_budget(self, large_repo: pathlib.Path) -> None:
627 start = time.monotonic()
628 runner.invoke(cli, ["code", "predict", "--json"])
629 _assert_under_budget(time.monotonic() - start, "predict --json")
630
631 def test_top_n_respected(self, large_repo: pathlib.Path) -> None:
632 result = runner.invoke(cli, ["code", "predict", "--top", "5", "--json"])
633 assert result.exit_code == 0
634 data = json.loads(result.output)
635 assert len(data.get("predictions", [])) <= 5
636
637
638 # ---------------------------------------------------------------------------
639 # muse code semantic-test-coverage — stress
640 # ---------------------------------------------------------------------------
641
642
643 class TestSemanticTestCoverageStress:
644 def test_exits_zero_on_large_repo(self, large_repo: pathlib.Path) -> None:
645 result = runner.invoke(cli, ["code", "semantic-test-coverage"])
646 assert result.exit_code == 0, result.output
647
648 def test_json_schema_correct(self, large_repo: pathlib.Path) -> None:
649 result = runner.invoke(cli, ["code", "semantic-test-coverage", "--json"])
650 assert result.exit_code == 0
651 data = json.loads(result.output)
652 assert "files" in data or "coverage" in data or "summary" in data
653
654 def test_completes_under_budget(self, large_repo: pathlib.Path) -> None:
655 start = time.monotonic()
656 runner.invoke(cli, ["code", "semantic-test-coverage", "--json"])
657 _assert_under_budget(time.monotonic() - start, "semantic-test-coverage --json")
658
659 def test_file_filter(self, large_repo: pathlib.Path) -> None:
660 result = runner.invoke(cli, ["code", "semantic-test-coverage", "--file", "billing.py"])
661 assert result.exit_code == 0
File History 1 commit
sha256:74b5023693ac2ab80e3b89fddc66e0d60d7d931a1266d3f9294f645c3102fe76 tests/test_lineage_supercharge.py, tests/test_narrative_sup… Human 4 days ago