#!/usr/bin/env python3 """Per-method RPC coverage gate. Walks every member of :class:`notebooklm.rpc.types.RPCMethod` and asserts that each one has **both**: 1. **A test reference** — at least one file under ``tests/`` (excluding the coverage gate itself, its allowlist record, and helper data files that merely enumerate enum members) mentions the enum member by its qualified name (e.g. ``RPCMethod.LIST_NOTEBOOKS``) OR by its raw RPC id string value (e.g. ``"wXbhsf"``). 2. **A cassette covering the RPC id** — at least one cassette YAML under ``tests/cassettes/`` contains the raw RPC id string in its body. This is a pure-text grep — the cassette format is YAML but we never parse it; any occurrence of the id within the file counts, because ``batchexecute`` URLs and request bodies include the id verbatim. Either failure prints a single line of the form:: MISSING: RPCMethod. (id=): and the script exits 1. Exit 0 on full coverage. Pre-existing-gap allowlist -------------------------- When this gate first landed, some methods (notably newer or write-rarely exercised ones) already lacked one or both forms of coverage. Forcing contributors to backfill cassettes for unrelated methods before they could ship anything new would have stalled the arc, so the gate accepts a :data:`PREEXISTING_GAPS` set listing the (RPCMethod-name) entries grandfathered in at landing time. The allowlist is intended as a **one-way ratchet**: * It **must not grow** when a new ``RPCMethod`` member is added — new methods must ship with at least one test reference and at least one cassette. * It **must shrink** when a maintainer backfills coverage for a grandfathered method; stale entries in :data:`PREEXISTING_GAPS` fail the gate. The script is intentionally a static check (pure text grep on the cassette files and on the contents of ``tests/``); it never runs pytest or imports anything that needs a network. That keeps it deterministic, fast, and safe to run as a CI gate on every PR. Usage:: uv run python tests/scripts/check_method_coverage.py Exit codes: 0 — every method (modulo :data:`PREEXISTING_GAPS`) is covered 1 — one or more methods are missing required coverage """ from __future__ import annotations import sys from pathlib import Path # Allow running the script from any CWD by putting the repo root on sys.path # so ``from notebooklm.rpc.types import RPCMethod`` resolves the same way the # test suite resolves it. _REPO_ROOT = Path(__file__).resolve().parents[2] _SRC = _REPO_ROOT / "src" for _path in (_SRC, _REPO_ROOT): if str(_path) not in sys.path: sys.path.insert(0, str(_path)) from notebooklm.rpc.types import RPCMethod # noqa: E402 TESTS_DIR = _REPO_ROOT / "tests" CASSETTES_DIR = TESTS_DIR / "cassettes" # This script's own path — exclude it from the "test reference" search so the # enum-walk loop doesn't trivially satisfy the test-reference check. _SELF = Path(__file__).resolve() # Files that mention every enum member structurally (the enum definition # itself, or helper scripts that iterate over ``RPCMethod`` by reflection) and # would therefore satisfy the test-reference check for everything. Excluding # them keeps the gate meaningful — we want a *real* test asserting behaviour, # not a structural enumeration. _TEST_REFERENCE_EXCLUDES: frozenset[Path] = frozenset( { _SELF, # Add other reflective enumerators here as they appear. } ) # Pre-existing gaps grandfathered in when this gate landed; new methods # must NOT be added here. See module docstring for the one-way-ratchet # policy. Each entry is the ``RPCMethod.`` member name (without the # ``RPCMethod.`` prefix). # Intentionally empty: every RPCMethod has full coverage (a test reference + # a cassette). The source-label RPCs' cassettes were recorded in this change, # so the temporary grandfather entries were removed. Record a cassette for any # new method rather than re-adding it here. PREEXISTING_GAPS: frozenset[str] = frozenset() def _iter_test_files() -> list[Path]: """Return every file under ``tests/`` to grep for enum references. We walk the directory tree once and snapshot it as a sorted list so repeated callers (and the per-method test-reference check) see a deterministic set. Cassettes live under ``tests/cassettes/`` but we intentionally exclude them — the cassette presence check covers those. """ files: list[Path] = [] for path in TESTS_DIR.rglob("*"): if not path.is_file(): continue # Skip cassettes — they're covered by the separate cassette check. if CASSETTES_DIR in path.parents or path == CASSETTES_DIR: continue # Skip compiled bytecode under ``__pycache__``: ``.pyc`` files inline # source-level string constants like ``"wXbhsf"`` as raw UTF-8 bytes, # which would let a stale bytecode file silently satisfy the test- # reference check after the source was deleted. Cheap to skip, avoids # the spurious-match class entirely. if "__pycache__" in path.parts: continue if path.resolve() in _TEST_REFERENCE_EXCLUDES: continue files.append(path) files.sort() return files def _iter_cassette_files() -> list[Path]: """Return every ``*.yaml`` cassette under ``tests/cassettes/`` (sorted). Sorted output keeps the gate deterministic when reporting failures and matches the style of the sister script ``check_cassettes_clean.py``. Returns an empty list cleanly when the directory is missing so the gate can run on fresh checkouts without exploding. """ if not CASSETTES_DIR.exists(): return [] return sorted(CASSETTES_DIR.glob("*.yaml")) def _file_contains(path: Path, needles: tuple[str, ...]) -> bool: """Return True iff ``path`` contains any of ``needles`` as raw text. Reads the file as bytes to avoid encoding surprises on large cassette files (some contain mojibake-survivable HTTP payloads) and matches the UTF-8 bytes of each needle. ``OSError`` on read is treated as "no match" rather than crashing the gate — a corrupted cassette would already trip ``check_cassettes_clean.py``. """ try: data = path.read_bytes() except OSError: return False return any(needle.encode("utf-8") in data for needle in needles) def _has_test_reference(method: RPCMethod, test_files: list[Path]) -> bool: """Return True iff any ``tests/`` file references the method by name or id. Accepts either the qualified enum reference (``RPCMethod.LIST_NOTEBOOKS``) or the raw RPC id string (``"wXbhsf"``) — the latter catches tests that assert on encoded request payloads without importing the enum. """ needles = (f"RPCMethod.{method.name}", method.value) return any(_file_contains(p, needles) for p in test_files) def _has_cassette_coverage(method: RPCMethod, cassette_files: list[Path]) -> bool: """Return True iff any cassette body contains the raw RPC id string.""" needles = (method.value,) return any(_file_contains(p, needles) for p in cassette_files) def main() -> int: """Run the gate and return the process exit code. No CLI flags today — the gate is a pure pass/fail static check. We sort enum members by name before iterating so the failure output is stable across runs/machines. (The sister script ``check_cassettes_clean.py`` accepts ``argv`` because it has ``--strict``/``--allowlist`` flags; this one has nothing to parse so we keep the signature flag-free rather than carrying an unused ``argv`` parameter.) """ test_files = _iter_test_files() cassette_files = _iter_cassette_files() misses: list[str] = [] unused_allowlist: list[str] = [] for method in sorted(RPCMethod, key=lambda m: m.name): has_test = _has_test_reference(method, test_files) has_cassette = _has_cassette_coverage(method, cassette_files) if method.name in PREEXISTING_GAPS: # Track entries that no longer need allowlisting so maintainers # are nudged to shrink the ratchet. if has_test and has_cassette: unused_allowlist.append(method.name) continue if has_test and has_cassette: continue problems: list[str] = [] if not has_test: problems.append( "missing test reference (add a test under tests/ that imports " f"RPCMethod.{method.name} or asserts on the raw id '{method.value}')" ) if not has_cassette: problems.append( "missing cassette coverage (record a cassette under " f"tests/cassettes/ whose body contains the RPC id '{method.value}')" ) misses.append( f"MISSING: RPCMethod.{method.name} (id={method.value}): {'; '.join(problems)}" ) for line in misses: print(line) if unused_allowlist: print( "STALE: PREEXISTING_GAPS entries now have full coverage and " "must be removed: " + ", ".join(sorted(unused_allowlist)), file=sys.stderr, ) total = len(RPCMethod) # ``EnumMeta`` defines ``__len__`` directly. grandfathered = len(PREEXISTING_GAPS) checked = total - grandfathered print( f"\nSummary: {total} RPCMethod members, " f"{grandfathered} grandfathered (PREEXISTING_GAPS), " f"{checked} actively enforced, " f"{len(misses)} missing coverage." ) return 1 if misses or unused_allowlist else 0 if __name__ == "__main__": raise SystemExit(main())