gabriel / muse public
semantic_test_coverage.py python
843 lines 32.2 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 125 days ago
1 """muse code semantic-test-coverage — static symbol-level test coverage.
2
3 Traditional coverage tools measure *which lines execute* during a test run.
4 They require an instrumented test suite, a working interpreter, and real I/O.
5
6 This command answers a different question:
7
8 **Which symbols are exercised by which test functions — without running any
9 tests?**
10
11 Because Muse tracks the complete symbol graph of every snapshot, we can
12 perform static call-graph analysis to determine — at the symbol level, across
13 every production file — which test functions exercise which symbols. No test
14 runner. No instrumentation. No I/O.
15
16 Coverage tiers
17 --------------
18 ``--depth 1`` (default, "direct")
19 A production symbol is covered if any test function body directly calls it
20 by bare name (e.g. ``compute_total(...)`` or ``obj.compute_total(...)``).
21 This is a conservative but high-precision signal.
22
23 ``--depth N / --transitive`` (N > 1)
24 Extends direct coverage by following the production call graph up to N−1
25 additional hops. If ``test_process_order`` calls ``process_order`` and
26 ``process_order`` calls ``Invoice.compute_total``, then
27 ``Invoice.compute_total`` is covered at depth 2.
28
29 Scope
30 -----
31 Production symbols
32 Every non-import symbol in a non-test file. Classes, functions, methods,
33 async functions, and async methods are all included.
34
35 Test functions
36 Every ``test_``-prefixed function in a test file, including methods inside
37 ``Test*`` classes. All top-level functions in ``conftest.py`` (fixtures)
38 are included because they often call production code directly.
39
40 Usage::
41
42 muse code semantic-test-coverage
43 muse code semantic-test-coverage --file billing.py
44 muse code semantic-test-coverage --kind function
45 muse code semantic-test-coverage --uncovered-only
46 muse code semantic-test-coverage --show-tests
47 muse code semantic-test-coverage --transitive --depth 2
48 muse code semantic-test-coverage --min-coverage 80
49 muse code semantic-test-coverage --json
50
51 Output::
52
53 Semantic test coverage — HEAD (378 symbols · 47 test functions)
54
55 billing.py
56 [████████████████░░░░] 80.0% (4/5)
57 ✅ compute_total meth
58 ✅ apply_discount meth
59 ✅ process_order func
60 ✅ Invoice clas
61 ❌ generate_pdf meth
62
63 ────────────────────────────────────────────────────────────────
64 TOTAL: 285/378 symbols covered (75.4%)
65
66 JSON output (``--json``)::
67
68 {
69 "ref": "HEAD",
70 "snapshot_id": "a3f2c9e1ab2c",
71 "depth": 1,
72 "transitive": false,
73 "filters": { "file": null, "kind": null, "min_coverage": null,
74 "uncovered_only": false },
75 "summary": {
76 "total_symbols": 378,
77 "covered_symbols": 285,
78 "uncovered_symbols": 93,
79 "coverage_pct": 75.4,
80 "total_test_functions": 47,
81 "total_production_files": 28
82 },
83 "files": [
84 {
85 "file": "billing.py",
86 "total_symbols": 5,
87 "covered_symbols": 4,
88 "uncovered_symbols": 1,
89 "coverage_pct": 80.0,
90 "symbols": [
91 {
92 "address": "billing.py::Invoice.compute_total",
93 "name": "compute_total",
94 "kind": "method",
95 "covered": true,
96 "test_functions": ["tests/test_billing.py::test_compute_total_basic"]
97 }
98 ]
99 }
100 ]
101 }
102
103 ``--min-coverage PCT`` — CI integration::
104
105 # Exit 1 if any file falls below 80% semantic coverage
106 muse code semantic-test-coverage --min-coverage 80
107
108 Performance
109 -----------
110 A single pass loads all Python blobs from the committed object store into
111 memory. AST parsing and walking are done once per file. For transitive
112 coverage, the production call graph is built from the already-loaded blobs
113 without a second store read. Typical runtimes:
114
115 200 files / 1 000 symbols / 100 test functions → < 2 s
116 1 000 files / 10 000 symbols / 500 test functions → < 10 s
117
118 Accuracy note
119 -------------
120 This is a *static* analysis. Dynamic dispatch, conditional imports, and
121 runtime code generation cannot be captured without execution. The analysis
122 may report:
123
124 * False negatives: dynamically-constructed call sites (``getattr(obj, name)()``)
125 are not detected.
126 * False positives: names like ``save`` match any production symbol named
127 ``save``, regardless of the actual class/instance.
128
129 Use the results as a coverage *signal*, not a proof.
130 """
131
132 import argparse
133 import ast
134 import json
135 import logging
136 import pathlib
137 import sys
138 from collections import defaultdict
139 from typing import TypedDict
140
141 from muse.core.envelope import EnvelopeJson, make_envelope
142 from muse.core.errors import ExitCode
143 from muse.core.repo import read_repo_id, require_repo
144 from muse.core.store import (
145 Manifest,
146 get_commit_snapshot_manifest,
147 read_current_branch,
148 resolve_commit_ref,
149 )
150 from muse.core.timing import start_timer
151 from muse.plugins.code._callgraph import ForwardGraph, call_name, find_func_node
152 from muse.plugins.code._query import is_test_file, symbols_for_snapshot
153 from muse.plugins.code.ast_parser import SymbolRecord, parse_symbols
154 from muse.core.validation import clamp_int, MAX_AST_BYTES, sanitize_display
155
156 logger = logging.getLogger(__name__)
157
158 type BlobMap = dict[str, bytes] # file_path → raw blob bytes
159 type TestRefs = dict[str, set[str]] # test_addr → set of callee names
160 type CoverageMap = dict[str, set[str]] # prod_addr → set of test_addrs
161 type NameToAddrs = dict[str, list[str]] # bare_name → list of prod_addrs
162 type _SymbolTree = dict[str, SymbolRecord]
163 type SymbolTreeMap = dict[str, _SymbolTree] # file_path → symbol tree
164 type FlatSymbolMap = dict[str, SymbolRecord] # address → symbol record
165
166 # ── Constants ──────────────────────────────────────────────────────────────────
167
168 _DEFAULT_TOP = 0 # 0 = unlimited
169 _DEFAULT_DEPTH = 1
170 _MAX_DEPTH = 10
171
172 _PY_SUFFIXES: frozenset[str] = frozenset({".py", ".pyi"})
173
174 _IMPORT_KIND = "import"
175
176 # Symbol kinds surfaced by this command.
177 _TRACKED_KINDS: frozenset[str] = frozenset(
178 {"function", "async_function", "method", "async_method", "class"}
179 )
180
181 # ── TypedDicts ─────────────────────────────────────────────────────────────────
182
183 class _SymbolCov(TypedDict):
184 """Coverage record for a single production symbol."""
185
186 address: str
187 name: str
188 kind: str
189 covered: bool
190 test_functions: list[str]
191
192 class _FileCov(TypedDict):
193 """Aggregated coverage record for one production file."""
194
195 file: str
196 total_symbols: int
197 covered_symbols: int
198 uncovered_symbols: int
199 coverage_pct: float
200 symbols: list[_SymbolCov]
201
202 class _FilterSpec(TypedDict):
203 """Filters applied to this analysis run."""
204
205 file: str | None
206 kind: str | None
207 min_coverage: int | None
208 uncovered_only: bool
209
210 class _SummarySpec(TypedDict):
211 """Aggregate statistics across all production files."""
212
213 total_symbols: int
214 covered_symbols: int
215 uncovered_symbols: int
216 coverage_pct: float
217 total_test_functions: int
218 total_production_files: int
219
220 class _JsonOut(EnvelopeJson):
221 """Top-level JSON output structure for ``muse code semantic-test-coverage --json``."""
222
223 ref: str
224 snapshot_id: str
225 depth: int
226 transitive: bool
227 filters: _FilterSpec
228 summary: _SummarySpec
229 files: list[_FileCov]
230
231 def _is_conftest(file_path: str) -> bool:
232 """Return True if *file_path* is a conftest module."""
233 return pathlib.PurePosixPath(file_path).name == "conftest.py"
234
235 # ── AST helpers ────────────────────────────────────────────────────────────────
236
237 def _collect_test_funcs(
238 stmts: list[ast.stmt],
239 prefix: str,
240 is_conftest: bool,
241 ) -> list[tuple[str, ast.FunctionDef | ast.AsyncFunctionDef]]:
242 """Recursively collect test function nodes from *stmts*.
243
244 Handles top-level test functions and methods inside ``Test*`` classes.
245 ``conftest.py`` includes all functions (fixture bodies also call production
246 code).
247
248 Args:
249 stmts: Statement list from a module or class body.
250 prefix: Dotted qualification prefix accumulated from enclosing classes.
251 is_conftest: True when the file is ``conftest.py``; includes all functions.
252
253 Returns:
254 List of ``(qualified_name, node)`` pairs for each test entry point found.
255 """
256 results: list[tuple[str, ast.FunctionDef | ast.AsyncFunctionDef]] = []
257 for stmt in stmts:
258 if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)):
259 qname = f"{prefix}.{stmt.name}" if prefix else stmt.name
260 if is_conftest or stmt.name.startswith("test_"):
261 results.append((qname, stmt))
262 elif isinstance(stmt, ast.ClassDef):
263 new_prefix = f"{prefix}.{stmt.name}" if prefix else stmt.name
264 results.extend(_collect_test_funcs(stmt.body, new_prefix, is_conftest))
265 return results
266
267 def _scan_calls(func_node: ast.FunctionDef | ast.AsyncFunctionDef) -> set[str]:
268 """Return the set of bare callee names called inside *func_node*.
269
270 Only ``ast.Call`` nodes are examined. Bare name references without a call
271 are excluded (they do not indicate the symbol is exercised).
272
273 Args:
274 func_node: The function or async-function AST node to scan.
275
276 Returns:
277 Set of bare callee names (e.g. ``{"compute_total", "Invoice"}``)
278 """
279 names: set[str] = set()
280 for node in ast.walk(func_node):
281 if isinstance(node, ast.Call):
282 name = call_name(node.func)
283 if name:
284 names.add(name)
285 return names
286
287 # ── Data collection ────────────────────────────────────────────────────────────
288
289 def _load_py_blobs(
290 root: pathlib.Path,
291 manifest: Manifest,
292 ) -> BlobMap:
293 """Load every Python/stub blob from *manifest* into memory.
294
295 A single bulk read avoids redundant object-store I/O for subsequent AST
296 parsing and call-graph construction phases.
297
298 Args:
299 root: Repository root (object store location).
300 manifest: Snapshot manifest from ``get_commit_snapshot_manifest``.
301
302 Returns:
303 Mapping ``{file_path: raw_bytes}`` for every Python file in the snapshot.
304 """
305 from muse.core.object_store import read_object as _read_obj
306
307 blobs: BlobMap = {}
308 for file_path, obj_id in manifest.items():
309 if pathlib.PurePosixPath(file_path).suffix.lower() not in _PY_SUFFIXES:
310 continue
311 raw = _read_obj(root, obj_id)
312 if raw is not None:
313 blobs[file_path] = raw
314 return blobs
315
316 def _build_test_refs(blobs: BlobMap) -> TestRefs:
317 """Scan test files and return ``{test_addr: set[bare_callee_name]}``.
318
319 Args:
320 blobs: Pre-loaded mapping from ``_load_py_blobs``.
321
322 Returns:
323 A mapping from each test-function address to the set of bare callee
324 names called anywhere inside that function body.
325 """
326 refs: TestRefs = {}
327 for file_path, raw in blobs.items():
328 if not is_test_file(file_path):
329 continue
330 try:
331 if len(raw) > MAX_AST_BYTES:
332 return {}
333 tree = ast.parse(raw)
334 except SyntaxError:
335 logger.warning("SyntaxError in %s — test file skipped", file_path)
336 continue
337 is_conf = _is_conftest(file_path)
338 for qname, func_node in _collect_test_funcs(tree.body, "", is_conf):
339 addr = f"{file_path}::{qname}"
340 refs[addr] = _scan_calls(func_node)
341 return refs
342
343 def _build_prod_forward_graph(
344 blobs: BlobMap,
345 prod_files: set[str],
346 ) -> ForwardGraph:
347 """Build a forward call graph for production files using pre-loaded blobs.
348
349 Reuses ``blobs`` to avoid a second round-trip through the object store.
350 Only callable symbols (functions, methods) are included in the graph.
351
352 Args:
353 blobs: Pre-loaded Python blobs from ``_load_py_blobs``.
354 prod_files: Set of production file paths to include in the graph.
355
356 Returns:
357 ``{caller_address: frozenset[callee_bare_name]}``
358 """
359 graph: ForwardGraph = {}
360 for file_path, raw in blobs.items():
361 if file_path not in prod_files:
362 continue
363 try:
364 if len(raw) > MAX_AST_BYTES:
365 return {}
366 tree = ast.parse(raw)
367 except SyntaxError:
368 continue
369 sym_tree = parse_symbols(raw, file_path)
370 for addr, rec in sym_tree.items():
371 if rec["kind"] not in {"function", "async_function", "method", "async_method"}:
372 continue
373 func_node = find_func_node(tree.body, rec["qualified_name"].split("."))
374 if func_node is None:
375 continue
376 callees: set[str] = set()
377 for node in ast.walk(func_node):
378 if isinstance(node, ast.Call):
379 n = call_name(node.func)
380 if n:
381 callees.add(n)
382 graph[addr] = frozenset(callees)
383 return graph
384
385 def _compute_coverage(
386 test_refs: TestRefs,
387 name_to_prod_addrs: NameToAddrs,
388 forward_graph: ForwardGraph | None,
389 depth: int,
390 ) -> CoverageMap:
391 """Compute ``{prod_addr: {test_addr, ...}}`` coverage mapping.
392
393 Phase 1 — Direct coverage:
394 For each test function, find every bare callee name that matches a
395 production symbol address via *name_to_prod_addrs*.
396
397 Phase 2 — Transitive expansion (only when ``forward_graph`` is provided
398 and ``depth > 1``):
399 BFS from each directly-covered production symbol, following calls in
400 *forward_graph*. Each newly-reached production symbol is added to the
401 test function's coverage set. The BFS is capped at ``depth - 1``
402 additional hops.
403
404 Args:
405 test_refs: ``{test_addr: set[bare_callee_name]}``.
406 name_to_prod_addrs: Reverse index: ``{bare_name: [prod_addr, ...]}``.
407 forward_graph: Production call graph; ``None`` for direct-only mode.
408 depth: Total coverage depth (1 = direct only).
409
410 Returns:
411 ``{prod_addr: {test_addr, ...}}``
412 """
413 # Phase 1: direct coverage
414 coverage: CoverageMap = defaultdict(set)
415 test_to_direct: CoverageMap = defaultdict(set)
416
417 for test_addr, bare_names in test_refs.items():
418 for bare in bare_names:
419 for prod_addr in name_to_prod_addrs.get(bare, []):
420 coverage[prod_addr].add(test_addr)
421 test_to_direct[test_addr].add(prod_addr)
422
423 # Phase 2: transitive expansion
424 if forward_graph is not None and depth > 1:
425 for test_addr, direct_prods in test_to_direct.items():
426 frontier: set[str] = set(direct_prods)
427 visited: set[str] = set(direct_prods)
428 for _ in range(depth - 1):
429 next_frontier: set[str] = set()
430 for prod_addr in frontier:
431 for callee_bare in forward_graph.get(prod_addr, frozenset()):
432 for callee_addr in name_to_prod_addrs.get(callee_bare, []):
433 if callee_addr not in visited:
434 visited.add(callee_addr)
435 next_frontier.add(callee_addr)
436 coverage[callee_addr].add(test_addr)
437 if not next_frontier:
438 break
439 frontier = next_frontier
440
441 return dict(coverage)
442
443 # ── Output builders ────────────────────────────────────────────────────────────
444
445 def _build_file_coverage(
446 prod_trees: SymbolTreeMap,
447 coverage: CoverageMap,
448 file_filter: str | None,
449 kind_filter: str | None,
450 ) -> list[_FileCov]:
451 """Build structured ``_FileCov`` records for every production file.
452
453 Returns all symbols per file (uncovered-only filtering is applied at
454 display/JSON time so summary statistics always reflect the full picture).
455
456 Args:
457 prod_trees: ``{file_path: {addr: SymbolRecord}}`` for production files.
458 coverage: ``{prod_addr: {test_addr, ...}}`` from ``_compute_coverage``.
459 file_filter: Optional path suffix; only files containing this string are included.
460 kind_filter: Optional symbol kind; only symbols of this kind are included.
461
462 Returns:
463 List of ``_FileCov`` records sorted by file path.
464 """
465 result: list[_FileCov] = []
466 for file_path in sorted(prod_trees):
467 if file_filter and file_filter not in file_path:
468 continue
469 sym_tree = prod_trees[file_path]
470 syms: list[_SymbolCov] = []
471 for addr in sorted(sym_tree):
472 rec = sym_tree[addr]
473 if rec["kind"] == _IMPORT_KIND:
474 continue
475 if rec["kind"] not in _TRACKED_KINDS:
476 continue
477 if kind_filter and rec["kind"] != kind_filter:
478 continue
479 test_fns = sorted(coverage.get(addr, set()))
480 syms.append(
481 _SymbolCov(
482 address=addr,
483 name=rec["name"],
484 kind=rec["kind"],
485 covered=bool(test_fns),
486 test_functions=test_fns,
487 )
488 )
489 if not syms:
490 continue
491 covered_count = sum(1 for s in syms if s["covered"])
492 result.append(
493 _FileCov(
494 file=file_path,
495 total_symbols=len(syms),
496 covered_symbols=covered_count,
497 uncovered_symbols=len(syms) - covered_count,
498 coverage_pct=round(100.0 * covered_count / len(syms), 1),
499 symbols=syms,
500 )
501 )
502 return result
503
504 # ── Formatting ─────────────────────────────────────────────────────────────────
505
506 _BAR_WIDTH = 20
507 _KindAbbrevMap = dict[str, str]
508 _KIND_ABBREV: _KindAbbrevMap = {
509 "function": "func",
510 "async_function": "afn ",
511 "method": "meth",
512 "async_method": "amth",
513 "class": "cls ",
514 }
515
516 def _bar(pct: float) -> str:
517 """Return a 20-char Unicode block bar representing *pct* (0–100)."""
518 filled = round(pct / (100 / _BAR_WIDTH))
519 return f"{'█' * filled}{'░' * (_BAR_WIDTH - filled)}"
520
521 def _print_table(
522 files: list[_FileCov],
523 uncovered_only: bool,
524 show_tests: bool,
525 min_coverage: int,
526 ref: str,
527 total_test_fns: int,
528 ) -> bool:
529 """Print human-readable coverage table to stdout.
530
531 Args:
532 files: Structured file coverage from ``_build_file_coverage``.
533 uncovered_only: When True, only uncovered symbols are printed per file.
534 show_tests: When True, list covering test functions under each symbol.
535 min_coverage: Coverage threshold for ⚠️ flag and exit-1 signalling.
536 ref: Display string for the analysed ref (e.g. ``"HEAD"``).
537 total_test_fns: Total number of test functions found in the snapshot.
538
539 Returns:
540 True if any file violates *min_coverage*; False otherwise.
541 """
542 total_syms = sum(f["total_symbols"] for f in files)
543 total_covered = sum(f["covered_symbols"] for f in files)
544 total_pct = round(100.0 * total_covered / total_syms, 1) if total_syms else 0.0
545
546 print(
547 f"Semantic test coverage — {ref}"
548 f" ({total_syms} symbols · {total_test_fns} test functions)\n"
549 )
550
551 violation = False
552 for fc in files:
553 pct = fc["coverage_pct"]
554 bar = _bar(pct)
555 below = pct < min_coverage and min_coverage > 0
556 if below:
557 violation = True
558 flag = " ⚠️" if below else ""
559 print(f" {fc['file']}")
560 print(
561 f" [{bar}] {pct:5.1f}%"
562 f" ({fc['covered_symbols']}/{fc['total_symbols']}){flag}"
563 )
564
565 for sym in fc["symbols"]:
566 if uncovered_only and sym["covered"]:
567 continue
568 icon = "✅" if sym["covered"] else "❌"
569 kind_short = _KIND_ABBREV.get(sym["kind"], sym["kind"][:4])
570 print(f" {icon} {sanitize_display(sym['name']):<42} {kind_short}")
571 if show_tests and sym["test_functions"]:
572 for tf in sym["test_functions"]:
573 print(f" ← {tf}")
574 print()
575
576 sep = "─" * 64
577 print(f" {sep}")
578 print(f" TOTAL: {total_covered}/{total_syms} symbols covered ({total_pct}%)\n")
579 if violation:
580 print(
581 f" ⚠️ One or more files are below the {min_coverage}%"
582 " minimum coverage threshold."
583 )
584 return violation
585
586 # ── Entry point ────────────────────────────────────────────────────────────────
587
588 def run(args: argparse.Namespace) -> None:
589 """Entry point for ``muse code semantic-test-coverage``.
590
591 Validates arguments, loads the HEAD snapshot, runs the static coverage
592 analysis, and prints results (or emits JSON). Exits 1 when
593 ``--min-coverage`` is set and any file falls below the threshold.
594
595 Pass ``--json`` (or ``-j``) for a stable, machine-readable result.
596
597 Agent quickstart::
598
599 muse code semantic-test-coverage --json
600 muse code semantic-test-coverage --file billing.py --json
601 muse code semantic-test-coverage --uncovered-only --json
602 muse code semantic-test-coverage --min-coverage 80 --json
603
604 JSON fields::
605
606 ref str Analysed ref (always "HEAD")
607 snapshot_id str Short commit ID of the analysed snapshot
608 depth int Call-graph depth used for the analysis
609 transitive bool True when transitive coverage was enabled
610 filters dict Applied filters: file, kind, min_coverage, uncovered_only
611 summary dict total_symbols, covered_symbols, uncovered_symbols, coverage_pct, total_test_functions, total_production_files
612 files list Per-file coverage records with per-symbol detail
613
614 Exit codes::
615
616 0 Success (or min-coverage gate passed).
617 1 Min-coverage threshold violated.
618 1 User error (bad arguments, no HEAD commit).
619 3 Internal error.
620 """
621 elapsed = start_timer()
622 root = require_repo()
623
624 # ── Argument validation ────────────────────────────────────────────────────
625 depth: int = clamp_int(args.depth, 1, 50, 'depth')
626 if depth < 1 or depth > _MAX_DEPTH:
627 print(f"❌ --depth must be between 1 and {_MAX_DEPTH}.", file=sys.stderr)
628 raise SystemExit(ExitCode.USER_ERROR)
629
630 min_coverage: int = clamp_int(args.min_coverage, 0, 100, 'min_coverage')
631 if not (0 <= min_coverage <= 100):
632 print("❌ --min-coverage must be between 0 and 100.", file=sys.stderr)
633 raise SystemExit(ExitCode.USER_ERROR)
634
635 transitive: bool = args.transitive or depth > 1
636 effective_depth: int = depth if transitive else 1
637
638 kind_filter: str | None = args.kind or None
639 file_filter: str | None = args.file or None
640
641 # ── Resolve HEAD snapshot ──────────────────────────────────────────────────
642 repo_id = read_repo_id(root)
643 branch = read_current_branch(root)
644
645 head = resolve_commit_ref(root, repo_id, branch, None)
646 if head is None:
647 print("❌ HEAD commit not found — is this an empty repository?", file=sys.stderr)
648 raise SystemExit(ExitCode.USER_ERROR)
649
650 manifest: Manifest = get_commit_snapshot_manifest(root, head.commit_id) or {}
651
652 # ── Single bulk read of all Python blobs ───────────────────────────────────
653 blobs = _load_py_blobs(root, manifest)
654
655 # ── Symbol extraction ──────────────────────────────────────────────────────
656 all_trees = symbols_for_snapshot(root, manifest)
657
658 # Partition: test files vs. production files.
659 prod_trees: SymbolTreeMap = {}
660 for file_path, sym_tree in all_trees.items():
661 if not is_test_file(file_path):
662 prod_trees[file_path] = sym_tree
663
664 # Flat index of all production symbols, excluding imports.
665 all_prod: FlatSymbolMap = {}
666 for sym_tree in prod_trees.values():
667 for addr, rec in sym_tree.items():
668 if rec["kind"] not in (_IMPORT_KIND,) and rec["kind"] in _TRACKED_KINDS:
669 all_prod[addr] = rec
670
671 # Reverse index: bare_name → [prod_addr, ...]
672 name_to_addrs: NameToAddrs = defaultdict(list)
673 for addr, rec in all_prod.items():
674 name_to_addrs[rec["name"]].append(addr)
675
676 # ── Test-function scanning ─────────────────────────────────────────────────
677 test_refs = _build_test_refs(blobs)
678 total_test_fns = len(test_refs)
679
680 # ── Optional: build production call graph for transitive coverage ──────────
681 forward_graph: ForwardGraph | None = None
682 if transitive:
683 forward_graph = _build_prod_forward_graph(blobs, set(prod_trees))
684
685 # ── Coverage computation ───────────────────────────────────────────────────
686 coverage = _compute_coverage(
687 test_refs, dict(name_to_addrs), forward_graph, effective_depth
688 )
689
690 # ── Structured output ──────────────────────────────────────────────────────
691 file_coverage = _build_file_coverage(prod_trees, coverage, file_filter, kind_filter)
692
693 # ── JSON output ────────────────────────────────────────────────────────────
694 if args.json_out:
695 total_syms = sum(f["total_symbols"] for f in file_coverage)
696 total_covered = sum(f["covered_symbols"] for f in file_coverage)
697 pct = round(100.0 * total_covered / total_syms, 1) if total_syms else 0.0
698
699 # Apply uncovered_only filter to symbol lists in JSON.
700 if args.uncovered_only:
701 filtered_files: list[_FileCov] = []
702 for fc in file_coverage:
703 uncov = [s for s in fc["symbols"] if not s["covered"]]
704 if uncov:
705 filtered_files.append(
706 _FileCov(
707 file=fc["file"],
708 total_symbols=fc["total_symbols"],
709 covered_symbols=fc["covered_symbols"],
710 uncovered_symbols=fc["uncovered_symbols"],
711 coverage_pct=fc["coverage_pct"],
712 symbols=uncov,
713 )
714 )
715 file_coverage = filtered_files
716
717 out = _JsonOut(
718 **make_envelope(elapsed),
719 ref="HEAD",
720 snapshot_id=head.commit_id,
721 depth=effective_depth,
722 transitive=transitive,
723 filters=_FilterSpec(
724 file=file_filter,
725 kind=kind_filter,
726 min_coverage=min_coverage if min_coverage > 0 else None,
727 uncovered_only=args.uncovered_only,
728 ),
729 summary=_SummarySpec(
730 total_symbols=total_syms,
731 covered_symbols=total_covered,
732 uncovered_symbols=total_syms - total_covered,
733 coverage_pct=pct,
734 total_test_functions=total_test_fns,
735 total_production_files=len(file_coverage),
736 ),
737 files=file_coverage,
738 )
739 print(json.dumps(out))
740 return
741
742 # ── Human-readable output ──────────────────────────────────────────────────
743 violation = _print_table(
744 file_coverage,
745 uncovered_only=args.uncovered_only,
746 show_tests=args.show_tests,
747 min_coverage=min_coverage,
748 ref="HEAD",
749 total_test_fns=total_test_fns,
750 )
751 if violation:
752 sys.exit(1)
753
754 # ── CLI registration ───────────────────────────────────────────────────────────
755
756 def register(
757 sub: argparse._SubParsersAction[argparse.ArgumentParser],
758 ) -> None:
759 """Register ``semantic-test-coverage`` under the ``code`` subcommand group.
760
761 Arguments registered
762 --------------------
763 --file SUFFIX Scope to production files whose path contains SUFFIX.
764 --kind KIND Filter symbols by kind (function, method, class, …).
765 --transitive Expand coverage through the production call graph.
766 --depth D Transitive call-graph depth (default: 1). Values > 1 imply --transitive.
767 --uncovered-only Show/emit only symbols with no test coverage.
768 --show-tests Under each covered symbol, list covering test functions.
769 --min-coverage PCT Exit 1 if any production file is below PCT% semantic coverage.
770 --json / -j Emit JSON instead of human-readable text.
771
772 Args:
773 sub: The subparser action from the ``code`` command group.
774 """
775 p = sub.add_parser(
776 "semantic-test-coverage",
777 help=(
778 "Static symbol-level test coverage — which symbols are exercised"
779 " by which tests, without running the test suite."
780 ),
781 description=__doc__,
782 formatter_class=argparse.RawDescriptionHelpFormatter,
783 )
784 p.add_argument(
785 "--file",
786 metavar="SUFFIX",
787 help=(
788 "Scope analysis to production files whose path contains SUFFIX"
789 " (e.g. --file billing.py)."
790 ),
791 )
792 p.add_argument(
793 "--kind",
794 metavar="KIND",
795 choices=["function", "async_function", "method", "async_method", "class"],
796 help="Filter symbols by kind.",
797 )
798 p.add_argument(
799 "--transitive",
800 action="store_true",
801 help=(
802 "Expand coverage through the production call graph."
803 " Combines with --depth (default: 2 when this flag is set)."
804 ),
805 )
806 p.add_argument(
807 "--depth",
808 type=int,
809 default=_DEFAULT_DEPTH,
810 metavar="D",
811 help=(
812 f"Transitive call-graph depth (default: {_DEFAULT_DEPTH})."
813 " Values > 1 imply --transitive."
814 f" Maximum: {_MAX_DEPTH}."
815 ),
816 )
817 p.add_argument(
818 "--uncovered-only",
819 action="store_true",
820 help="Show (or emit in JSON) only symbols with no test coverage.",
821 )
822 p.add_argument(
823 "--show-tests",
824 action="store_true",
825 help="Under each covered symbol, list the test functions that exercise it.",
826 )
827 p.add_argument(
828 "--min-coverage",
829 type=int,
830 default=0,
831 metavar="PCT",
832 help=(
833 "Exit 1 if any production file is below PCT%% semantic coverage."
834 " Useful for CI gates."
835 ),
836 )
837 p.add_argument(
838 "--json", "-j",
839 action="store_true",
840 dest="json_out",
841 help="Emit JSON instead of human-readable text.",
842 )
843 p.set_defaults(func=run)
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 125 days ago