"""muse code semantic-test-coverage — static symbol-level test coverage. Traditional coverage tools measure *which lines execute* during a test run. They require an instrumented test suite, a working interpreter, and real I/O. This command answers a different question: **Which symbols are exercised by which test functions — without running any tests?** Because Muse tracks the complete symbol graph of every snapshot, we can perform static call-graph analysis to determine — at the symbol level, across every production file — which test functions exercise which symbols. No test runner. No instrumentation. No I/O. Coverage tiers -------------- ``--depth 1`` (default, "direct") A production symbol is covered if any test function body directly calls it by bare name (e.g. ``compute_total(...)`` or ``obj.compute_total(...)``). This is a conservative but high-precision signal. ``--depth N / --transitive`` (N > 1) Extends direct coverage by following the production call graph up to N−1 additional hops. If ``test_process_order`` calls ``process_order`` and ``process_order`` calls ``Invoice.compute_total``, then ``Invoice.compute_total`` is covered at depth 2. Scope ----- Production symbols Every non-import symbol in a non-test file. Classes, functions, methods, async functions, and async methods are all included. Test functions Every ``test_``-prefixed function in a test file, including methods inside ``Test*`` classes. All top-level functions in ``conftest.py`` (fixtures) are included because they often call production code directly. Usage:: muse code semantic-test-coverage muse code semantic-test-coverage --file billing.py muse code semantic-test-coverage --kind function muse code semantic-test-coverage --uncovered-only muse code semantic-test-coverage --show-tests muse code semantic-test-coverage --transitive --depth 2 muse code semantic-test-coverage --min-coverage 80 muse code semantic-test-coverage --json Output:: Semantic test coverage — HEAD (378 symbols · 47 test functions) billing.py [████████████████░░░░] 80.0% (4/5) ✅ compute_total meth ✅ apply_discount meth ✅ process_order func ✅ Invoice clas ❌ generate_pdf meth ──────────────────────────────────────────────────────────────── TOTAL: 285/378 symbols covered (75.4%) JSON output (``--json``):: { "ref": "HEAD", "snapshot_id": "a3f2c9e1ab2c", "depth": 1, "transitive": false, "filters": { "file": null, "kind": null, "min_coverage": null, "uncovered_only": false }, "summary": { "total_symbols": 378, "covered_symbols": 285, "uncovered_symbols": 93, "coverage_pct": 75.4, "total_test_functions": 47, "total_production_files": 28 }, "files": [ { "file": "billing.py", "total_symbols": 5, "covered_symbols": 4, "uncovered_symbols": 1, "coverage_pct": 80.0, "symbols": [ { "address": "billing.py::Invoice.compute_total", "name": "compute_total", "kind": "method", "covered": true, "test_functions": ["tests/test_billing.py::test_compute_total_basic"] } ] } ] } ``--min-coverage PCT`` — CI integration:: # Exit 1 if any file falls below 80% semantic coverage muse code semantic-test-coverage --min-coverage 80 Performance ----------- A single pass loads all Python blobs from the committed object store into memory. AST parsing and walking are done once per file. For transitive coverage, the production call graph is built from the already-loaded blobs without a second store read. Typical runtimes: 200 files / 1 000 symbols / 100 test functions → < 2 s 1 000 files / 10 000 symbols / 500 test functions → < 10 s Accuracy note ------------- This is a *static* analysis. Dynamic dispatch, conditional imports, and runtime code generation cannot be captured without execution. The analysis may report: * False negatives: dynamically-constructed call sites (``getattr(obj, name)()``) are not detected. * False positives: names like ``save`` match any production symbol named ``save``, regardless of the actual class/instance. Use the results as a coverage *signal*, not a proof. """ from __future__ import annotations import argparse import ast import json import logging import pathlib import re import sys from collections import defaultdict from typing import TypedDict from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( Manifest, get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref, ) from muse.plugins.code._callgraph import ForwardGraph, call_name, find_func_node from muse.plugins.code._query import symbols_for_snapshot from muse.plugins.code.ast_parser import SymbolRecord, parse_symbols from muse.core.validation import clamp_int, MAX_AST_BYTES, sanitize_display logger = logging.getLogger(__name__) type BlobMap = dict[str, bytes] # file_path → raw blob bytes type TestRefs = dict[str, set[str]] # test_addr → set of callee names type CoverageMap = dict[str, set[str]] # prod_addr → set of test_addrs type NameToAddrs = dict[str, list[str]] # bare_name → list of prod_addrs type SymbolTreeMap = dict[str, dict[str, SymbolRecord]] # file_path → symbol tree type FlatSymbolMap = dict[str, SymbolRecord] # address → symbol record # ── Constants ────────────────────────────────────────────────────────────────── _DEFAULT_TOP = 0 # 0 = unlimited _DEFAULT_DEPTH = 1 _MAX_DEPTH = 10 _PY_SUFFIXES: frozenset[str] = frozenset({".py", ".pyi"}) # Test file heuristics — same patterns used across the code-domain commands. _TEST_PATTERNS: tuple[re.Pattern[str], ...] = ( re.compile(r"(^|/)test_"), re.compile(r"_test\.py$"), re.compile(r"(^|/)tests/"), re.compile(r"(^|/)spec/"), re.compile(r"(^|/)conftest\.py$"), ) _IMPORT_KIND = "import" # Symbol kinds surfaced by this command. _TRACKED_KINDS: frozenset[str] = frozenset( {"function", "async_function", "method", "async_method", "class"} ) # ── TypedDicts ───────────────────────────────────────────────────────────────── class _SymbolCov(TypedDict): """Coverage record for a single production symbol.""" address: str name: str kind: str covered: bool test_functions: list[str] class _FileCov(TypedDict): """Aggregated coverage record for one production file.""" file: str total_symbols: int covered_symbols: int uncovered_symbols: int coverage_pct: float symbols: list[_SymbolCov] class _FilterSpec(TypedDict): """Filters applied to this analysis run.""" file: str | None kind: str | None min_coverage: int | None uncovered_only: bool class _SummarySpec(TypedDict): """Aggregate statistics across all production files.""" total_symbols: int covered_symbols: int uncovered_symbols: int coverage_pct: float total_test_functions: int total_production_files: int class _JsonOut(TypedDict): """Top-level JSON output structure.""" ref: str snapshot_id: str depth: int transitive: bool filters: _FilterSpec summary: _SummarySpec files: list[_FileCov] # ── Test-file detection ──────────────────────────────────────────────────────── def _is_test_file(file_path: str) -> bool: """Return True if *file_path* looks like a test or conftest file.""" return any(p.search(file_path) for p in _TEST_PATTERNS) def _is_conftest(file_path: str) -> bool: """Return True if *file_path* is a conftest module.""" return pathlib.PurePosixPath(file_path).name == "conftest.py" # ── AST helpers ──────────────────────────────────────────────────────────────── def _collect_test_funcs( stmts: list[ast.stmt], prefix: str, is_conftest: bool, ) -> list[tuple[str, ast.FunctionDef | ast.AsyncFunctionDef]]: """Recursively collect test function nodes from *stmts*. Handles top-level test functions and methods inside ``Test*`` classes. ``conftest.py`` includes all functions (fixture bodies also call production code). Args: stmts: Statement list from a module or class body. prefix: Dotted qualification prefix accumulated from enclosing classes. is_conftest: True when the file is ``conftest.py``; includes all functions. Returns: List of ``(qualified_name, node)`` pairs for each test entry point found. """ results: list[tuple[str, ast.FunctionDef | ast.AsyncFunctionDef]] = [] for stmt in stmts: if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)): qname = f"{prefix}.{stmt.name}" if prefix else stmt.name if is_conftest or stmt.name.startswith("test_"): results.append((qname, stmt)) elif isinstance(stmt, ast.ClassDef): new_prefix = f"{prefix}.{stmt.name}" if prefix else stmt.name results.extend(_collect_test_funcs(stmt.body, new_prefix, is_conftest)) return results def _scan_calls(func_node: ast.FunctionDef | ast.AsyncFunctionDef) -> set[str]: """Return the set of bare callee names called inside *func_node*. Only ``ast.Call`` nodes are examined. Bare name references without a call are excluded (they do not indicate the symbol is exercised). Args: func_node: The function or async-function AST node to scan. Returns: Set of bare callee names (e.g. ``{"compute_total", "Invoice"}``) """ names: set[str] = set() for node in ast.walk(func_node): if isinstance(node, ast.Call): name = call_name(node.func) if name: names.add(name) return names # ── Data collection ──────────────────────────────────────────────────────────── def _load_py_blobs( root: pathlib.Path, manifest: Manifest, ) -> BlobMap: """Load every Python/stub blob from *manifest* into memory. A single bulk read avoids redundant object-store I/O for subsequent AST parsing and call-graph construction phases. Args: root: Repository root (object store location). manifest: Snapshot manifest from ``get_commit_snapshot_manifest``. Returns: Mapping ``{file_path: raw_bytes}`` for every Python file in the snapshot. """ from muse.core.object_store import read_object as _read_obj blobs: BlobMap = {} for file_path, obj_id in manifest.items(): if pathlib.PurePosixPath(file_path).suffix.lower() not in _PY_SUFFIXES: continue raw = _read_obj(root, obj_id) if raw is not None: blobs[file_path] = raw return blobs def _build_test_refs(blobs: BlobMap) -> TestRefs: """Scan test files and return ``{test_addr: set[bare_callee_name]}``. Args: blobs: Pre-loaded mapping from ``_load_py_blobs``. Returns: A mapping from each test-function address to the set of bare callee names called anywhere inside that function body. """ refs: TestRefs = {} for file_path, raw in blobs.items(): if not _is_test_file(file_path): continue try: if len(raw) > MAX_AST_BYTES: return {} tree = ast.parse(raw) except SyntaxError: logger.warning("SyntaxError in %s — test file skipped", file_path) continue is_conf = _is_conftest(file_path) for qname, func_node in _collect_test_funcs(tree.body, "", is_conf): addr = f"{file_path}::{qname}" refs[addr] = _scan_calls(func_node) return refs def _build_prod_forward_graph( blobs: BlobMap, prod_files: set[str], ) -> ForwardGraph: """Build a forward call graph for production files using pre-loaded blobs. Reuses ``blobs`` to avoid a second round-trip through the object store. Only callable symbols (functions, methods) are included in the graph. Args: blobs: Pre-loaded Python blobs from ``_load_py_blobs``. prod_files: Set of production file paths to include in the graph. Returns: ``{caller_address: frozenset[callee_bare_name]}`` """ graph: ForwardGraph = {} for file_path, raw in blobs.items(): if file_path not in prod_files: continue try: if len(raw) > MAX_AST_BYTES: return {} tree = ast.parse(raw) except SyntaxError: continue sym_tree = parse_symbols(raw, file_path) for addr, rec in sym_tree.items(): if rec["kind"] not in {"function", "async_function", "method", "async_method"}: continue func_node = find_func_node(tree.body, rec["qualified_name"].split(".")) if func_node is None: continue callees: set[str] = set() for node in ast.walk(func_node): if isinstance(node, ast.Call): n = call_name(node.func) if n: callees.add(n) graph[addr] = frozenset(callees) return graph def _compute_coverage( test_refs: TestRefs, name_to_prod_addrs: NameToAddrs, forward_graph: ForwardGraph | None, depth: int, ) -> CoverageMap: """Compute ``{prod_addr: {test_addr, ...}}`` coverage mapping. Phase 1 — Direct coverage: For each test function, find every bare callee name that matches a production symbol address via *name_to_prod_addrs*. Phase 2 — Transitive expansion (only when ``forward_graph`` is provided and ``depth > 1``): BFS from each directly-covered production symbol, following calls in *forward_graph*. Each newly-reached production symbol is added to the test function's coverage set. The BFS is capped at ``depth - 1`` additional hops. Args: test_refs: ``{test_addr: set[bare_callee_name]}``. name_to_prod_addrs: Reverse index: ``{bare_name: [prod_addr, ...]}``. forward_graph: Production call graph; ``None`` for direct-only mode. depth: Total coverage depth (1 = direct only). Returns: ``{prod_addr: {test_addr, ...}}`` """ # Phase 1: direct coverage coverage: CoverageMap = defaultdict(set) test_to_direct: CoverageMap = defaultdict(set) for test_addr, bare_names in test_refs.items(): for bare in bare_names: for prod_addr in name_to_prod_addrs.get(bare, []): coverage[prod_addr].add(test_addr) test_to_direct[test_addr].add(prod_addr) # Phase 2: transitive expansion if forward_graph is not None and depth > 1: for test_addr, direct_prods in test_to_direct.items(): frontier: set[str] = set(direct_prods) visited: set[str] = set(direct_prods) for _ in range(depth - 1): next_frontier: set[str] = set() for prod_addr in frontier: for callee_bare in forward_graph.get(prod_addr, frozenset()): for callee_addr in name_to_prod_addrs.get(callee_bare, []): if callee_addr not in visited: visited.add(callee_addr) next_frontier.add(callee_addr) coverage[callee_addr].add(test_addr) if not next_frontier: break frontier = next_frontier return dict(coverage) # ── Output builders ──────────────────────────────────────────────────────────── def _build_file_coverage( prod_trees: SymbolTreeMap, coverage: CoverageMap, file_filter: str | None, kind_filter: str | None, ) -> list[_FileCov]: """Build structured ``_FileCov`` records for every production file. Returns all symbols per file (uncovered-only filtering is applied at display/JSON time so summary statistics always reflect the full picture). Args: prod_trees: ``{file_path: {addr: SymbolRecord}}`` for production files. coverage: ``{prod_addr: {test_addr, ...}}`` from ``_compute_coverage``. file_filter: Optional path suffix; only files containing this string are included. kind_filter: Optional symbol kind; only symbols of this kind are included. Returns: List of ``_FileCov`` records sorted by file path. """ result: list[_FileCov] = [] for file_path in sorted(prod_trees): if file_filter and file_filter not in file_path: continue sym_tree = prod_trees[file_path] syms: list[_SymbolCov] = [] for addr in sorted(sym_tree): rec = sym_tree[addr] if rec["kind"] == _IMPORT_KIND: continue if rec["kind"] not in _TRACKED_KINDS: continue if kind_filter and rec["kind"] != kind_filter: continue test_fns = sorted(coverage.get(addr, set())) syms.append( _SymbolCov( address=addr, name=rec["name"], kind=rec["kind"], covered=bool(test_fns), test_functions=test_fns, ) ) if not syms: continue covered_count = sum(1 for s in syms if s["covered"]) result.append( _FileCov( file=file_path, total_symbols=len(syms), covered_symbols=covered_count, uncovered_symbols=len(syms) - covered_count, coverage_pct=round(100.0 * covered_count / len(syms), 1), symbols=syms, ) ) return result # ── Formatting ───────────────────────────────────────────────────────────────── _BAR_WIDTH = 20 _KindAbbrevMap = dict[str, str] _KIND_ABBREV: _KindAbbrevMap = { "function": "func", "async_function": "afn ", "method": "meth", "async_method": "amth", "class": "cls ", } def _bar(pct: float) -> str: """Return a 20-char Unicode block bar representing *pct* (0–100).""" filled = round(pct / (100 / _BAR_WIDTH)) return "█" * filled + "░" * (_BAR_WIDTH - filled) def _print_table( files: list[_FileCov], uncovered_only: bool, show_tests: bool, min_coverage: int, ref: str, total_test_fns: int, ) -> bool: """Print human-readable coverage table to stdout. Args: files: Structured file coverage from ``_build_file_coverage``. uncovered_only: When True, only uncovered symbols are printed per file. show_tests: When True, list covering test functions under each symbol. min_coverage: Coverage threshold for ⚠️ flag and exit-1 signalling. ref: Display string for the analysed ref (e.g. ``"HEAD"``). total_test_fns: Total number of test functions found in the snapshot. Returns: True if any file violates *min_coverage*; False otherwise. """ total_syms = sum(f["total_symbols"] for f in files) total_covered = sum(f["covered_symbols"] for f in files) total_pct = round(100.0 * total_covered / total_syms, 1) if total_syms else 0.0 print( f"Semantic test coverage — {ref}" f" ({total_syms} symbols · {total_test_fns} test functions)\n" ) violation = False for fc in files: pct = fc["coverage_pct"] bar = _bar(pct) below = pct < min_coverage and min_coverage > 0 if below: violation = True flag = " ⚠️" if below else "" print(f" {fc['file']}") print( f" [{bar}] {pct:5.1f}%" f" ({fc['covered_symbols']}/{fc['total_symbols']}){flag}" ) for sym in fc["symbols"]: if uncovered_only and sym["covered"]: continue icon = "✅" if sym["covered"] else "❌" kind_short = _KIND_ABBREV.get(sym["kind"], sym["kind"][:4]) print(f" {icon} {sanitize_display(sym['name']):<42} {kind_short}") if show_tests and sym["test_functions"]: for tf in sym["test_functions"]: print(f" ← {tf}") print() sep = "─" * 64 print(f" {sep}") print(f" TOTAL: {total_covered}/{total_syms} symbols covered ({total_pct}%)\n") if violation: print( f" ⚠️ One or more files are below the {min_coverage}%" " minimum coverage threshold." ) return violation # ── Entry point ──────────────────────────────────────────────────────────────── def run(args: argparse.Namespace) -> None: """Entry point for ``muse code semantic-test-coverage``. Validates arguments, loads the HEAD snapshot, runs the coverage analysis, and prints results (or emits JSON). Exits 1 when ``--min-coverage`` is set and any file falls below the threshold. """ root = require_repo() # ── Argument validation ──────────────────────────────────────────────────── depth: int = clamp_int(args.depth, 1, 50, 'depth') if depth < 1 or depth > _MAX_DEPTH: print(f"❌ --depth must be between 1 and {_MAX_DEPTH}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) min_coverage: int = clamp_int(args.min_coverage, 0, 100, 'min_coverage') if not (0 <= min_coverage <= 100): print("❌ --min-coverage must be between 0 and 100.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) transitive: bool = args.transitive or depth > 1 effective_depth: int = depth if transitive else 1 kind_filter: str | None = args.kind or None file_filter: str | None = args.file or None # ── Resolve HEAD snapshot ────────────────────────────────────────────────── repo_id = read_repo_id(root) branch = read_current_branch(root) head = resolve_commit_ref(root, repo_id, branch, None) if head is None: print("❌ HEAD commit not found — is this an empty repository?", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) manifest: Manifest = get_commit_snapshot_manifest(root, head.commit_id) or {} # ── Single bulk read of all Python blobs ─────────────────────────────────── blobs = _load_py_blobs(root, manifest) # ── Symbol extraction ────────────────────────────────────────────────────── all_trees = symbols_for_snapshot(root, manifest) # Partition: test files vs. production files. prod_trees: SymbolTreeMap = {} for file_path, sym_tree in all_trees.items(): if not _is_test_file(file_path): prod_trees[file_path] = sym_tree # Flat index of all production symbols, excluding imports. all_prod: FlatSymbolMap = {} for sym_tree in prod_trees.values(): for addr, rec in sym_tree.items(): if rec["kind"] not in (_IMPORT_KIND,) and rec["kind"] in _TRACKED_KINDS: all_prod[addr] = rec # Reverse index: bare_name → [prod_addr, ...] name_to_addrs: NameToAddrs = defaultdict(list) for addr, rec in all_prod.items(): name_to_addrs[rec["name"]].append(addr) # ── Test-function scanning ───────────────────────────────────────────────── test_refs = _build_test_refs(blobs) total_test_fns = len(test_refs) # ── Optional: build production call graph for transitive coverage ────────── forward_graph: ForwardGraph | None = None if transitive: forward_graph = _build_prod_forward_graph(blobs, set(prod_trees)) # ── Coverage computation ─────────────────────────────────────────────────── coverage = _compute_coverage( test_refs, dict(name_to_addrs), forward_graph, effective_depth ) # ── Structured output ────────────────────────────────────────────────────── file_coverage = _build_file_coverage(prod_trees, coverage, file_filter, kind_filter) # ── JSON output ──────────────────────────────────────────────────────────── if args.json: total_syms = sum(f["total_symbols"] for f in file_coverage) total_covered = sum(f["covered_symbols"] for f in file_coverage) pct = round(100.0 * total_covered / total_syms, 1) if total_syms else 0.0 # Apply uncovered_only filter to symbol lists in JSON. if args.uncovered_only: filtered_files: list[_FileCov] = [] for fc in file_coverage: uncov = [s for s in fc["symbols"] if not s["covered"]] if uncov: filtered_files.append( _FileCov( file=fc["file"], total_symbols=fc["total_symbols"], covered_symbols=fc["covered_symbols"], uncovered_symbols=fc["uncovered_symbols"], coverage_pct=fc["coverage_pct"], symbols=uncov, ) ) file_coverage = filtered_files out: _JsonOut = _JsonOut( ref="HEAD", snapshot_id=head.commit_id[:12], depth=effective_depth, transitive=transitive, filters=_FilterSpec( file=file_filter, kind=kind_filter, min_coverage=min_coverage if min_coverage > 0 else None, uncovered_only=args.uncovered_only, ), summary=_SummarySpec( total_symbols=total_syms, covered_symbols=total_covered, uncovered_symbols=total_syms - total_covered, coverage_pct=pct, total_test_functions=total_test_fns, total_production_files=len(file_coverage), ), files=file_coverage, ) print(json.dumps(out, indent=2)) return # ── Human-readable output ────────────────────────────────────────────────── violation = _print_table( file_coverage, uncovered_only=args.uncovered_only, show_tests=args.show_tests, min_coverage=min_coverage, ref="HEAD", total_test_fns=total_test_fns, ) if violation: sys.exit(1) # ── CLI registration ─────────────────────────────────────────────────────────── def register( sub: argparse._SubParsersAction[argparse.ArgumentParser], ) -> None: """Register ``semantic-test-coverage`` under the ``code`` subcommand group. Args: sub: The subparser action from the ``code`` command group. """ p = sub.add_parser( "semantic-test-coverage", help=( "Static symbol-level test coverage — which symbols are exercised" " by which tests, without running the test suite." ), description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) p.add_argument( "--file", metavar="SUFFIX", help=( "Scope analysis to production files whose path contains SUFFIX" " (e.g. --file billing.py)." ), ) p.add_argument( "--kind", metavar="KIND", choices=["function", "async_function", "method", "async_method", "class"], help="Filter symbols by kind.", ) p.add_argument( "--transitive", action="store_true", help=( "Expand coverage through the production call graph." " Combines with --depth (default: 2 when this flag is set)." ), ) p.add_argument( "--depth", type=int, default=_DEFAULT_DEPTH, metavar="D", help=( f"Transitive call-graph depth (default: {_DEFAULT_DEPTH})." " Values > 1 imply --transitive." f" Maximum: {_MAX_DEPTH}." ), ) p.add_argument( "--uncovered-only", action="store_true", help="Show (or emit in JSON) only symbols with no test coverage.", ) p.add_argument( "--show-tests", action="store_true", help="Under each covered symbol, list the test functions that exercise it.", ) p.add_argument( "--min-coverage", type=int, default=0, metavar="PCT", help=( "Exit 1 if any production file is below PCT%% semantic coverage." " Useful for CI gates." ), ) p.add_argument( "--json", action="store_true", help="Emit JSON instead of human-readable text.", ) p.set_defaults(func=run)