transcription/tests/test_orphan_sweep.py

"""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] = {
    "BenchmarkManifest": (
        "Benchmark manifest model in benchmarking.py with no current caller. "
        "Uncertain - follow-up: confirm whether the benchmarking entrypoint is still "
        "intended before removing."
    ),
    "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."
    ),
    "dispose_all_engines": (
        "Engine lifecycle helper in db/engine.py. Uncertain - follow-up: operational "
        "teardown utility with no runtime or test caller."
    ),
    "refresh_engine": (
        "Engine lifecycle helper in db/engine.py. Uncertain - follow-up: paired with "
        "dispose_all_engines and equally unreferenced."
    ),
    "summarize_error": (
        "Error-presentation helper in ui/components/error_presenter.py that no page or "
        "component calls. Uncertain - follow-up: superseded by the presenter's other "
        "entrypoints."
    ),
}


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 == []