"""muse code coverage — class interface call-coverage. Reports which methods of a class are actually called somewhere in the committed snapshot and which are never reached. This command answers the question: *"Is my API actually used?"* Every ``class`` symbol with method children is a candidate interface. ``muse code coverage`` builds the reverse call graph for the snapshot, then checks each method's bare name against the set of called names. Why this matters ---------------- Traditional coverage tools measure *test* coverage — how many lines are executed during a test run. That requires a running test suite. Muse's *interface coverage* measures *call-site* coverage — how many of a class's methods are invoked anywhere in the production codebase. It runs in O(snapshot_size) without executing any code. It is ideal for: * Auditing API surface before a deprecation. * Finding method pairs where one is always called and the other never is. * Verifying that a new interface is actually adopted after landing. * Tracking coverage drift across commits with ``--compare``. Usage:: muse code coverage "src/models.py::User" muse code coverage "src/auth.py::TokenValidator" --commit HEAD~5 muse code coverage "src/billing.py::Invoice" --json muse code coverage "src/billing.py::Invoice" --exclude-dunder muse code coverage "src/billing.py::Invoice" --exclude-self muse code coverage "src/billing.py::Invoice" --min-callers 2 muse code coverage "src/billing.py::Invoice" --compare HEAD~10 muse code coverage "src/billing.py::Invoice" --count Output:: Interface coverage: src/models.py::User ────────────────────────────────────────────────────────────── ✅ User.__init__ called by: src/api.py::create_user, src/api.py::update_user ✅ User.save called by: src/api.py::create_user ❌ User.delete (no callers detected) ❌ User.to_dict (no callers detected) ────────────────────────────────────────────────────────────── Coverage: 2/4 methods called (50%) 🟡 Partial coverage — 2 uncovered method(s) may be dead API surface. Flags: ``--commit, -c REF`` Analyse a historical snapshot instead of HEAD. ``--compare REF`` Diff coverage against this commit. Shows which methods gained or lost callers between the two snapshots. ``--exclude-dunder`` Exclude dunder methods (``__init__``, ``__repr__``, ``__eq__``, …) from the coverage count. These are called implicitly by Python internals and never appear in the call graph, producing unavoidable false negatives. ``--exclude-private`` Exclude methods whose name starts with a single underscore. ``--min-callers N`` A method only counts as "covered" when called from at least N distinct call-sites. Separates "used in one test" from "widely adopted". ``--exclude-self`` Only count callers outside the class's own file. Answers: "does anyone *external* actually use this method?" ``--no-show-callers`` Suppress caller addresses next to each covered method. ``--count`` Print only ``n_covered/total`` (scriptable). ``--json`` Emit results as JSON. """ from __future__ import annotations import argparse import difflib import json import logging import pathlib import sys from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from typing import TypedDict from muse.core.store import ( CommitRecord, get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref, ) from muse.core.symbol_cache import load_symbol_cache from muse.plugins.code._callgraph import ReverseGraph, build_reverse_graph from muse.plugins.code._query import symbols_for_snapshot from muse.plugins.code.ast_parser import SymbolRecord from muse.core.validation import sanitize_display logger = logging.getLogger(__name__) type _SymbolMap = dict[str, dict[str, SymbolRecord]] type _AddrKindMap = dict[str, str] class _CoverageFilters(TypedDict): exclude_dunder: bool exclude_private: bool min_callers: int exclude_self: bool class _MethodJson(TypedDict): address: str name: str called: bool callers: list[str] class _CoveragePayload(TypedDict, total=False): """JSON payload for ``muse code coverage`` output.""" address: str commit_id: str total_methods: int covered: int percent: float filters: _CoverageFilters methods: list[_MethodJson] compare_commit_id: str newly_covered: list[str] newly_uncovered: list[str] percent_change: float _METHOD_KINDS: frozenset[str] = frozenset({"method", "async_method"}) _CLASS_KINDS: frozenset[str] = frozenset({"class", "async_class"}) # --------------------------------------------------------------------------- # Analysis helpers # --------------------------------------------------------------------------- def _class_methods( file_path: str, class_name: str, symbol_map: _SymbolMap, ) -> list[tuple[str, str]]: """Return ``(address, bare_name)`` pairs for all methods of *class_name*. Uses a direct dict lookup on *file_path* — O(1) instead of iterating all files in the symbol map. """ methods: list[tuple[str, str]] = [] prefix = f"{file_path}::{class_name}." tree = symbol_map.get(file_path, {}) for address, rec in sorted(tree.items()): if rec["kind"] not in _METHOD_KINDS: continue if address.startswith(prefix): bare = rec["name"].split(".")[-1] methods.append((address, bare)) return sorted(methods, key=lambda t: t[1]) def _filter_methods( methods: list[tuple[str, str]], exclude_dunder: bool, exclude_private: bool, ) -> list[tuple[str, str]]: """Apply name-based filters to the raw method list.""" result: list[tuple[str, str]] = [] for addr, bare in methods: if exclude_dunder and bare.startswith("__") and bare.endswith("__"): continue if exclude_private and bare.startswith("_") and not (bare.startswith("__") and bare.endswith("__")): continue result.append((addr, bare)) return result def _classify_methods( methods: list[tuple[str, str]], reverse: ReverseGraph, min_callers: int, exclude_self_file: str | None, ) -> tuple[list[tuple[str, str, list[str]]], list[tuple[str, str]]]: """Separate methods into covered and uncovered lists. Returns ``(covered, uncovered)`` where covered entries are ``(address, bare_name, callers)`` and uncovered are ``(address, bare_name)``. """ covered: list[tuple[str, str, list[str]]] = [] uncovered: list[tuple[str, str]] = [] for method_addr, bare_name in methods: all_callers = sorted(reverse.get(bare_name, [])) if exclude_self_file: all_callers = [c for c in all_callers if c.split("::")[0] != exclude_self_file] if len(all_callers) >= min_callers: covered.append((method_addr, bare_name, all_callers)) else: uncovered.append((method_addr, bare_name)) return covered, uncovered def _analyse_snapshot( root: pathlib.Path, commit: CommitRecord, file_path: str, class_name: str, exclude_dunder: bool, exclude_private: bool, min_callers: int, exclude_self_file: str | None, ) -> tuple[ list[tuple[str, str, list[str]]], list[tuple[str, str]], list[tuple[str, str]], ]: """Full analysis pipeline for one commit snapshot. Returns ``(covered, uncovered, all_methods)`` after applying all filters. Uses a single shared ``SymbolCache`` for both ``symbols_for_snapshot`` and ``build_reverse_graph``. """ manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} cache = load_symbol_cache(root) symbol_map = symbols_for_snapshot(root, manifest, cache=cache) reverse = build_reverse_graph(root, manifest, cache=cache) cache.save() all_methods = _class_methods(file_path, class_name, symbol_map) filtered = _filter_methods(all_methods, exclude_dunder, exclude_private) covered, uncovered = _classify_methods(filtered, reverse, min_callers, exclude_self_file) return covered, uncovered, filtered def _find_class_suggestions( class_addr: str, file_path: str, class_name: str, symbol_map: _SymbolMap, ) -> tuple[list[str], list[str], list[str]]: """Return (same_file_classes, same_name_classes, fuzzy_matches).""" all_syms: _AddrKindMap = { addr: rec["kind"] for tree in symbol_map.values() for addr, rec in tree.items() } same_file = sorted( a for a, k in all_syms.items() if k in _CLASS_KINDS and a.startswith(f"{file_path}::") ) same_name = sorted( a for a, k in all_syms.items() if k in _CLASS_KINDS and a.endswith(f"::{class_name}") ) all_class_addrs = [a for a, k in all_syms.items() if k in _CLASS_KINDS] fuzzy = difflib.get_close_matches(class_addr, all_class_addrs, n=5, cutoff=0.4) fuzzy = [a for a in fuzzy if a not in same_file and a not in same_name] return same_file, same_name, fuzzy # --------------------------------------------------------------------------- # CLI registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the coverage subcommand.""" parser = subparsers.add_parser( "coverage", help="Show which methods of a class are called anywhere in the snapshot.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "address", metavar="CLASS_ADDRESS", help='Class symbol address, e.g. "src/models.py::User".', ) parser.add_argument( "--commit", "-c", default=None, metavar="REF", dest="ref", help="Analyse a historical snapshot instead of HEAD.", ) parser.add_argument( "--compare", default=None, metavar="REF", dest="compare_ref", help="Diff coverage against this commit (shows methods gained/lost).", ) parser.add_argument( "--exclude-dunder", action="store_true", dest="exclude_dunder", help="Exclude dunder methods (__init__, __repr__, …) from the count.", ) parser.add_argument( "--exclude-private", action="store_true", dest="exclude_private", help="Exclude single-underscore private methods from the count.", ) parser.add_argument( "--min-callers", type=int, default=1, metavar="N", dest="min_callers", help="Minimum distinct call-sites to count a method as covered (default: 1).", ) parser.add_argument( "--exclude-self", action="store_true", dest="exclude_self", help="Only count callers outside the class's own file.", ) parser.add_argument( "--no-show-callers", action="store_false", dest="show_callers", help="Suppress caller addresses next to each covered method.", ) parser.add_argument( "--count", action="store_true", dest="count_only", help="Print only 'n_covered/total' (scriptable).", ) parser.add_argument( "--json", action="store_true", dest="as_json", help="Emit results as JSON.", ) parser.set_defaults(func=run, show_callers=True) # --------------------------------------------------------------------------- # Command entry point # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Show which methods of a class are called anywhere in the snapshot. Builds the reverse call graph, then checks each method's bare name against the set of called names. Reports covered and uncovered methods and a percentage coverage score. Python only (call-graph analysis uses stdlib ``ast``). """ address: str = args.address ref: str | None = args.ref compare_ref: str | None = args.compare_ref show_callers: bool = args.show_callers exclude_dunder: bool = args.exclude_dunder exclude_private: bool = args.exclude_private min_callers: int = max(1, args.min_callers) exclude_self: bool = args.exclude_self count_only: bool = args.count_only as_json: bool = args.as_json root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) if "::" not in address: print("❌ ADDRESS must be a symbol address like 'src/models.py::User'.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) file_path, class_name = address.split("::", 1) class_addr = f"{file_path}::{class_name}" exclude_self_file = file_path if exclude_self else None commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: print(f"❌ Commit '{ref or 'HEAD'}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # Verify the class exists — load symbol map for validation and suggestions. manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} cache = load_symbol_cache(root) symbol_map = symbols_for_snapshot(root, manifest, cache=cache) tree_for_file = symbol_map.get(file_path, {}) if class_addr not in tree_for_file: cache.save() _emit_not_found(class_addr, file_path, class_name, symbol_map, as_json, commit) raise SystemExit(ExitCode.USER_ERROR) all_methods = _class_methods(file_path, class_name, symbol_map) if not all_methods: cache.save() print(f"⚠️ No methods found for '{class_addr}'.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # Build reverse graph sharing the already-warm cache. reverse = build_reverse_graph(root, manifest, cache=cache) cache.save() filtered = _filter_methods(all_methods, exclude_dunder, exclude_private) covered, uncovered = _classify_methods(filtered, reverse, min_callers, exclude_self_file) total = len(filtered) n_covered = len(covered) pct = round(n_covered / total * 100) if total else 0 # --compare diff compare_commit: CommitRecord | None = None newly_covered: list[str] = [] newly_uncovered: list[str] = [] pct_change: int = 0 if compare_ref: compare_commit = resolve_commit_ref(root, repo_id, branch, compare_ref) if compare_commit is None: print(f"❌ --compare commit '{compare_ref}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) cmp_covered, cmp_uncovered, _ = _analyse_snapshot( root, compare_commit, file_path, class_name, exclude_dunder, exclude_private, min_callers, exclude_self_file, ) cmp_covered_names = {name for _, name, _ in cmp_covered} cur_covered_names = {name for _, name, _ in covered} newly_covered = sorted(cur_covered_names - cmp_covered_names) newly_uncovered = sorted(cmp_covered_names - cur_covered_names) cmp_total = len(cmp_covered) + len(cmp_uncovered) cmp_pct = round(len(cmp_covered) / cmp_total * 100) if cmp_total else 0 pct_change = pct - cmp_pct # --count if count_only and not as_json: print(f"{n_covered}/{total}") return if as_json: methods: list[_MethodJson] = [ _MethodJson(address=addr, name=name, called=True, callers=callers) for addr, name, callers in covered ] + [ _MethodJson(address=addr, name=name, called=False, callers=[]) for addr, name in uncovered ] payload = _CoveragePayload( address=class_addr, commit_id=commit.commit_id, total_methods=total, covered=n_covered, percent=pct, filters=_CoverageFilters( exclude_dunder=exclude_dunder, exclude_private=exclude_private, min_callers=min_callers, exclude_self=exclude_self, ), methods=methods, ) if compare_commit is not None: payload["compare_commit_id"] = compare_commit.commit_id payload["newly_covered"] = newly_covered payload["newly_uncovered"] = newly_uncovered payload["percent_change"] = pct_change print(json.dumps(payload, indent=2)) return print(f"\nInterface coverage: {class_addr}") active_filters: list[str] = [] if exclude_dunder: active_filters.append("no dunders") if exclude_private: active_filters.append("no private") if min_callers > 1: active_filters.append(f"min {min_callers} callers") if exclude_self: active_filters.append("external callers only") if active_filters: print(f"Filters: {', '.join(active_filters)}") print("─" * 62) max_name = max( (len(f"{class_name}.{name}") for _, name in filtered), default=0, ) for addr, bare_name, callers in covered: display = f"{class_name}.{bare_name}" line = f" ✅ {display:<{max_name}}" if show_callers: caller_str = ", ".join(callers[:3]) if len(callers) > 3: caller_str += f" (+{len(callers) - 3} more)" line += f" ← {caller_str}" print(line) for addr, bare_name in uncovered: display = f"{class_name}.{bare_name}" print(f" ❌ {display:<{max_name}} (no callers detected)") print("\n" + "─" * 62) print(f"Coverage: {n_covered}/{total} methods called ({pct}%)") if pct == 100: print("✅ Full coverage — all methods are called at least once.") elif pct >= 75: print(f"🟢 Good coverage — {total - n_covered} uncovered method(s).") elif pct >= 50: print(f"🟡 Partial coverage — {total - n_covered} uncovered method(s) may be dead API surface.") else: print(f"🔴 Low coverage — {total - n_covered} of {total} methods have no detected callers.") if compare_commit is not None: sign = "+" if pct_change >= 0 else "" print(f"\nCoverage diff vs {compare_commit.commit_id[:8]}: {sign}{pct_change}%") if newly_covered: print(f" Newly covered ({len(newly_covered)}): {', '.join(newly_covered)}") if newly_uncovered: print(f" Lost coverage ({len(newly_uncovered)}): {', '.join(newly_uncovered)}") print( "\nNote: dynamic dispatch, subclass overrides, and external callers are not detected." ) # --------------------------------------------------------------------------- # Error rendering # --------------------------------------------------------------------------- def _emit_not_found( class_addr: str, file_path: str, class_name: str, symbol_map: _SymbolMap, as_json: bool, commit: CommitRecord, ) -> None: """Print a helpful not-found error with suggestions.""" same_file, same_name, fuzzy = _find_class_suggestions( class_addr, file_path, class_name, symbol_map ) if as_json: print(json.dumps({ "error": "symbol_not_found", "address": class_addr, "commit_id": commit.commit_id, "suggestions": same_file[:8] or same_name[:5] or fuzzy[:5], }, indent=2)) return print(f"❌ '{class_addr}' not found in snapshot {commit.commit_id[:8]}.", file=sys.stderr) if same_file: print(f"\n Classes in {sanitize_display(file_path)}:", file=sys.stderr) for c in same_file[:8]: print(f" {c}", file=sys.stderr) if same_name and same_name != same_file: print(f"\n '{class_name}' found at:", file=sys.stderr) for c in same_name[:5]: print(f" {c}", file=sys.stderr) if fuzzy and not same_file and not same_name: print("\n Did you mean:", file=sys.stderr) for c in fuzzy[:5]: print(f" {c}", file=sys.stderr) if not same_file and not same_name and not fuzzy: print( f"\n No classes found in {file_path}. " "Check 'muse code symbols --json | jq' for valid addresses.", file=sys.stderr, )