ha-nvidia-shield/tests/ha_api_drift.py

"""Home-Assistant API-drift sweep (t37).

Why this exists
---------------
The unit suite runs our production modules against hand-written stubs
(``tests/entity_stublib.py`` plus the per-file stub fixtures), so a call site
that no longer matches the REAL Home Assistant API stays green. t36's sweep
found exactly that three times — two of them silent in production:

* **D1** ``from homeassistant.components.media_player import BrowseMediaSource``
  — media_player has not exported that name for years (it lives in
  ``homeassistant.components.media_source.models``), so every real browse
  raised ImportError;
* **D2** ``from homeassistant.util.diagnostic import async_redact_data`` — that
  module does not exist at all, and the swallowed ImportError turned the
  documented redaction pass into a no-op;
* **D3** ``self._async_handle_discovery_without_unique_id()`` — a coroutine
  ``ConfigFlow`` awaits in core; called without ``await`` its body never ran,
  so HA's discovery dedup was dead and Python warned about the coroutine.

What it does
------------
``scan()`` parses every production module and resolves each Home-Assistant
dependency against an INSTALLED homeassistant:

1. every ``homeassistant.*`` import — module present, attribute present (or a
   submodule of that name exists, which is what ``from x import submodule``
   really means);
2. every keyword argument passed to a known HA API (see ``HA_CALL_OWNERS``)
   must exist in the real signature — the ``options_updates`` class of bug;
3. every ``self.<method>()`` call inside a class that inherits from an HA class
   is compared against the real definition: a coroutine must be awaited, and an
   awaited call must actually be a coroutine.

Known limits (deliberate, and why the sweep still catches the real classes):
``**kwargs`` splats cannot be checked statically, calls on objects other than
``self`` (``self.hass.config_entries.…``) are only keyword-checked, and a
method our own class overrides is not resolved against HA. Those gaps are
visible in the ``Report`` counters, which the pytest wrapper asserts on so the
sweep cannot silently become vacuous.

Without homeassistant installed nothing can be resolved; the pytest wrapper
(``tests/test_ha_api_drift.py``) skips cleanly in that case, and CI runs one job
with homeassistant installed for exactly this check.
"""

from __future__ import annotations

import ast
import importlib
import inspect
from dataclasses import dataclass, field
from pathlib import Path

SHIELD_DIR = (
    Path(__file__).resolve().parent.parent / "custom_components" / "nvidia_shield"
)

#: Attribute name -> (module, class) that owns the REAL signature. Resolved with
#: getattr at scan time, so a renamed or dropped API is a finding, not a crash.
HA_CALL_OWNERS: dict[str, tuple[str, str]] = {
    "async_update_reload_and_abort": ("homeassistant.config_entries", "ConfigFlow"),
    "async_show_form": ("homeassistant.config_entries", "ConfigFlow"),
    "async_create_entry": ("homeassistant.config_entries", "ConfigFlow"),
    "async_abort": ("homeassistant.config_entries", "ConfigFlow"),
    "async_set_unique_id": ("homeassistant.config_entries", "ConfigFlow"),
    "_abort_if_unique_id_configured": (
        "homeassistant.config_entries",
        "ConfigFlow",
    ),
    "async_update_entry": ("homeassistant.config_entries", "ConfigEntries"),
    "async_schedule_reload": ("homeassistant.config_entries", "ConfigEntries"),
    "async_get_entry": ("homeassistant.config_entries", "ConfigEntries"),
    "async_forward_entry_setups": ("homeassistant.config_entries", "ConfigEntries"),
    "async_unload_platforms": ("homeassistant.config_entries", "ConfigEntries"),
    "async_add_executor_job": ("homeassistant.core", "HomeAssistant"),
}

_MISSING = object()


@dataclass(frozen=True)
class Finding:
    """One drifted API use."""

    path: str
    lineno: int
    kind: str
    detail: str

    def __str__(self) -> str:
        return f"{self.path}:{self.lineno}: {self.kind}: {self.detail}"


@dataclass
class Report:
    """Findings plus the counters that prove the sweep actually walked things."""

    findings: list[Finding] = field(default_factory=list)
    modules: int = 0
    imports: int = 0
    call_args: int = 0
    awaited_calls: int = 0

    def __str__(self) -> str:
        if not self.findings:
            return (
                "no Home-Assistant API drift "
                f"({self.modules} modules, {self.imports} imports, "
                f"{self.call_args} call arguments, "
                f"{self.awaited_calls} awaited HA calls)"
            )
        return "\n".join(str(finding) for finding in self.findings)


