Files
transcription/tests/test_orphan_sweep.py
T
2026-08-23 18:53:54 -05:00

197 lines
8.7 KiB
Python

"""Deterministic orphan sweep for the `transcription` package.
`.github/skills/python-code-reviewer/skill.md` requires every review to report
orphaned code as removed, retained-with-justification, or uncertain-follow-up. This
guard makes that sweep reproducible: it locks the current set of unreferenced public
definitions, so a newly stranded function fails the build instead of accumulating
silently, and deleting a known orphan requires deleting its entry here.
The sweep is intentionally conservative. It considers module-level public
definitions and public methods on public classes, and honours the dynamic-wiring
exceptions the skill calls out: framework route registration, string-based
entrypoint references, and use from `tests/` or `tools/`.
"""
from __future__ import annotations
import ast
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[1]
SOURCE_DIR = PROJECT_ROOT / "src" / "transcription"
# Every tree that may legitimately consume package API.
REFERENCE_ROOTS = (SOURCE_DIR, PROJECT_ROOT / "tests", PROJECT_ROOT / "tools")
# Decorators that hand a callable to a framework registry, making the definition
# reachable without any in-repo reference to its name.
REGISTRATION_DECORATOR_PREFIXES = ("router.", "app.", "ui.page")
# Confirmed orphans, retained by decision rather than by reference. Each entry needs a
# rationale. Removing the code means removing the entry; adding an entry means an
# explicit decision to keep unreferenced code.
KNOWN_ORPHANS: dict[str, str] = {
"DocumentService.is_document_type_referenced": (
"Read helper currently bypassed by callers in favor of direct delete guards. "
"Retained for now to preserve service API stability during cleanup."
),
"DocumentService.is_tag_referenced": (
"Read helper currently bypassed by callers in favor of direct delete guards. "
"Retained for now to preserve service API stability during cleanup."
),
"DocumentService.read_document": (
"Public CRUD read method currently unused by runtime routes, but retained as "
"part of the service API shape for downstream call sites."
),
"DocumentService.read_document_type": (
"Public CRUD read method currently unused by runtime routes, but retained as "
"part of the service API shape for downstream call sites."
),
"DocumentService.read_tag": (
"Public CRUD read method currently unused by runtime routes, but retained as "
"part of the service API shape for downstream call sites."
),
"JSONBCompat.load_dialect_impl": (
"SQLAlchemy type hook is invoked by ORM internals via subclass protocol rather "
"than direct in-repo references; keep this implementation method."
),
"JobService.delete_job": (
"Legacy delete method retained as a compatibility shim while guarded deletion "
"flows migrate fully to delete_job_with_guardrails."
),
"JobService.update_job": (
"Legacy update method retained for compatibility while route and workflow "
"callers continue converging on narrower state-transition APIs."
),
"PeopleService.is_person_role_referenced": (
"Reference-check helper currently not called by route workflows, but kept with "
"the service surface while role-management cleanup remains in progress."
),
"PeopleService.read_person": (
"Public CRUD read method currently unused by runtime routes, but retained as "
"part of the service API shape for downstream call sites."
),
"PeopleService.read_person_role": (
"Public CRUD read method currently unused by runtime routes, but retained as "
"part of the service API shape for downstream call sites."
),
"Settings.normalize_provider_models": (
"Pydantic model-level validator is executed by framework hooks using decorator "
"registration and therefore has no direct symbolic call site."
),
"Settings.validate_provider_models_input": (
"Pydantic field validator is executed by framework hooks using decorator "
"registration and therefore has no direct symbolic call site."
),
"SourceService.read_job_source_for_job": (
"Public read helper currently unused by runtime flows, but retained as part of "
"the SourceService API pending endpoint consolidation."
),
}
def _source_files() -> list[Path]:
return sorted(SOURCE_DIR.rglob("*.py"))
def _is_registered_with_framework(node: ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef) -> bool:
return any(ast.unparse(decorator).startswith(REGISTRATION_DECORATOR_PREFIXES) for decorator in node.decorator_list)
def _public_definitions() -> dict[str, str]:
"""Public definitions mapped to `path:line` (module-level + class methods)."""
definitions: dict[str, str] = {}
for path in _source_files():
tree = ast.parse(path.read_text(encoding="utf-8"))
for node in tree.body:
if not isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef):
continue
if node.name.startswith("_") or _is_registered_with_framework(node):
continue
definitions[node.name] = f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}"
if isinstance(node, ast.ClassDef):
for method in node.body:
if not isinstance(method, ast.FunctionDef | ast.AsyncFunctionDef):
continue
if method.name.startswith("_"):
continue
definitions[f"{node.name}.{method.name}"] = (
f"{path.relative_to(PROJECT_ROOT).as_posix()}:{method.lineno}"
)
return definitions
def _referenced_names() -> tuple[set[str], str]:
"""Names referenced anywhere, plus every string literal joined for dotted lookups."""
names: set[str] = set()
literals: list[str] = []
for root in REFERENCE_ROOTS:
for path in sorted(root.rglob("*.py")):
# This module names every known orphan in `KNOWN_ORPHANS`; counting those
# strings as references would make the allowlist self-satisfying.
if path == Path(__file__).resolve():
continue
tree = ast.parse(path.read_text(encoding="utf-8"))
for node in ast.walk(tree):
if isinstance(node, ast.Name):
names.add(node.id)
elif isinstance(node, ast.Attribute):
names.add(node.attr)
elif isinstance(node, ast.ImportFrom):
for alias in node.names:
names.add(alias.name)
names.add(alias.asname or alias.name)
elif isinstance(node, ast.Constant) and isinstance(node.value, str):
literals.append(node.value)
return names, "\n".join(literals)
def _orphans() -> dict[str, str]:
definitions = _public_definitions()
names, literal_blob = _referenced_names()
return {
name: location
for name, location in definitions.items()
# A definition is referenced if its name is used directly, or appears inside a
# string such as "transcription.__main__:create_cli_app".
if (
name not in names
and name not in literal_blob
and name.split(".")[-1] not in names
and name.split(".")[-1] not in literal_blob
)
}
def test_public_definitions_are_discovered():
"""Guard the guard: the sweep is meaningless if nothing is scanned."""
definitions = _public_definitions()
assert len(definitions) >= 420
assert "create_app" in definitions
def test_framework_registered_routes_are_exempt():
"""Route handlers are reachable via decorator registration, not by name."""
definitions = _public_definitions()
assert "healthz_route" not in definitions
assert "read_document_source_media" not in definitions
def test_no_unexpected_orphaned_definitions():
"""Check 10: no public definition becomes unreferenced without a recorded decision."""
unexpected = {name: location for name, location in _orphans().items() if name not in KNOWN_ORPHANS}
assert unexpected == {}
def test_known_orphans_are_still_orphaned():
"""Keep the allowlist honest: an entry that regained callers must be removed."""
current = set(_orphans())
stale = sorted(name for name in KNOWN_ORPHANS if name not in current)
assert stale == [], "these definitions are referenced again; drop them from KNOWN_ORPHANS"
def test_known_orphans_document_a_rationale():
"""An allowlist without reasons is just suppressed output."""
missing = sorted(name for name, reason in KNOWN_ORPHANS.items() if len(reason.strip()) < 40)
assert missing == []