gabriel / muse public
test_predict_supercharge.py python
818 lines 31.6 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 143 days ago
1 """TDD supercharge tests for ``muse code predict``.
2
3 Gaps being closed
4 -----------------
5 - ``-j`` alias for ``--json``
6 - ``exit_code`` and ``duration_ms`` in JSON envelope
7 - ``--explain --json`` structured output for agents (new ``_ExplainJson``)
8 - Zero existing test coverage — unit + integration + security added
9
10 Unit tests
11 ----------
12 - ``_sanitise`` — truncation, control chars, ANSI stripping, unicode
13 - ``_module_key`` — depth variants, single-component path, nested path
14 - ``_pct_bar`` — 0.0, 0.5, 1.0, clamping outside [0, 1]
15 - ``_confidence_label`` — boundary values at 0.70 and 0.45
16 - ``_iter_symbol_ops`` — None, empty ops, patch children, non-:: filtered
17 - ``_build_predictions`` — empty commits, single op, scoring in [0, 1]
18
19 Integration tests
20 -----------------
21 - predict --json schema (all required keys present)
22 - -j alias works
23 - exit_code + duration_ms in JSON
24 - --top limits output
25 - --min-confidence filter
26 - --file filter
27 - --explain on existing symbol (human output)
28 - --explain ADDRESS --json → _ExplainJson structured output
29 - --explain missing address exits 1
30 - empty repo exits non-zero
31
32 Security tests
33 --------------
34 - --explain without ``::`` exits 1
35 - --min-confidence out of range exits 1
36 """
37
38 from __future__ import annotations
39
40 import datetime
41 import json
42 import pathlib
43 import textwrap
44 import typing
45
46 import pytest
47
48 from muse.cli.commands.predict import (
49 _build_predictions,
50 _confidence_label,
51 _iter_symbol_ops,
52 _module_key,
53 _pct_bar,
54 _sanitise,
55 )
56 from muse.core.store import CommitRecord
57 from tests.cli_test_helper import CliRunner
58
59 cli = None
60 runner = CliRunner()
61
62
63 # ---------------------------------------------------------------------------
64 # Helpers for building fake CommitRecords
65 # ---------------------------------------------------------------------------
66
67
68 def _fake_commit(
69 *,
70 commit_id: str = "sha256:aaa",
71 ops: list[dict] | None = None,
72 seconds_ago: int = 0,
73 ) -> CommitRecord:
74 """Build a minimal CommitRecord for unit testing _build_predictions."""
75 structured_delta = None
76 if ops is not None:
77 structured_delta = {
78 "domain": "code",
79 "ops": ops,
80 "summary": "",
81 "sem_ver_bump": "none",
82 "breaking_changes": [],
83 }
84 return CommitRecord(
85 commit_id=commit_id,
86 repo_id="repo-1",
87 branch="dev",
88 snapshot_id="snap-1",
89 message="test commit",
90 committed_at=datetime.datetime.now(datetime.timezone.utc)
91 - datetime.timedelta(seconds=seconds_ago),
92 structured_delta=structured_delta,
93 )
94
95
96 def _sym_op(address: str, op: str = "modify", new_summary: str = "") -> dict:
97 """Build a minimal symbol-level op dict (non-patch, has ::)."""
98 return {"op": op, "address": address, "new_summary": new_summary}
99
100
101 def _patch_op(file_addr: str, children: list[dict]) -> dict:
102 """Build a PatchOp with symbol-level child ops."""
103 return {
104 "op": "patch",
105 "address": file_addr,
106 "child_ops": children,
107 "child_domain": "code",
108 "child_summary": "",
109 "from_address": None,
110 "file_change": None,
111 }
112
113
114 # ---------------------------------------------------------------------------
115 # Unit — _sanitise
116 # ---------------------------------------------------------------------------
117
118
119 class TestSanitise:
120 def test_passthrough_normal_string(self) -> None:
121 assert _sanitise("hello world") == "hello world"
122
123 def test_strips_control_chars(self) -> None:
124 assert _sanitise("hello\x00world") == "helloworld"
125
126 def test_strips_ansi_escape(self) -> None:
127 result = _sanitise("hello\x1b[31mworld\x1b[0m")
128 assert "\x1b" not in result
129
130 def test_truncates_at_max_len(self) -> None:
131 long_str = "a" * 100
132 result = _sanitise(long_str, max_len=10)
133 assert len(result) <= 10
134 assert result.endswith("…")
135
136 def test_exact_length_no_truncation(self) -> None:
137 s = "a" * 80
138 result = _sanitise(s, max_len=80)
139 assert result == s
140
141 def test_strips_leading_trailing_whitespace(self) -> None:
142 assert _sanitise(" hello ") == "hello"
143
144 def test_unicode_passthrough(self) -> None:
145 assert _sanitise("café résumé") == "café résumé"
146
147 def test_empty_string(self) -> None:
148 assert _sanitise("") == ""
149
150
151 # ---------------------------------------------------------------------------
152 # Unit — _module_key
153 # ---------------------------------------------------------------------------
154
155
156 class TestModuleKey:
157 def test_depth_1_single_dir(self) -> None:
158 key = _module_key("muse/core/store.py::foo", depth=1)
159 assert key == "muse/"
160
161 def test_depth_2_two_dirs(self) -> None:
162 key = _module_key("muse/core/store.py::foo", depth=2)
163 assert key == "muse/core/"
164
165 def test_depth_3_deep(self) -> None:
166 key = _module_key("a/b/c/d.py::foo", depth=3)
167 assert key == "a/b/c/"
168
169 def test_single_component_no_dir(self) -> None:
170 key = _module_key("billing.py::compute", depth=2)
171 assert key == "billing.py"
172
173 def test_depth_beyond_path_length(self) -> None:
174 # Should not crash when depth exceeds actual path depth.
175 key = _module_key("a/b.py::foo", depth=10)
176 assert "/" in key or key # just must not raise
177
178 def test_strips_symbol_part(self) -> None:
179 key1 = _module_key("muse/core/store.py::func_a", depth=2)
180 key2 = _module_key("muse/core/store.py::func_b", depth=2)
181 assert key1 == key2 # same module, different symbols
182
183
184 # ---------------------------------------------------------------------------
185 # Unit — _pct_bar
186 # ---------------------------------------------------------------------------
187
188
189 class TestPctBar:
190 def test_zero_all_empty(self) -> None:
191 bar = _pct_bar(0.0, width=10)
192 assert bar == "░" * 10
193
194 def test_one_all_filled(self) -> None:
195 bar = _pct_bar(1.0, width=10)
196 assert bar == "█" * 10
197
198 def test_half_half(self) -> None:
199 bar = _pct_bar(0.5, width=10)
200 assert bar.count("█") == 5
201 assert bar.count("░") == 5
202
203 def test_below_zero_clamps(self) -> None:
204 bar = _pct_bar(-0.5, width=10)
205 assert bar == "░" * 10
206
207 def test_above_one_clamps(self) -> None:
208 bar = _pct_bar(1.5, width=10)
209 assert bar == "█" * 10
210
211 def test_width_respected(self) -> None:
212 for w in (5, 10, 20, 40):
213 bar = _pct_bar(0.7, width=w)
214 assert len(bar) == w
215
216
217 # ---------------------------------------------------------------------------
218 # Unit — _confidence_label
219 # ---------------------------------------------------------------------------
220
221
222 class TestConfidenceLabel:
223 def test_high_at_threshold(self) -> None:
224 assert _confidence_label(0.70) == "high"
225
226 def test_high_above_threshold(self) -> None:
227 assert _confidence_label(0.99) == "high"
228
229 def test_medium_just_below_high(self) -> None:
230 assert _confidence_label(0.69) == "medium"
231
232 def test_medium_at_threshold(self) -> None:
233 assert _confidence_label(0.45) == "medium"
234
235 def test_low_just_below_medium(self) -> None:
236 assert _confidence_label(0.44) == "low"
237
238 def test_low_at_zero(self) -> None:
239 assert _confidence_label(0.0) == "low"
240
241
242 # ---------------------------------------------------------------------------
243 # Unit — _iter_symbol_ops
244 # ---------------------------------------------------------------------------
245
246
247 class TestIterSymbolOps:
248 def test_none_delta_yields_nothing(self) -> None:
249 assert list(_iter_symbol_ops(None)) == []
250
251 def test_empty_ops_yields_nothing(self) -> None:
252 delta = {"domain": "code", "ops": [], "summary": "",
253 "sem_ver_bump": "none", "breaking_changes": []}
254 assert list(_iter_symbol_ops(delta)) == []
255
256 def test_non_patch_op_with_symbol_address_yielded(self) -> None:
257 op = _sym_op("billing.py::compute")
258 delta = {"domain": "code", "ops": [op], "summary": "",
259 "sem_ver_bump": "none", "breaking_changes": []}
260 result = list(_iter_symbol_ops(delta))
261 assert len(result) == 1
262 assert result[0]["address"] == "billing.py::compute"
263
264 def test_non_symbol_op_filtered_out(self) -> None:
265 op = {"op": "modify", "address": "billing.py"} # no ::
266 delta = {"domain": "code", "ops": [op], "summary": "",
267 "sem_ver_bump": "none", "breaking_changes": []}
268 assert list(_iter_symbol_ops(delta)) == []
269
270 def test_patch_op_yields_symbol_children(self) -> None:
271 child = _sym_op("billing.py::compute")
272 op = _patch_op("billing.py", [child])
273 delta = {"domain": "code", "ops": [op], "summary": "",
274 "sem_ver_bump": "none", "breaking_changes": []}
275 result = list(_iter_symbol_ops(delta))
276 assert len(result) == 1
277 assert result[0]["address"] == "billing.py::compute"
278
279 def test_patch_op_filters_non_symbol_children(self) -> None:
280 file_child = {"op": "modify", "address": "billing.py", "new_summary": ""}
281 sym_child = _sym_op("billing.py::compute")
282 op = _patch_op("billing.py", [file_child, sym_child])
283 delta = {"domain": "code", "ops": [op], "summary": "",
284 "sem_ver_bump": "none", "breaking_changes": []}
285 result = list(_iter_symbol_ops(delta))
286 assert len(result) == 1
287
288 def test_multiple_ops_all_yielded(self) -> None:
289 ops = [_sym_op(f"billing.py::func_{i}") for i in range(5)]
290 delta = {"domain": "code", "ops": ops, "summary": "",
291 "sem_ver_bump": "none", "breaking_changes": []}
292 result = list(_iter_symbol_ops(delta))
293 assert len(result) == 5
294
295
296 # ---------------------------------------------------------------------------
297 # Unit — _build_predictions
298 # ---------------------------------------------------------------------------
299
300
301 class TestBuildPredictions:
302 def test_empty_commits_returns_empty(self) -> None:
303 result = _build_predictions([], horizon=10, module_depth=2)
304 assert result == []
305
306 def test_commits_with_no_ops_returns_empty(self) -> None:
307 commits = [_fake_commit(ops=None), _fake_commit(ops=[])]
308 result = _build_predictions(commits, horizon=10, module_depth=2)
309 assert result == []
310
311 def test_single_symbol_appears_in_predictions(self) -> None:
312 op = _sym_op("billing.py::compute")
313 commits = [_fake_commit(ops=[op])]
314 result = _build_predictions(commits, horizon=10, module_depth=2)
315 addresses = [r["address"] for r in result]
316 assert "billing.py::compute" in addresses
317
318 def test_score_in_zero_one(self) -> None:
319 op = _sym_op("billing.py::compute")
320 commits = [_fake_commit(ops=[op])]
321 result = _build_predictions(commits, horizon=10, module_depth=2)
322 for r in result:
323 assert 0.0 <= r["score"] <= 1.0, f"score out of range: {r['score']}"
324
325 def test_sorted_by_score_descending(self) -> None:
326 ops = [_sym_op(f"billing.py::func_{i}") for i in range(10)]
327 # All commits different symbols — scores may vary.
328 commits = [_fake_commit(commit_id=f"sha256:{i:064}", ops=[op])
329 for i, op in enumerate(ops)]
330 result = _build_predictions(commits, horizon=10, module_depth=2)
331 scores = [r["score"] for r in result]
332 assert scores == sorted(scores, reverse=True)
333
334 def test_confidence_label_consistent_with_score(self) -> None:
335 op = _sym_op("billing.py::compute")
336 commits = [_fake_commit(ops=[op])] * 5
337 result = _build_predictions(commits, horizon=10, module_depth=2)
338 for r in result:
339 if r["score"] >= 0.70:
340 assert r["confidence"] == "high"
341 elif r["score"] >= 0.45:
342 assert r["confidence"] == "medium"
343 else:
344 assert r["confidence"] == "low"
345
346 def test_reasons_list_non_empty(self) -> None:
347 op = _sym_op("billing.py::compute")
348 commits = [_fake_commit(ops=[op])]
349 result = _build_predictions(commits, horizon=10, module_depth=2)
350 for r in result:
351 assert len(r["reasons"]) >= 1
352
353 def test_frequent_symbol_scores_higher(self) -> None:
354 """A symbol touched many times in the horizon window scores higher."""
355 freq_op = _sym_op("billing.py::hot")
356 rare_op = _sym_op("billing.py::cold")
357 # hot appears in 10 commits, cold in 1
358 commits = (
359 [_fake_commit(commit_id=f"sha256:{i:064}", ops=[freq_op])
360 for i in range(10)]
361 + [_fake_commit(commit_id=f"sha256:{10:064}", ops=[rare_op])]
362 )
363 result = _build_predictions(commits, horizon=15, module_depth=2)
364 by_addr = {r["address"]: r for r in result}
365 assert "billing.py::hot" in by_addr
366 assert "billing.py::cold" in by_addr
367 assert by_addr["billing.py::hot"]["score"] > by_addr["billing.py::cold"]["score"]
368
369 def test_horizon_window_controls_frequency_signal(self) -> None:
370 """Commits outside the horizon don't boost frequency signal."""
371 op = _sym_op("billing.py::compute")
372 # 5 commits all beyond horizon=3
373 commits = [
374 _fake_commit(commit_id=f"sha256:{i:064}", ops=[op])
375 for i in range(10)
376 ]
377 result_wide = _build_predictions(commits, horizon=10, module_depth=2)
378 result_narrow = _build_predictions(commits, horizon=2, module_depth=2)
379 # Wide horizon → more frequency signal → equal or higher score
380 by_addr_w = {r["address"]: r for r in result_wide}
381 by_addr_n = {r["address"]: r for r in result_narrow}
382 if "billing.py::compute" in by_addr_w and "billing.py::compute" in by_addr_n:
383 assert (
384 by_addr_w["billing.py::compute"]["signals"]["frequency"]
385 >= by_addr_n["billing.py::compute"]["signals"]["frequency"]
386 )
387
388 def test_co_change_partners_populated(self) -> None:
389 """Symbols that always co-occur get co_change signal and partners."""
390 op_a = _sym_op("billing.py::func_a")
391 op_b = _sym_op("billing.py::func_b")
392 commits = [
393 _fake_commit(commit_id=f"sha256:{i:064}", ops=[op_a, op_b])
394 for i in range(5)
395 ]
396 result = _build_predictions(commits, horizon=10, module_depth=2)
397 by_addr = {r["address"]: r for r in result}
398 for addr in ("billing.py::func_a", "billing.py::func_b"):
399 assert addr in by_addr
400 assert by_addr[addr]["signals"]["co_change"] > 0
401
402 def test_required_fields_present(self) -> None:
403 op = _sym_op("billing.py::compute")
404 commits = [_fake_commit(ops=[op])]
405 result = _build_predictions(commits, horizon=10, module_depth=2)
406 required = {
407 "address", "name", "kind", "file", "score", "confidence",
408 "reasons", "signals", "last_changed_commit", "last_changed_date",
409 "top_partners",
410 }
411 for r in result:
412 assert required <= set(r.keys()), f"Missing keys: {required - set(r.keys())}"
413
414 def test_signal_keys_present(self) -> None:
415 op = _sym_op("billing.py::compute")
416 commits = [_fake_commit(ops=[op])]
417 result = _build_predictions(commits, horizon=10, module_depth=2)
418 signal_keys = {"recency", "frequency", "co_change", "sig_instability", "module_velocity"}
419 for r in result:
420 assert signal_keys <= set(r["signals"].keys())
421
422
423 # ---------------------------------------------------------------------------
424 # Integration fixtures
425 # ---------------------------------------------------------------------------
426
427
428 @pytest.fixture
429 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
430 """Repo with 3 commits touching billing.py::compute repeatedly."""
431 monkeypatch.chdir(tmp_path)
432 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
433 runner.invoke(cli, ["init", "--domain", "code"])
434
435 src = tmp_path / "billing.py"
436
437 # Commit 1 — initial version
438 src.write_text("def compute(x: int) -> int:\n return x * 2\n\ndef helper() -> int:\n return 0\n")
439 runner.invoke(cli, ["commit", "-m", "initial"])
440
441 # Commit 2 — modify compute
442 src.write_text("def compute(x: int) -> int:\n return x * 3\n\ndef helper() -> int:\n return 0\n")
443 runner.invoke(cli, ["commit", "-m", "tweak compute"])
444
445 # Commit 3 — modify compute again
446 src.write_text("def compute(x: int) -> int:\n return x * 4\n\ndef helper() -> int:\n return 0\n")
447 runner.invoke(cli, ["commit", "-m", "tweak compute again"])
448
449 return tmp_path
450
451
452 # ---------------------------------------------------------------------------
453 # Integration — basic invocation
454 # ---------------------------------------------------------------------------
455
456
457 class TestPredictBasic:
458 def test_exits_zero(self, repo: pathlib.Path) -> None:
459 result = runner.invoke(cli, ["code", "predict"])
460 assert result.exit_code == 0, result.output
461
462 def test_emits_some_output(self, repo: pathlib.Path) -> None:
463 result = runner.invoke(cli, ["code", "predict"])
464 assert result.exit_code == 0
465 assert len(result.output) > 0
466
467 def test_empty_repo_exits_nonzero(
468 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
469 ) -> None:
470 monkeypatch.chdir(tmp_path)
471 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
472 runner.invoke(cli, ["init", "--domain", "code"])
473 # No commits → HEAD is None
474 result = runner.invoke(cli, ["code", "predict"])
475 assert result.exit_code != 0
476
477
478 # ---------------------------------------------------------------------------
479 # Integration — --json schema
480 # ---------------------------------------------------------------------------
481
482
483 class TestPredictJsonSchema:
484 def test_json_exits_zero(self, repo: pathlib.Path) -> None:
485 result = runner.invoke(cli, ["code", "predict", "--json"])
486 assert result.exit_code == 0, result.output
487
488 def test_json_is_valid(self, repo: pathlib.Path) -> None:
489 result = runner.invoke(cli, ["code", "predict", "--json"])
490 data = json.loads(result.output.strip())
491 assert isinstance(data, dict)
492
493 def test_json_has_generated_at(self, repo: pathlib.Path) -> None:
494 result = runner.invoke(cli, ["code", "predict", "--json"])
495 data = json.loads(result.output)
496 assert "generated_at" in data
497
498 def test_json_has_horizon_commits(self, repo: pathlib.Path) -> None:
499 result = runner.invoke(cli, ["code", "predict", "--json"])
500 data = json.loads(result.output)
501 assert "horizon_commits" in data
502 assert isinstance(data["horizon_commits"], int)
503
504 def test_json_has_commits_analysed(self, repo: pathlib.Path) -> None:
505 result = runner.invoke(cli, ["code", "predict", "--json"])
506 data = json.loads(result.output)
507 assert "commits_analysed" in data
508
509 def test_json_has_truncated(self, repo: pathlib.Path) -> None:
510 result = runner.invoke(cli, ["code", "predict", "--json"])
511 data = json.loads(result.output)
512 assert "truncated" in data
513 assert isinstance(data["truncated"], bool)
514
515 def test_json_has_predictions_list(self, repo: pathlib.Path) -> None:
516 result = runner.invoke(cli, ["code", "predict", "--json"])
517 data = json.loads(result.output)
518 assert "predictions" in data
519 assert isinstance(data["predictions"], list)
520
521 def test_json_has_exit_code(self, repo: pathlib.Path) -> None:
522 result = runner.invoke(cli, ["code", "predict", "--json"])
523 data = json.loads(result.output)
524 assert "exit_code" in data
525
526 def test_json_exit_code_is_zero(self, repo: pathlib.Path) -> None:
527 result = runner.invoke(cli, ["code", "predict", "--json"])
528 data = json.loads(result.output)
529 assert data["exit_code"] == 0
530
531 def test_json_has_duration_ms(self, repo: pathlib.Path) -> None:
532 result = runner.invoke(cli, ["code", "predict", "--json"])
533 data = json.loads(result.output)
534 assert "duration_ms" in data
535 assert isinstance(data["duration_ms"], float)
536 assert data["duration_ms"] > 0
537
538 def test_prediction_record_schema(self, repo: pathlib.Path) -> None:
539 result = runner.invoke(cli, ["code", "predict", "--json"])
540 data = json.loads(result.output)
541 if not data["predictions"]:
542 pytest.skip("no predictions in this repo")
543 pred = data["predictions"][0]
544 required = {
545 "address", "name", "kind", "file", "score", "confidence",
546 "reasons", "signals", "last_changed_commit", "last_changed_date",
547 "top_partners",
548 }
549 assert required <= set(pred.keys())
550
551 def test_signal_set_schema(self, repo: pathlib.Path) -> None:
552 result = runner.invoke(cli, ["code", "predict", "--json"])
553 data = json.loads(result.output)
554 if not data["predictions"]:
555 pytest.skip("no predictions in this repo")
556 signals = data["predictions"][0]["signals"]
557 assert set(signals.keys()) == {
558 "recency", "frequency", "co_change", "sig_instability", "module_velocity"
559 }
560
561
562 # ---------------------------------------------------------------------------
563 # Integration — -j alias
564 # ---------------------------------------------------------------------------
565
566
567 class TestJsonAlias:
568 def test_j_alias_exits_zero(self, repo: pathlib.Path) -> None:
569 result = runner.invoke(cli, ["code", "predict", "-j"])
570 assert result.exit_code == 0, result.output
571
572 def test_j_alias_emits_valid_json(self, repo: pathlib.Path) -> None:
573 result = runner.invoke(cli, ["code", "predict", "-j"])
574 data = json.loads(result.output.strip())
575 assert isinstance(data, dict)
576
577 def test_j_alias_matches_json_flag(self, repo: pathlib.Path) -> None:
578 r1 = runner.invoke(cli, ["code", "predict", "-j"])
579 r2 = runner.invoke(cli, ["code", "predict", "--json"])
580 d1 = json.loads(r1.output)
581 d2 = json.loads(r2.output)
582 # Dynamic fields differ; structural shape must match.
583 d1.pop("generated_at", None)
584 d2.pop("generated_at", None)
585 d1.pop("duration_ms", None)
586 d2.pop("duration_ms", None)
587 assert set(d1.keys()) == set(d2.keys())
588
589
590 # ---------------------------------------------------------------------------
591 # Integration — filters
592 # ---------------------------------------------------------------------------
593
594
595 class TestPredictFilters:
596 def test_top_limits_predictions(self, repo: pathlib.Path) -> None:
597 result = runner.invoke(cli, ["code", "predict", "--json", "--top", "1"])
598 data = json.loads(result.output)
599 assert len(data["predictions"]) <= 1
600
601 def test_top_zero_shows_all(self, repo: pathlib.Path) -> None:
602 r_all = runner.invoke(cli, ["code", "predict", "--json", "--top", "0"])
603 r_one = runner.invoke(cli, ["code", "predict", "--json", "--top", "1"])
604 d_all = json.loads(r_all.output)
605 d_one = json.loads(r_one.output)
606 assert len(d_all["predictions"]) >= len(d_one["predictions"])
607
608 def test_min_confidence_filters(self, repo: pathlib.Path) -> None:
609 result = runner.invoke(cli, [
610 "code", "predict", "--json", "--min-confidence", "0.99",
611 ])
612 data = json.loads(result.output)
613 for pred in data["predictions"]:
614 assert pred["score"] >= 0.99
615
616 def test_file_filter(self, repo: pathlib.Path) -> None:
617 result = runner.invoke(cli, [
618 "code", "predict", "--json", "--file", "nonexistent_file_xyz.py",
619 ])
620 data = json.loads(result.output)
621 assert data["predictions"] == []
622
623 def test_horizon_reflected_in_json(self, repo: pathlib.Path) -> None:
624 result = runner.invoke(cli, ["code", "predict", "--json", "--horizon", "5"])
625 data = json.loads(result.output)
626 assert data["horizon_commits"] == 5
627
628
629 # ---------------------------------------------------------------------------
630 # Integration — --explain
631 # ---------------------------------------------------------------------------
632
633
634 class TestPredictExplain:
635 def test_explain_missing_separator_exits_one(self, repo: pathlib.Path) -> None:
636 result = runner.invoke(cli, [
637 "code", "predict", "--explain", "billing_compute",
638 ])
639 assert result.exit_code == 1
640
641 def test_explain_unknown_address_exits_one(self, repo: pathlib.Path) -> None:
642 result = runner.invoke(cli, [
643 "code", "predict", "--explain", "billing.py::zzz_nonexistent_xyz",
644 ])
645 assert result.exit_code == 1
646
647 def test_explain_human_output_for_known_symbol(self, repo: pathlib.Path) -> None:
648 # First get a prediction to know a valid address.
649 r = runner.invoke(cli, ["code", "predict", "--json"])
650 data = json.loads(r.output)
651 if not data["predictions"]:
652 pytest.skip("no predictions")
653 addr = data["predictions"][0]["address"]
654 result = runner.invoke(cli, ["code", "predict", "--explain", addr])
655 assert result.exit_code == 0
656 assert "signal breakdown" in result.output.lower() or "score" in result.output.lower()
657
658
659 # ---------------------------------------------------------------------------
660 # Integration — --explain --json (_ExplainJson)
661 # ---------------------------------------------------------------------------
662
663
664 class TestPredictExplainJson:
665 def test_explain_json_exits_zero(self, repo: pathlib.Path) -> None:
666 r = runner.invoke(cli, ["code", "predict", "--json"])
667 data = json.loads(r.output)
668 if not data["predictions"]:
669 pytest.skip("no predictions")
670 addr = data["predictions"][0]["address"]
671 result = runner.invoke(cli, [
672 "code", "predict", "--explain", addr, "--json",
673 ])
674 assert result.exit_code == 0, result.output
675
676 def test_explain_json_valid_json(self, repo: pathlib.Path) -> None:
677 r = runner.invoke(cli, ["code", "predict", "--json"])
678 data = json.loads(r.output)
679 if not data["predictions"]:
680 pytest.skip("no predictions")
681 addr = data["predictions"][0]["address"]
682 result = runner.invoke(cli, [
683 "code", "predict", "--explain", addr, "-j",
684 ])
685 explain = json.loads(result.output.strip())
686 assert isinstance(explain, dict)
687
688 def test_explain_json_has_address(self, repo: pathlib.Path) -> None:
689 r = runner.invoke(cli, ["code", "predict", "--json"])
690 data = json.loads(r.output)
691 if not data["predictions"]:
692 pytest.skip("no predictions")
693 addr = data["predictions"][0]["address"]
694 result = runner.invoke(cli, ["code", "predict", "--explain", addr, "-j"])
695 explain = json.loads(result.output)
696 assert explain["address"] == addr
697
698 def test_explain_json_has_score(self, repo: pathlib.Path) -> None:
699 r = runner.invoke(cli, ["code", "predict", "--json"])
700 data = json.loads(r.output)
701 if not data["predictions"]:
702 pytest.skip("no predictions")
703 addr = data["predictions"][0]["address"]
704 result = runner.invoke(cli, ["code", "predict", "--explain", addr, "-j"])
705 explain = json.loads(result.output)
706 assert "score" in explain
707 assert isinstance(explain["score"], float)
708
709 def test_explain_json_has_signals(self, repo: pathlib.Path) -> None:
710 r = runner.invoke(cli, ["code", "predict", "--json"])
711 data = json.loads(r.output)
712 if not data["predictions"]:
713 pytest.skip("no predictions")
714 addr = data["predictions"][0]["address"]
715 result = runner.invoke(cli, ["code", "predict", "--explain", addr, "-j"])
716 explain = json.loads(result.output)
717 assert "signals" in explain
718 assert set(explain["signals"].keys()) == {
719 "recency", "frequency", "co_change", "sig_instability", "module_velocity"
720 }
721
722 def test_explain_json_has_reasons(self, repo: pathlib.Path) -> None:
723 r = runner.invoke(cli, ["code", "predict", "--json"])
724 data = json.loads(r.output)
725 if not data["predictions"]:
726 pytest.skip("no predictions")
727 addr = data["predictions"][0]["address"]
728 result = runner.invoke(cli, ["code", "predict", "--explain", addr, "-j"])
729 explain = json.loads(result.output)
730 assert "reasons" in explain
731 assert isinstance(explain["reasons"], list)
732
733 def test_explain_json_has_top_partners(self, repo: pathlib.Path) -> None:
734 r = runner.invoke(cli, ["code", "predict", "--json"])
735 data = json.loads(r.output)
736 if not data["predictions"]:
737 pytest.skip("no predictions")
738 addr = data["predictions"][0]["address"]
739 result = runner.invoke(cli, ["code", "predict", "--explain", addr, "-j"])
740 explain = json.loads(result.output)
741 assert "top_partners" in explain
742 assert isinstance(explain["top_partners"], list)
743
744 def test_explain_json_has_exit_code(self, repo: pathlib.Path) -> None:
745 r = runner.invoke(cli, ["code", "predict", "--json"])
746 data = json.loads(r.output)
747 if not data["predictions"]:
748 pytest.skip("no predictions")
749 addr = data["predictions"][0]["address"]
750 result = runner.invoke(cli, ["code", "predict", "--explain", addr, "-j"])
751 explain = json.loads(result.output)
752 assert "exit_code" in explain
753 assert explain["exit_code"] == 0
754
755 def test_explain_json_has_duration_ms(self, repo: pathlib.Path) -> None:
756 r = runner.invoke(cli, ["code", "predict", "--json"])
757 data = json.loads(r.output)
758 if not data["predictions"]:
759 pytest.skip("no predictions")
760 addr = data["predictions"][0]["address"]
761 result = runner.invoke(cli, ["code", "predict", "--explain", addr, "-j"])
762 explain = json.loads(result.output)
763 assert "duration_ms" in explain
764 assert explain["duration_ms"] >= 0
765
766 def test_explain_json_importable_typeddict(self) -> None:
767 from muse.cli.commands.predict import _ExplainJson
768 hints = typing.get_type_hints(_ExplainJson)
769 assert "address" in hints
770 assert "score" in hints
771 assert "signals" in hints
772 assert "reasons" in hints
773 assert "top_partners" in hints
774 assert "exit_code" in hints
775 assert "duration_ms" in hints
776
777
778 # ---------------------------------------------------------------------------
779 # Integration — security
780 # ---------------------------------------------------------------------------
781
782
783 class TestPredictSecurity:
784 def test_min_confidence_above_one_exits_one(self, repo: pathlib.Path) -> None:
785 result = runner.invoke(cli, [
786 "code", "predict", "--min-confidence", "1.5",
787 ])
788 assert result.exit_code == 1
789
790 def test_min_confidence_below_zero_exits_one(self, repo: pathlib.Path) -> None:
791 result = runner.invoke(cli, [
792 "code", "predict", "--min-confidence", "-0.1",
793 ])
794 assert result.exit_code == 1
795
796 def test_explain_without_double_colon_exits_one(self, repo: pathlib.Path) -> None:
797 result = runner.invoke(cli, [
798 "code", "predict", "--explain", "no_separator_here",
799 ])
800 assert result.exit_code == 1
801
802
803 # ---------------------------------------------------------------------------
804 # TypedDict coverage
805 # ---------------------------------------------------------------------------
806
807
808 class TestTypedDicts:
809 def test_predict_json_typeddict_importable(self) -> None:
810 from muse.cli.commands.predict import _PredictJson
811 hints = typing.get_type_hints(_PredictJson)
812 assert "predictions" in hints
813 assert "exit_code" in hints
814 assert "duration_ms" in hints
815
816 def test_explain_json_typeddict_importable(self) -> None:
817 from muse.cli.commands.predict import _ExplainJson
818 assert _ExplainJson is not None
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 143 days ago