# --------------------------------------------------------------------- imports


def _check_import(module_name: str, attr: str | None) -> str | None:
    """Return a reason string when the import target is missing, else None."""
    try:
        module = importlib.import_module(module_name)
    except Exception as err:  # noqa: BLE001 - the error IS the finding
        return (
            f"module {module_name!r} is not importable "
            f"({type(err).__name__}: {err})"
        )

    if attr is None or hasattr(module, attr):
        return None

    # `from package import submodule` resolves through the import system even
    # when the attribute is not visible on the package yet.
    try:
        importlib.import_module(f"{module_name}.{attr}")
    except Exception:  # noqa: BLE001 - the error IS the finding
        return f"{module_name}.{attr} does not exist"
    return None


def _scan_imports(tree: ast.AST, path: str, report: Report) -> None:
    for node in ast.walk(tree):
        if isinstance(node, ast.ImportFrom):
            module_name = node.module or ""
            if not module_name.startswith("homeassistant"):
                continue
            for alias in node.names:
                report.imports += 1
                reason = _check_import(module_name, alias.name)
                if reason:
                    report.findings.append(
                        Finding(
                            path,
                            node.lineno,
                            "broken-import",
                            f"from {module_name} import {alias.name}: {reason}",
                        )
                    )
        elif isinstance(node, ast.Import):
            for alias in node.names:
                if not alias.name.startswith("homeassistant"):
                    continue
                report.imports += 1
                reason = _check_import(alias.name, None)
                if reason:
                    report.findings.append(
                        Finding(
                            path,
                            node.lineno,
                            "broken-import",
                            f"import {alias.name}: {reason}",
                        )
                    )


# -------------------------------------------------------------------- keywords


def _real_signature(module_name: str, owner: str, attr: str):
    """(signature, reason) for an HA API; both None when uninspectable."""
    try:
        module = importlib.import_module(module_name)
    except Exception as err:  # noqa: BLE001 - the error IS the finding
        return None, f"{module_name} is not importable ({type(err).__name__})"

    owner_obj = getattr(module, owner, None)
    if owner_obj is None:
        return None, f"{module_name}.{owner} does not exist"

    api = getattr(owner_obj, attr, None)
    if api is None:
        return None, f"{module_name}.{owner}.{attr} does not exist"

    try:
        return inspect.signature(api), None
    except (TypeError, ValueError):  # pragma: no cover - C callables
        return None, None


def _scan_call_keywords(tree: ast.AST, path: str, report: Report) -> None:
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call) or not node.keywords:
            continue
        func = node.func
        if not isinstance(func, ast.Attribute):
            continue
        owner = HA_CALL_OWNERS.get(func.attr)
        if owner is None:
            continue

        signature, reason = _real_signature(*owner, func.attr)
        if reason:
            report.findings.append(
                Finding(path, node.lineno, "unknown-ha-api", reason)
            )
            continue
        if signature is None:  # pragma: no cover - uninspectable
            continue
        if any(
            param.kind is inspect.Parameter.VAR_KEYWORD
            for param in signature.parameters.values()
        ):
            continue

        for keyword in node.keywords:
            if keyword.arg is None:  # **splat: not statically checkable
                continue
            report.call_args += 1
            if keyword.arg not in signature.parameters:
                report.findings.append(
                    Finding(
                        path,
                        keyword.lineno,
                        "unknown-kwarg",
                        f"{func.attr}() has no {keyword.arg!r} parameter; "
                        f"real signature: {signature}",
                    )
                )


# ---------------------------------------------------------------- await/sync


def _module_imports(tree: ast.AST) -> dict[str, tuple[str, str | None]]:
    """Local name -> (module, attribute-or-None) for the module's HA imports."""
    imports: dict[str, tuple[str, str | None]] = {}
    for node in ast.walk(tree):
        if isinstance(node, ast.ImportFrom) and (node.module or "").startswith(
            "homeassistant"
        ):
            for alias in node.names:
                imports[alias.asname or alias.name] = (node.module, alias.name)
        elif isinstance(node, ast.Import):
            for alias in node.names:
                if alias.name.startswith("homeassistant"):
                    imports[alias.asname or alias.name.split(".")[0]] = (
                        alias.name,
                        None,
                    )
    return imports


