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")
MODULE_REFERENCE_ROOTS = (SOURCE_DIR, 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] = {
    "src/transcription/services/documents.py::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."
    ),
    "src/transcription/services/documents.py::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."
    ),
    "src/transcription/services/documents.py::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."
    ),
    "src/transcription/services/documents.py::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."
    ),
    "src/transcription/services/documents.py::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."
    ),
    "src/transcription/services/jobs.py::JobService.delete_job": (
        "Legacy delete method retained as a compatibility shim while guarded deletion "
        "flows migrate fully to delete_job_with_guardrails."
    ),
    "src/transcription/services/jobs.py::JobService.update_job": (
        "Legacy update method retained for compatibility while route and workflow "
        "callers continue converging on narrower state-transition APIs."
    ),
    "src/transcription/services/people.py::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."
    ),
    "src/transcription/services/people.py::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."
    ),
    "src/transcription/services/people.py::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."
    ),
    "src/transcription/config.py::Settings.normalize_provider_models": (
        "Pydantic model-level validator is executed by framework hooks using decorator "
        "registration and therefore has no direct symbolic call site."
    ),
    "src/transcription/config.py::Settings.validate_provider_models_input": (
        "Pydantic field validator is executed by framework hooks using decorator "
        "registration and therefore has no direct symbolic call site."
    ),
    "src/transcription/benchmarking.py": (
        "Retained as the evaluation-policy implementation for scoring preserved execution "
        "attempts, even though application runtime paths do not import it directly."
    ),
    "src/transcription/db/models.py::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."
    ),
    "src/transcription/services/sources.py::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."
    ),
    "src/transcription/ui/pages/tags_page.py": (
        "Retained temporarily as explicitly dead code until the planned route-retirement "
        "cleanup deletes the stranded module."
    ),
}


def _source_files() -> list[Path]:
    return sorted(SOURCE_DIR.rglob("*.py"))


def _entrypoint_modules() -> set[str]:
    return {
        "src/transcription/app.py",
        "src/transcription/__main__.py",
        "src/transcription/worker_service.py",
    }


def _module_key(path: Path) -> str:
    return path.relative_to(PROJECT_ROOT).as_posix()


def _leaf_definition_name(qualified_name: str) -> str:
    return qualified_name.split("::", 1)[-1].split(".")[-1]


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
            module = _module_key(path)
            definitions[f"{module}::{node.name}"] = f"{module}:{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"{module}::{node.name}.{method.name}"] = f"{module}:{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 _imported_source_modules() -> set[str]:
    imported: set[str] = set()

    def _mark_module_and_packages(module_path: Path) -> None:
        imported.add(_module_key(module_path))
        for parent in module_path.parents:
            package_init = parent / "__init__.py"
            if package_init.exists() and package_init.is_relative_to(SOURCE_DIR):
                imported.add(_module_key(package_init))

    for root in MODULE_REFERENCE_ROOTS:
        for path in sorted(root.rglob("*.py")):
            tree = ast.parse(path.read_text(encoding="utf-8"))
            for node in ast.walk(tree):
                if not isinstance(node, ast.ImportFrom) or not node.module:
                    continue
                if node.level > 0:
                    anchor = path.parent
                    for _ in range(node.level - 1):
                        anchor = anchor.parent
                    target = anchor / Path(*node.module.split("."))
                    file_candidate = target.with_suffix(".py")
                    package_candidate = target / "__init__.py"
                    if file_candidate.exists():
                        _mark_module_and_packages(file_candidate)
                    elif package_candidate.exists():
                        _mark_module_and_packages(package_candidate)
                    continue
                module_path = Path(*node.module.split("."))
                if not module_path.parts or module_path.parts[0] != "transcription":
                    continue
                target = SOURCE_DIR / Path(*module_path.parts[1:])
                file_candidate = target.with_suffix(".py")
                package_candidate = target / "__init__.py"
                if file_candidate.exists():
                    _mark_module_and_packages(file_candidate)
                elif package_candidate.exists():
                    _mark_module_and_packages(package_candidate)
    return imported


def _orphan_modules() -> dict[str, str]:
    imported = _imported_source_modules()
    entrypoints = _entrypoint_modules()
    return {
        module: module
        for path in _source_files()
        if (module := _module_key(path)) not in imported and module not in entrypoints
    }


def _orphans() -> dict[str, str]:
    definitions = _public_definitions()
    names, literal_blob = _referenced_names()
    orphan_modules = set(_orphan_modules())
    return {
        name: location
        for name, location in definitions.items()
        if name.split("::", 1)[0] not in orphan_modules
        # 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 literal_blob
            and name.split("::", 1)[-1] not in literal_blob
            and _leaf_definition_name(name) not in names
            and _leaf_definition_name(name) 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 "src/transcription/app.py::create_app" in definitions
    assert "src/transcription/services/documents.py::DocumentService.create_document" 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_no_unexpected_orphaned_modules():
    """Dead modules must be removed or recorded explicitly with a rationale."""
    unexpected = {name: location for name, location in _orphan_modules().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()) | set(_orphan_modules())
    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 == []