coverage.py file-level

at sha256:6 · View file ↗ · Intel ↗

History
1 files
1 commits
0 hotspots
0 🧊 dead
0 πŸ’₯ blast risk
sha256:c docs(KD-STAGING): sync governance after KD-6b DONE Mirror workspace go… · aaronrene · Jul 10, 2026
1 """muse code coverage β€” class interface call-coverage.
2
3 Reports which methods of a class are actually called somewhere in the
4 committed snapshot and which are never reached.
5
6 This command answers the question: *"Is my API actually used?"*
7
8 Every ``class`` symbol with method children is a candidate interface.
9 ``muse code coverage`` builds the reverse call graph for the snapshot, then
10 checks each method's bare name against the set of called names.
11
12 Why this matters
13 ----------------
14 Traditional coverage tools measure *test* coverage β€” how many lines are
15 executed during a test run. That requires a running test suite.
16
17 Muse's *interface coverage* measures *call-site* coverage β€” how many of
18 a class's methods are invoked anywhere in the production codebase. It
19 runs in O(snapshot_size) without executing any code. It is ideal for:
20
21 * Auditing API surface before a deprecation.
22 * Finding method pairs where one is always called and the other never is.
23 * Verifying that a new interface is actually adopted after landing.
24 * Tracking coverage drift across commits with ``--compare``.
25
26 Usage::
27
28 muse code coverage "src/models.py::User"
29 muse code coverage "src/auth.py::TokenValidator" --commit HEAD~5
30 muse code coverage "src/billing.py::Invoice" --json
31 muse code coverage "src/billing.py::Invoice" --exclude-dunder
32 muse code coverage "src/billing.py::Invoice" --exclude-self
33 muse code coverage "src/billing.py::Invoice" --min-callers 2
34 muse code coverage "src/billing.py::Invoice" --compare HEAD~10
35 muse code coverage "src/billing.py::Invoice" --count
36
37 Output::
38
39 Interface coverage: src/models.py::User
40 ──────────────────────────────────────────────────────────────
41
42 βœ… User.__init__ called by: src/api.py::create_user, src/api.py::update_user
43 βœ… User.save called by: src/api.py::create_user
44 ❌ User.delete (no callers detected)
45 ❌ User.to_dict (no callers detected)
46
47 ──────────────────────────────────────────────────────────────
48 Coverage: 2/4 methods called (50%)
49 🟑 Partial coverage β€” 2 uncovered method(s) may be dead API surface.
50
51 Flags:
52
53 ``--commit, -c REF``
54 Analyse a historical snapshot instead of HEAD.
55
56 ``--compare REF``
57 Diff coverage against this commit. Shows which methods gained or lost
58 callers between the two snapshots.
59
60 ``--exclude-dunder``
61 Exclude dunder methods (``__init__``, ``__repr__``, ``__eq__``, …) from
62 the coverage count. These are called implicitly by Python internals and
63 never appear in the call graph, producing unavoidable false negatives.
64
65 ``--exclude-private``
66 Exclude methods whose name starts with a single underscore.
67
68 ``--min-callers N``
69 A method only counts as "covered" when called from at least N distinct
70 call-sites. Separates "used in one test" from "widely adopted".
71
72 ``--exclude-self``
73 Only count callers outside the class's own file. Answers: "does anyone
74 *external* actually use this method?"
75
76 ``--no-show-callers``
77 Suppress caller addresses next to each covered method.
78
79 ``--count``
80 Print only ``n_covered/total`` (scriptable).
81
82 ``--json``
83 Emit results as JSON.
84 """
85
86 from __future__ import annotations
87
88 import argparse
89 import difflib
90 import json
91 import logging
92 import pathlib
93 import sys
94
95 from muse.core.errors import ExitCode
96 from muse.core.repo import read_repo_id, require_repo
97 from typing import TypedDict
98
99 from muse.core.store import (
100 CommitRecord,
101 get_commit_snapshot_manifest,
102 read_current_branch,
103 resolve_commit_ref,
104 )
105 from muse.core.symbol_cache import load_symbol_cache
106 from muse.plugins.code._callgraph import ReverseGraph, build_reverse_graph
107 from muse.plugins.code._query import symbols_for_snapshot
108 from muse.plugins.code.ast_parser import SymbolRecord
109 from muse.core.validation import sanitize_display
110
111 logger = logging.getLogger(__name__)
112
113 type _SymbolMap = dict[str, dict[str, SymbolRecord]]
114 type _AddrKindMap = dict[str, str]
115
116
117 class _CoverageFilters(TypedDict):
118 exclude_dunder: bool
119 exclude_private: bool
120 min_callers: int
121 exclude_self: bool
122
123
124 class _MethodJson(TypedDict):
125 address: str
126 name: str
127 called: bool
128 callers: list[str]
129
130
131 class _CoveragePayload(TypedDict, total=False):
132 """JSON payload for ``muse code coverage`` output."""
133
134 address: str
135 commit_id: str
136 total_methods: int
137 covered: int
138 percent: float
139 filters: _CoverageFilters
140 methods: list[_MethodJson]
141 compare_commit_id: str
142 newly_covered: list[str]
143 newly_uncovered: list[str]
144 percent_change: float
145
146
147 _METHOD_KINDS: frozenset[str] = frozenset({"method", "async_method"})
148 _CLASS_KINDS: frozenset[str] = frozenset({"class", "async_class"})
149
150
151
152 # ---------------------------------------------------------------------------
153 # Analysis helpers
154 # ---------------------------------------------------------------------------
155
156
157 def _class_methods(
158 file_path: str,
159 class_name: str,
160 symbol_map: _SymbolMap,
161 ) -> list[tuple[str, str]]:
162 """Return ``(address, bare_name)`` pairs for all methods of *class_name*.
163
164 Uses a direct dict lookup on *file_path* β€” O(1) instead of iterating all
165 files in the symbol map.
166 """
167 methods: list[tuple[str, str]] = []
168 prefix = f"{file_path}::{class_name}."
169 tree = symbol_map.get(file_path, {})
170 for address, rec in sorted(tree.items()):
171 if rec["kind"] not in _METHOD_KINDS:
172 continue
173 if address.startswith(prefix):
174 bare = rec["name"].split(".")[-1]
175 methods.append((address, bare))
176 return sorted(methods, key=lambda t: t[1])
177
178
179 def _filter_methods(
180 methods: list[tuple[str, str]],
181 exclude_dunder: bool,
182 exclude_private: bool,
183 ) -> list[tuple[str, str]]:
184 """Apply name-based filters to the raw method list."""
185 result: list[tuple[str, str]] = []
186 for addr, bare in methods:
187 if exclude_dunder and bare.startswith("__") and bare.endswith("__"):
188 continue
189 if exclude_private and bare.startswith("_") and not (bare.startswith("__") and bare.endswith("__")):
190 continue
191 result.append((addr, bare))
192 return result
193
194
195 def _classify_methods(
196 methods: list[tuple[str, str]],
197 reverse: ReverseGraph,
198 min_callers: int,
199 exclude_self_file: str | None,
200 ) -> tuple[list[tuple[str, str, list[str]]], list[tuple[str, str]]]:
201 """Separate methods into covered and uncovered lists.
202
203 Returns ``(covered, uncovered)`` where covered entries are
204 ``(address, bare_name, callers)`` and uncovered are ``(address, bare_name)``.
205 """
206 covered: list[tuple[str, str, list[str]]] = []
207 uncovered: list[tuple[str, str]] = []
208
209 for method_addr, bare_name in methods:
210 all_callers = sorted(reverse.get(bare_name, []))
211 if exclude_self_file:
212 all_callers = [c for c in all_callers if c.split("::")[0] != exclude_self_file]
213 if len(all_callers) >= min_callers:
214 covered.append((method_addr, bare_name, all_callers))
215 else:
216 uncovered.append((method_addr, bare_name))
217
218 return covered, uncovered
219
220
221 def _analyse_snapshot(
222 root: pathlib.Path,
223 commit: CommitRecord,
224 file_path: str,
225 class_name: str,
226 exclude_dunder: bool,
227 exclude_private: bool,
228 min_callers: int,
229 exclude_self_file: str | None,
230 ) -> tuple[
231 list[tuple[str, str, list[str]]],
232 list[tuple[str, str]],
233 list[tuple[str, str]],
234 ]:
235 """Full analysis pipeline for one commit snapshot.
236
237 Returns ``(covered, uncovered, all_methods)`` after applying all filters.
238 Uses a single shared ``SymbolCache`` for both ``symbols_for_snapshot``
239 and ``build_reverse_graph``.
240 """
241 manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {}
242 cache = load_symbol_cache(root)
243 symbol_map = symbols_for_snapshot(root, manifest, cache=cache)
244 reverse = build_reverse_graph(root, manifest, cache=cache)
245 cache.save()
246
247 all_methods = _class_methods(file_path, class_name, symbol_map)
248 filtered = _filter_methods(all_methods, exclude_dunder, exclude_private)
249 covered, uncovered = _classify_methods(filtered, reverse, min_callers, exclude_self_file)
250 return covered, uncovered, filtered
251
252
253 def _find_class_suggestions(
254 class_addr: str,
255 file_path: str,
256 class_name: str,
257 symbol_map: _SymbolMap,
258 ) -> tuple[list[str], list[str], list[str]]:
259 """Return (same_file_classes, same_name_classes, fuzzy_matches)."""
260 all_syms: _AddrKindMap = {
261 addr: rec["kind"]
262 for tree in symbol_map.values()
263 for addr, rec in tree.items()
264 }
265 same_file = sorted(
266 a for a, k in all_syms.items()
267 if k in _CLASS_KINDS and a.startswith(f"{file_path}::")
268 )
269 same_name = sorted(
270 a for a, k in all_syms.items()
271 if k in _CLASS_KINDS and a.endswith(f"::{class_name}")
272 )
273 all_class_addrs = [a for a, k in all_syms.items() if k in _CLASS_KINDS]
274 fuzzy = difflib.get_close_matches(class_addr, all_class_addrs, n=5, cutoff=0.4)
275 fuzzy = [a for a in fuzzy if a not in same_file and a not in same_name]
276 return same_file, same_name, fuzzy
277
278
279 # ---------------------------------------------------------------------------
280 # CLI registration
281 # ---------------------------------------------------------------------------
282
283
284 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
285 """Register the coverage subcommand."""
286 parser = subparsers.add_parser(
287 "coverage",
288 help="Show which methods of a class are called anywhere in the snapshot.",
289 description=__doc__,
290 formatter_class=argparse.RawDescriptionHelpFormatter,
291 )
292 parser.add_argument(
293 "address", metavar="CLASS_ADDRESS",
294 help='Class symbol address, e.g. "src/models.py::User".',
295 )
296 parser.add_argument(
297 "--commit", "-c", default=None, metavar="REF", dest="ref",
298 help="Analyse a historical snapshot instead of HEAD.",
299 )
300 parser.add_argument(
301 "--compare", default=None, metavar="REF", dest="compare_ref",
302 help="Diff coverage against this commit (shows methods gained/lost).",
303 )
304 parser.add_argument(
305 "--exclude-dunder", action="store_true", dest="exclude_dunder",
306 help="Exclude dunder methods (__init__, __repr__, …) from the count.",
307 )
308 parser.add_argument(
309 "--exclude-private", action="store_true", dest="exclude_private",
310 help="Exclude single-underscore private methods from the count.",
311 )
312 parser.add_argument(
313 "--min-callers", type=int, default=1, metavar="N", dest="min_callers",
314 help="Minimum distinct call-sites to count a method as covered (default: 1).",
315 )
316 parser.add_argument(
317 "--exclude-self", action="store_true", dest="exclude_self",
318 help="Only count callers outside the class's own file.",
319 )
320 parser.add_argument(
321 "--no-show-callers", action="store_false", dest="show_callers",
322 help="Suppress caller addresses next to each covered method.",
323 )
324 parser.add_argument(
325 "--count", action="store_true", dest="count_only",
326 help="Print only 'n_covered/total' (scriptable).",
327 )
328 parser.add_argument(
329 "--json", action="store_true", dest="as_json",
330 help="Emit results as JSON.",
331 )
332 parser.set_defaults(func=run, show_callers=True)
333
334
335 # ---------------------------------------------------------------------------
336 # Command entry point
337 # ---------------------------------------------------------------------------
338
339
340 def run(args: argparse.Namespace) -> None:
341 """Show which methods of a class are called anywhere in the snapshot.
342
343 Builds the reverse call graph, then checks each method's bare name
344 against the set of called names. Reports covered and uncovered methods
345 and a percentage coverage score.
346
347 Python only (call-graph analysis uses stdlib ``ast``).
348 """
349 address: str = args.address
350 ref: str | None = args.ref
351 compare_ref: str | None = args.compare_ref
352 show_callers: bool = args.show_callers
353 exclude_dunder: bool = args.exclude_dunder
354 exclude_private: bool = args.exclude_private
355 min_callers: int = max(1, args.min_callers)
356 exclude_self: bool = args.exclude_self
357 count_only: bool = args.count_only
358 as_json: bool = args.as_json
359
360 root = require_repo()
361 repo_id = read_repo_id(root)
362 branch = read_current_branch(root)
363
364 if "::" not in address:
365 print("❌ ADDRESS must be a symbol address like 'src/models.py::User'.", file=sys.stderr)
366 raise SystemExit(ExitCode.USER_ERROR)
367
368 file_path, class_name = address.split("::", 1)
369 class_addr = f"{file_path}::{class_name}"
370 exclude_self_file = file_path if exclude_self else None
371
372 commit = resolve_commit_ref(root, repo_id, branch, ref)
373 if commit is None:
374 print(f"❌ Commit '{ref or 'HEAD'}' not found.", file=sys.stderr)
375 raise SystemExit(ExitCode.USER_ERROR)
376
377 # Verify the class exists β€” load symbol map for validation and suggestions.
378 manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {}
379 cache = load_symbol_cache(root)
380 symbol_map = symbols_for_snapshot(root, manifest, cache=cache)
381
382 tree_for_file = symbol_map.get(file_path, {})
383 if class_addr not in tree_for_file:
384 cache.save()
385 _emit_not_found(class_addr, file_path, class_name, symbol_map, as_json, commit)
386 raise SystemExit(ExitCode.USER_ERROR)
387
388 all_methods = _class_methods(file_path, class_name, symbol_map)
389 if not all_methods:
390 cache.save()
391 print(f"⚠️ No methods found for '{class_addr}'.", file=sys.stderr)
392 raise SystemExit(ExitCode.USER_ERROR)
393
394 # Build reverse graph sharing the already-warm cache.
395 reverse = build_reverse_graph(root, manifest, cache=cache)
396 cache.save()
397
398 filtered = _filter_methods(all_methods, exclude_dunder, exclude_private)
399 covered, uncovered = _classify_methods(filtered, reverse, min_callers, exclude_self_file)
400
401 total = len(filtered)
402 n_covered = len(covered)
403 pct = round(n_covered / total * 100) if total else 0
404
405 # --compare diff
406 compare_commit: CommitRecord | None = None
407 newly_covered: list[str] = []
408 newly_uncovered: list[str] = []
409 pct_change: int = 0
410
411 if compare_ref:
412 compare_commit = resolve_commit_ref(root, repo_id, branch, compare_ref)
413 if compare_commit is None:
414 print(f"❌ --compare commit '{compare_ref}' not found.", file=sys.stderr)
415 raise SystemExit(ExitCode.USER_ERROR)
416 cmp_covered, cmp_uncovered, _ = _analyse_snapshot(
417 root, compare_commit, file_path, class_name,
418 exclude_dunder, exclude_private, min_callers, exclude_self_file,
419 )
420 cmp_covered_names = {name for _, name, _ in cmp_covered}
421 cur_covered_names = {name for _, name, _ in covered}
422 newly_covered = sorted(cur_covered_names - cmp_covered_names)
423 newly_uncovered = sorted(cmp_covered_names - cur_covered_names)
424 cmp_total = len(cmp_covered) + len(cmp_uncovered)
425 cmp_pct = round(len(cmp_covered) / cmp_total * 100) if cmp_total else 0
426 pct_change = pct - cmp_pct
427
428 # --count
429 if count_only and not as_json:
430 print(f"{n_covered}/{total}")
431 return
432
433 if as_json:
434 methods: list[_MethodJson] = [
435 _MethodJson(address=addr, name=name, called=True, callers=callers)
436 for addr, name, callers in covered
437 ] + [
438 _MethodJson(address=addr, name=name, called=False, callers=[])
439 for addr, name in uncovered
440 ]
441 payload = _CoveragePayload(
442 address=class_addr,
443 commit_id=commit.commit_id,
444 total_methods=total,
445 covered=n_covered,
446 percent=pct,
447 filters=_CoverageFilters(
448 exclude_dunder=exclude_dunder,
449 exclude_private=exclude_private,
450 min_callers=min_callers,
451 exclude_self=exclude_self,
452 ),
453 methods=methods,
454 )
455 if compare_commit is not None:
456 payload["compare_commit_id"] = compare_commit.commit_id
457 payload["newly_covered"] = newly_covered
458 payload["newly_uncovered"] = newly_uncovered
459 payload["percent_change"] = pct_change
460 print(json.dumps(payload, indent=2))
461 return
462
463 print(f"\nInterface coverage: {class_addr}")
464 active_filters: list[str] = []
465 if exclude_dunder:
466 active_filters.append("no dunders")
467 if exclude_private:
468 active_filters.append("no private")
469 if min_callers > 1:
470 active_filters.append(f"min {min_callers} callers")
471 if exclude_self:
472 active_filters.append("external callers only")
473 if active_filters:
474 print(f"Filters: {', '.join(active_filters)}")
475 print("─" * 62)
476
477 max_name = max(
478 (len(f"{class_name}.{name}") for _, name in filtered),
479 default=0,
480 )
481
482 for addr, bare_name, callers in covered:
483 display = f"{class_name}.{bare_name}"
484 line = f" βœ… {display:<{max_name}}"
485 if show_callers:
486 caller_str = ", ".join(callers[:3])
487 if len(callers) > 3:
488 caller_str += f" (+{len(callers) - 3} more)"
489 line += f" ← {caller_str}"
490 print(line)
491
492 for addr, bare_name in uncovered:
493 display = f"{class_name}.{bare_name}"
494 print(f" ❌ {display:<{max_name}} (no callers detected)")
495
496 print("\n" + "─" * 62)
497 print(f"Coverage: {n_covered}/{total} methods called ({pct}%)")
498
499 if pct == 100:
500 print("βœ… Full coverage β€” all methods are called at least once.")
501 elif pct >= 75:
502 print(f"🟒 Good coverage β€” {total - n_covered} uncovered method(s).")
503 elif pct >= 50:
504 print(f"🟑 Partial coverage β€” {total - n_covered} uncovered method(s) may be dead API surface.")
505 else:
506 print(f"πŸ”΄ Low coverage β€” {total - n_covered} of {total} methods have no detected callers.")
507
508 if compare_commit is not None:
509 sign = "+" if pct_change >= 0 else ""
510 print(f"\nCoverage diff vs {compare_commit.commit_id[:8]}: {sign}{pct_change}%")
511 if newly_covered:
512 print(f" Newly covered ({len(newly_covered)}): {', '.join(newly_covered)}")
513 if newly_uncovered:
514 print(f" Lost coverage ({len(newly_uncovered)}): {', '.join(newly_uncovered)}")
515
516 print(
517 "\nNote: dynamic dispatch, subclass overrides, and external callers are not detected."
518 )
519
520
521 # ---------------------------------------------------------------------------
522 # Error rendering
523 # ---------------------------------------------------------------------------
524
525
526 def _emit_not_found(
527 class_addr: str,
528 file_path: str,
529 class_name: str,
530 symbol_map: _SymbolMap,
531 as_json: bool,
532 commit: CommitRecord,
533 ) -> None:
534 """Print a helpful not-found error with suggestions."""
535 same_file, same_name, fuzzy = _find_class_suggestions(
536 class_addr, file_path, class_name, symbol_map
537 )
538
539 if as_json:
540 print(json.dumps({
541 "error": "symbol_not_found",
542 "address": class_addr,
543 "commit_id": commit.commit_id,
544 "suggestions": same_file[:8] or same_name[:5] or fuzzy[:5],
545 }, indent=2))
546 return
547
548 print(f"❌ '{class_addr}' not found in snapshot {commit.commit_id[:8]}.", file=sys.stderr)
549
550 if same_file:
551 print(f"\n Classes in {sanitize_display(file_path)}:", file=sys.stderr)
552 for c in same_file[:8]:
553 print(f" {c}", file=sys.stderr)
554 if same_name and same_name != same_file:
555 print(f"\n '{class_name}' found at:", file=sys.stderr)
556 for c in same_name[:5]:
557 print(f" {c}", file=sys.stderr)
558 if fuzzy and not same_file and not same_name:
559 print("\n Did you mean:", file=sys.stderr)
560 for c in fuzzy[:5]:
561 print(f" {c}", file=sys.stderr)
562 if not same_file and not same_name and not fuzzy:
563 print(
564 f"\n No classes found in {file_path}. "
565 "Check 'muse code symbols --json | jq' for valid addresses.",
566 file=sys.stderr,
567 )