def _resolve_base(expr: ast.AST, imports: dict[str, tuple[str, str | None]]):
    """Resolve a class base expression to (module, attribute) when it is HA's."""
    if isinstance(expr, ast.Attribute) and isinstance(expr.value, ast.Name):
        target = imports.get(expr.value.id)
        if target is None:
            return None
        module_name, attr = target
        if attr is None:
            # `import homeassistant.config_entries as config_entries`
            return module_name, expr.attr
        if module_name == "homeassistant":
            # `from homeassistant import config_entries` binds the SUBMODULE,
            # so `config_entries.ConfigFlow` is that module's attribute.
            return f"{module_name}.{attr}", expr.attr
        return None
    if isinstance(expr, ast.Name):
        target = imports.get(expr.id)
        if target is not None and target[1] is not None:
            return target
    return None


def _ha_bases(
    class_node: ast.ClassDef, imports: dict[str, tuple[str, str | None]]
) -> list[type]:
    bases: list[type] = []
    for expr in class_node.bases:
        target = _resolve_base(expr, imports)
        if target is None:
            continue
        try:
            candidate = getattr(importlib.import_module(target[0]), target[1])
        except Exception:  # noqa: BLE001 - unresolvable base, skip it
            continue
        if isinstance(candidate, type):
            bases.append(candidate)
    return bases


def _method_on_bases(bases: list[type], attr: str):
    for base in bases:
        found = getattr(base, attr, _MISSING)
        if found is not _MISSING:
            return found, base
    return None, None


class _AwaitVisitor(ast.NodeVisitor):
    """Flags awaited/not-awaited mismatches for `self.<ha method>()` calls."""

    def __init__(self, bases: list[type], path: str, report: Report) -> None:
        self._bases = bases
        self._path = path
        self._report = report
        self._in_await = 0

    def visit_Await(self, node: ast.Await) -> None:
        self._in_await += 1
        self.generic_visit(node)
        self._in_await -= 1

    def visit_Call(self, node: ast.Call) -> None:
        func = node.func
        if (
            isinstance(func, ast.Attribute)
            and isinstance(func.value, ast.Name)
            and func.value.id == "self"
        ):
            api, owner = _method_on_bases(self._bases, func.attr)
            if api is not None:
                is_coro = inspect.iscoroutinefunction(api)
                awaited = self._in_await > 0
                where = f"{owner.__module__}.{owner.__qualname__}"
                if is_coro and not awaited:
                    self._report.findings.append(
                        Finding(
                            self._path,
                            node.lineno,
                            "missing-await",
                            f"self.{func.attr}() is a coroutine on {where} — "
                            f"core awaits it; without `await` its body never runs",
                        )
                    )
                elif awaited and not is_coro:
                    self._report.findings.append(
                        Finding(
                            self._path,
                            node.lineno,
                            "awaits-sync-api",
                            f"self.{func.attr}() is not a coroutine on {where}, "
                            f"so awaiting it raises TypeError",
                        )
                    )
                if awaited:
                    self._report.awaited_calls += 1
        self.generic_visit(node)


def _scan_awaits(tree: ast.AST, path: str, report: Report) -> None:
    imports = _module_imports(tree)
    for node in ast.walk(tree):
        if not isinstance(node, ast.ClassDef):
            continue
        bases = _ha_bases(node, imports)
        if not bases:
            continue
        for statement in node.body:
            # visit() walks the statement itself: a flat ast.walk would lose the
            # `await` nesting and report every awaited call as un-awaited.
            _AwaitVisitor(bases, path, report).visit(statement)


# ----------------------------------------------------------------------- scan


def scan(paths=None) -> Report:
    """Resolve every Home-Assistant API use in the production tree."""
    if paths is None:
        paths = [SHIELD_DIR]
    elif isinstance(paths, (str, Path)):
        paths = [paths]

    report = Report()
    for root in paths:
        root = Path(root)
        files = sorted(root.glob("*.py")) if root.is_dir() else [root]
        for file in files:
            report.modules += 1
            tree = ast.parse(file.read_text(encoding="utf-8"), filename=str(file))
            name = file.name
            _scan_imports(tree, name, report)
            _scan_call_keywords(tree, name, report)
            _scan_awaits(tree, name, report)
    return report


if __name__ == "__main__":  # pragma: no cover - manual red-proof entry point
    import sys

    targets = sys.argv[1:] or None
    result = scan(targets)
    print(result)
    raise SystemExit(1 if result.findings else 0)