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