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