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;
4. every class inheriting from an HA entity base has its HA-READ attributes
   (``extra_state_attributes``, ``available``, …) declared as properties — HA
   merges them with ``attr |= self.extra_state_attributes``, so a plain method
   raises ``TypeError: 'method' object is not iterable`` on every state write and
   the entity can never render (t40 live install: ``media_player.py``'s
   ``extra_state_attributes`` — a property today, so the number rotted). The
   stub suites call the method happily; this pass is pure AST and, unlike 1-3,
   needs no homeassistant at all;
5. the package's OWN relative imports (t41): ``from .helpers import X`` must
   resolve to a file of this package that defines ``X`` (a submodule counts).
   ``button.py`` imported a helper that a refactor had deleted, and because no
   harness loaded that file the ImportError waited for a user's log — pass 1
   cannot see it (it only judges ``homeassistant.*``). Also pure AST, with its
   own entry point ``scan_package_imports()`` for the fast suite.
6. unreachable statements (t62 F2): a statement after
   ``return``/``raise``/``continue``/``break`` in the SAME block can never run.
   ``connection.py`` carried a full duplicated copy of the label-drain body
   (docstring included) after its ``return cached`` — valid Python, dead, and
   invisible to every other pass here, so it shipped. Pure AST, with its own
   entry point ``scan_unreachable()``.

All three AST passes run before the homeassistant availability check, and the file
walk is recursive (``_iter_python_files``), so a platform or helper one
directory down cannot hide from any of them.

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.

The standalone entry point (``python tests/ha_api_drift.py [path …]``) exits
non-zero on findings *and* on counters below ``FLOORS`` — an empty or
mis-targeted directory is a failure, never a green sweep. ``FLOORS`` is the one
source of truth for both that entry point and the pytest wrapper, and the
interpreter floor Home Assistant itself declares lives here too
(``HOMEASSISTANT_PYTHON_FLOOR``), where CI's guard reads it instead of repeating
the number.

Behaviour note for users (t37 D3): MAC-less zeroconf discovery now follows Home
Assistant's stricter contract — the awaited
``_async_handle_discovery_without_unique_id`` aborts ``already_configured`` as
soon as ANY entry exists for this domain, so a second Shield that advertises no
MAC is added manually. See README "Discovery (zeroconf)".

Without homeassistant installed the sweep cannot resolve anything: ``scan()``
returns a single ``no-homeassistant`` finding, the counters stay at zero (so the
floors fail on purpose) and nothing is ever reported green. 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 importlib.util
import inspect
import re
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"),
    # t57 UNIT 0a: the FLOW-DISPLAY API surface. These are the calls that decide
    # what the user sees, and a wrong keyword here is not a style error — it is a
    # TypeError the moment a user submits a form, on every path through it.
    # `async_show_menu` takes NO `errors` (HA 2026.10.1: step_id, menu_options,
    # sort, description_placeholders), which is precisely the bug this guards.
    "async_show_menu": ("homeassistant.data_entry_flow", "FlowHandler"),
    "async_show_progress": ("homeassistant.data_entry_flow", "FlowHandler"),
    "async_show_progress_done": ("homeassistant.data_entry_flow", "FlowHandler"),
    "async_abort": ("homeassistant.config_entries", "ConfigFlow"),
    "async_create_entry": ("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"),
}

#: Non-vacuity floors — the ONE source of truth for the standalone entry point
#: and the pytest wrapper (t39 F1). They sit below the current walk (24 modules,
#: 58 imports, 179 relative imports, 46 checked arguments, 7 awaited HA calls,
#: 1102 statement blocks) so a broken or mis-targeted walk fails loudly instead of
#: printing zeros and exiting 0.
FLOORS: dict[str, int] = {
    # t41: the sweep is recursive, so the floor sits above the top-level-only
    # count (13) — a regression to the shallow glob trips it.
    "modules": 20,
    "imports": 40,
    "relative_imports": 100,
    # t62 F2: raised from 20 to just under the current 46. A floor's job is to
    # notice an API OWNER going away, not only a broken walk: at 20, deleting
    # `async_show_menu` from HA_CALL_OWNERS left 40 checked arguments and still
    # looked green — only the pre-fix fixture tripped it. At 44 every owner that
    # carries a real settlement argument is load-bearing (show_form, show_menu,
    # abort, create_entry, update_entry). The floor sits two below the count so a
    # one-call-site refactor does not redden the gate.
    "call_args": 44,
    "awaited_calls": 5,
    # t62 F2: the unreachable-statement pass walks ~1100 statement blocks. A
    # floor here means a pass that silently stopped parsing cannot look green.
    "statement_blocks": 500,
}

#: The interpreter floor Home Assistant itself declares: PyPI metadata for
#: 2026.10.1 is ``Requires-Python >=3.14.2``, so a 3.13 runner cannot even
#: install the version this sweep must run against. CI's ``ha-api-drift`` job
#: provisions a matching interpreter and guards with THIS constant, so the
#: number is written once; tests/test_ha_api_drift.py pins it against the
#: installed distribution's own metadata.
HOMEASSISTANT_PYTHON_FLOOR: tuple[int, int, int] = (3, 14, 2)

_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
    relative_imports: int = 0
    call_args: int = 0
    awaited_calls: int = 0
    statement_blocks: int = 0
    def floor_violations(self, floors: dict[str, int] | None = None) -> list[str]:
        """Counters below the non-vacuity floors — a sweep nobody should trust.

        The floors default to module-level ``FLOORS`` so the standalone entry
        point and the pytest wrapper cannot drift apart (t39 F1).
        """
        floors = FLOORS if floors is None else floors
        return [
            f"only {getattr(self, name)} {name} (floor: {floor})"
            for name, floor in floors.items()
            if getattr(self, name) < floor
        ]

    def __str__(self) -> str:
        if not self.findings:
            return (
                "no Home-Assistant API drift "
                f"({self.modules} modules, {self.imports} imports, "
                f"{self.relative_imports} relative imports, "
                f"{self.statement_blocks} statement blocks, "
                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 _iter_python_files(root: Path) -> list[Path]:
    """Every Python file under a target, recursively (t41).

    The sweep used to glob one level, so a platform or helper one directory down
    was invisible to both passes. Exhaustive-by-construction beats a list.
    """
    if not root.is_dir():
        return [root]
    return sorted(
        path for path in root.rglob("*.py") if "__pycache__" not in path.parts
    )


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}",
                        )
                    )


# ------------------------------------------------------- the package's own imports
#
# Pure AST, no Home Assistant required — the same contract as the entity pass, so
# the fast suite can prove it even in the androidtvremote2-less variant.


def _package_root(path: Path) -> Path:
    """The integration package a file belongs to.

    The topmost ancestor directory that still has an ``__init__.py``: that is
    ``custom_components/nvidia_shield`` for both its top-level modules and the
    vendored ``ha_adb_common`` subpackage, and it holds for a copied tree too
    (tests scan temp copies).
    """
    root = path.parent
    while (root.parent / "__init__.py").exists():
        root = root.parent
    return root


class _ModuleLevelNames(ast.NodeVisitor):
    """Names a module binds at import time.

    Functions, classes, assignments and imports — including the ones inside
    ``if``/``try`` guards (the optional androidtvremote2 import lives in one).
    Function and class BODIES are skipped: a name bound at call time is not
    importable, which is exactly the distinction this pass needs.
    """

    def __init__(self) -> None:
        self.names: set[str] = set()

    def visit_FunctionDef(self, node: ast.AST) -> None:
        self.names.add(node.name)

    def visit_AsyncFunctionDef(self, node: ast.AST) -> None:
        self.names.add(node.name)

    def visit_ClassDef(self, node: ast.AST) -> None:
        self.names.add(node.name)

    def visit_Import(self, node: ast.Import) -> None:
        for alias in node.names:
            self.names.add(alias.asname or alias.name.split(".")[0])

    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
        for alias in node.names:
            if alias.name != "*":
                self.names.add(alias.asname or alias.name)

    def _bind(self, target: ast.AST) -> None:
        if isinstance(target, ast.Name):
            self.names.add(target.id)
        elif isinstance(target, (ast.Tuple, ast.List)):
            for element in target.elts:
                self._bind(element)
        elif isinstance(target, ast.Starred):
            self._bind(target.value)

    def visit_Assign(self, node: ast.Assign) -> None:
        for target in node.targets:
            self._bind(target)

    def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
        self._bind(node.target)

    def visit_AugAssign(self, node: ast.AugAssign) -> None:
        self._bind(node.target)

    def visit_For(self, node: ast.For) -> None:
        self._bind(node.target)
        self.generic_visit(node)

    def visit_AsyncFor(self, node: ast.AsyncFor) -> None:
        self._bind(node.target)
        self.generic_visit(node)

    def visit_With(self, node: ast.With) -> None:
        for item in node.items:
            if item.optional_vars is not None:
                self._bind(item.optional_vars)
        self.generic_visit(node)

    def visit_AsyncWith(self, node: ast.AsyncWith) -> None:
        for item in node.items:
            if item.optional_vars is not None:
                self._bind(item.optional_vars)
        self.generic_visit(node)


def _module_level_names(tree: ast.AST) -> set[str]:
    visitor = _ModuleLevelNames()
    visitor.visit(tree)
    return visitor.names


def _resolve_relative_import(path: Path, node: ast.ImportFrom, root: Path) -> Path | None:
    """The file ``from ... import`` points at, or None when it leaves the package."""
    base = path.parent
    for _ in range(node.level - 1):
        base = base.parent
    if not base.is_relative_to(root):
        return None
    if not node.module:
        return base / "__init__.py"
    target = base.joinpath(*node.module.split("."))
    if target.is_dir():
        return target / "__init__.py"
    return target.with_suffix(".py")


def _scan_relative_imports(path: Path, tree: ast.AST, report: Report) -> None:
    """t41: resolve the integration's OWN imports, not only ``homeassistant.*``.

    ``_scan_imports`` proves the HA surface; it says nothing about
    ``from .helpers import discover_adb_host`` once that helper is deleted — the
    exact defect that shipped from ``button.py`` because no harness loaded the
    file. This pass is pure AST against the tree, so it needs no importable
    Home Assistant and cannot be masked by a missing stub: the target file must
    exist and must define every name imported from it (a submodule counts too).
    """
    root = _package_root(path)
    for node in ast.walk(tree):
        if not isinstance(node, ast.ImportFrom) or not node.level:
            continue  # absolute imports are _scan_imports' business
        spelling = f"{'.' * node.level}{node.module or ''}"
        target = _resolve_relative_import(path, node, root)
        if target is None:
            report.findings.append(
                Finding(
                    str(path),
                    node.lineno,
                    "broken-relative-import",
                    f"from {spelling} import ... escapes the package — the "
                    "integration may only import from itself",
                )
            )
            continue
        if not target.exists():
            report.findings.append(
                Finding(
                    str(path),
                    node.lineno,
                    "broken-relative-import",
                    f"from {spelling} import ...: no module at "
                    f"{target.relative_to(root)}",
                )
            )
            continue
        defined = _module_level_names(
            ast.parse(target.read_text(encoding="utf-8"), filename=str(target))
        )
        for alias in node.names:
            if alias.name == "*":
                continue
            report.relative_imports += 1
            if alias.name in defined:
                continue
            # `from .package import submodule` is valid without the name being
            # bound in __init__ — the import system resolves the submodule.
            sibling = target.parent / f"{alias.name}.py"
            sibling_package = target.parent / alias.name / "__init__.py"
            if sibling.exists() or sibling_package.exists():
                continue
            report.findings.append(
                Finding(
                    str(path),
                    node.lineno,
                    "broken-relative-import",
                    f"from {spelling} import {alias.name}: not defined in "
                    f"{target.relative_to(root)}",
                )
            )


def _scan_package_imports(paths, report: Report) -> None:
    """Walk the tree and resolve every one of the package's own imports."""
    for root in paths:
        root = Path(root)
        if not root.exists():
            continue  # scan() reports bad-path for the same target
        for file in _iter_python_files(root):
            try:
                tree = ast.parse(file.read_text(encoding="utf-8"), filename=str(file))
            except (OSError, SyntaxError):
                continue
            _scan_relative_imports(file, tree, report)


def scan_package_imports(paths=None) -> list[Finding]:
    """The relative-import pass on its own (t41) — the fast suite's entry point."""
    if paths is None:
        paths = [SHIELD_DIR]
    elif isinstance(paths, (str, Path)):
        paths = [paths]
    report = Report()
    _scan_package_imports(paths, report)
    return report.findings


# -------------------------------------------------------------------- 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)


# ------------------------------------------------------ floors and versioning


def parse_requires_python(spec: str | None) -> tuple[int, ...] | None:
    """Parse a ``>=X.Y[.Z]`` Requires-Python string into a version tuple.

    The CI interpreter guard and the metadata pin in the pytest wrapper both
    need the same reading of Home Assistant's own declaration.
    """
    if not spec:
        return None
    match = re.match(r"\s*>=\s*(\d+(?:\.\d+)*)", spec)
    if match is None:
        return None
    return tuple(int(part) for part in match.group(1).split("."))


# ----------------------------------------------------------- entity contracts

#: Attributes Home Assistant reads as ATTRIBUTES — it never calls them. A plain
#: method with one of these names is invisible to the stub suites (permissive
#: stubs, and the harness replaces ``async_write_ha_state``) and fatal under real
#: HA: ``Entity.__async_calculate_state`` merges ``extra_state_attributes`` with
#: ``attr |= self.extra_state_attributes``, which raises
#: ``TypeError: 'method' object is not iterable`` on every state write. That was
#: the t40 live-install blocker — ``media_player.py``'s
#: ``extra_state_attributes``, a property today (the old line number rotted).
ENTITY_ATTRIBUTES = frozenset(
    {
        "activity",
        "activity_list",
        "app_id",
        "app_name",
        "assumed_state",
        "attribution",
        "available",
        "capability_attributes",
        "current_activity",
        "device_class",
        "device_info",
        "entity_category",
        "entity_picture",
        "extra_state_attributes",
        "icon",
        "is_volume_muted",
        "last_reset",
        "media_album_name",
        "media_artist",
        "media_content_id",
        "media_content_type",
        "media_duration",
        "media_image_url",
        "media_position",
        "media_title",
        "name",
        "native_unit_of_measurement",
        "native_value",
        "options",
        "should_poll",
        "sound_mode",
        "source",
        "source_list",
        "state",
        "state_attributes",
        "supported_features",
        "translation_key",
        "unique_id",
        "unit_of_measurement",
        "volume_level",
    }
)

#: Entity base classes Home Assistant ships. A class counts as an entity when its
#: transitive bases include one of these, so our own intermediate bases
#: (``NvidiaShieldSensorBase``) are covered as well.
HA_ENTITY_BASES = frozenset(
    {
        "AlarmControlPanelEntity",
        "BinarySensorEntity",
        "ButtonEntity",
        "CalendarEntity",
        "CameraEntity",
        "ClimateEntity",
        "CoverEntity",
        "DeviceTrackerEntity",
        "Entity",
        "EventEntity",
        "FanEntity",
        "HumidifierEntity",
        "ImageEntity",
        "LawnMowerEntity",
        "LightEntity",
        "LockEntity",
        "MediaPlayerEntity",
        "NotifyEntity",
        "NumberEntity",
        "RemoteEntity",
        "SelectEntity",
        "SensorEntity",
        "SirenEntity",
        "SwitchEntity",
        "TextEntity",
        "TodoListEntity",
        "UpdateEntity",
        "VacuumEntity",
        "ValveEntity",
        "WaterHeaterEntity",
        "WeatherEntity",
    }
)


def _is_attribute_decorated(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
    """True when the method is declared ``@property`` / ``@cached_property``."""
    for decorator in node.decorator_list:
        name = decorator.id if isinstance(decorator, ast.Name) else getattr(decorator, "attr", "")
        if name in {"property", "cached_property"}:
            return True
    return False


def _class_bases(node: ast.ClassDef) -> set[str]:
    """The names in a class's base list (``**kwargs``-style bases are ignored)."""
    return {
        base.id if isinstance(base, ast.Name) else getattr(base, "attr", "")
        for base in node.bases
    }


def entity_class_names(tree: ast.Module) -> set[str]:
    """Names of the classes in one module that inherit from an HA entity base."""
    bases = {
        node.name: _class_bases(node) for node in ast.walk(tree) if isinstance(node, ast.ClassDef)
    }
    entities = {name for name, parents in bases.items() if parents & HA_ENTITY_BASES}
    # Transitive: our own intermediate bases (NvidiaShieldSensorBase → SensorEntity)
    changed = True
    while changed:
        changed = False
        for name, parents in bases.items():
            if name not in entities and parents & entities:
                entities.add(name)
                changed = True
    return entities


def _scan_entity_contracts(paths, report: Report) -> None:
    """t40: flag HA-read attributes declared as plain methods.

    The stub suites cannot see this class of defect — they call the method
    successfully, because the harnesses replace ``async_write_ha_state`` — and
    only real HA turns it into a TypeError. Pure AST, so it also runs when
    homeassistant is not installed.
    """
    for root in paths:
        root = Path(root)
        if not root.exists():
            continue  # scan() reports bad-path for the same target
        files = _iter_python_files(root)
        for file in files:
            try:
                tree = ast.parse(file.read_text(encoding="utf-8"), filename=str(file))
            except (OSError, SyntaxError):
                continue
            entity_classes = entity_class_names(tree)
            for node in ast.walk(tree):
                if not isinstance(node, ast.ClassDef) or node.name not in entity_classes:
                    continue
                for item in node.body:
                    if not isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
                        continue
                    if item.name not in ENTITY_ATTRIBUTES or _is_attribute_decorated(item):
                        continue
                    report.findings.append(
                        Finding(
                            str(file),
                            item.lineno,
                            "entity-attribute-is-a-method",
                            f"{node.name}.{item.name} is a plain method, but HA reads it as an "
                            "attribute (``attr |= self.<name>``) — declare it @property",
                        )
                    )


def scan_entity_contracts(paths=None) -> list[Finding]:
    """The entity-contract pass on its own (t40) — the fast suite's entry point."""
    if paths is None:
        paths = [SHIELD_DIR]
    elif isinstance(paths, (str, Path)):
        paths = [paths]
    report = Report()
    _scan_entity_contracts(paths, report)
    return report.findings

# ------------------------------------------------------------ unreachable code

#: Statements that end the block they sit in: nothing AFTER one of these, in the
#: SAME block, can ever run. ``return``/``raise`` end the block outright;
#: ``continue``/``break`` end this pass through the enclosing loop, which is the
#: same thing for the statements that follow them there.
_TERMINAL_STATEMENTS = (ast.Return, ast.Raise, ast.Continue, ast.Break)


def _statement_blocks(node: ast.AST):
    """Every list-of-statements a node owns (bodies, ``orelse``, ``finalbody``)."""
    for _field, value in ast.iter_fields(node):
        if not isinstance(value, list) or not value:
            continue
        if all(isinstance(item, ast.stmt) for item in value):
            yield value


def _scan_unreachable(tree: ast.AST, path: str, report: Report) -> None:
    """t62 F2: flag statements that sit after a terminal one in the same block.

    Per-block on purpose: a statement in a DIFFERENT block (an ``else``, an
    ``except`` handler, a nested function) is not dead, and flagging those would
    make this pass noise. What it does see is the shape that shipped twice — a
    duplicated body left after ``return cached`` inside the same function
    (connection.py's label drain) — which is valid Python, invisible to every
    other gate here, and simply never executed.
    """
    for node in ast.walk(tree):
        for block in _statement_blocks(node):
            report.statement_blocks += 1
            for index, statement in enumerate(block[:-1]):
                if not isinstance(statement, _TERMINAL_STATEMENTS):
                    continue
                dead = block[index + 1]
                report.findings.append(
                    Finding(
                        path,
                        dead.lineno,
                        "unreachable-statement",
                        f"{type(statement).__name__.lower()} at line "
                        f"{statement.lineno} ends this block, so this statement "
                        "can never run (duplicated or leftover code)",
                    )
                )
                break


def _scan_unreachable_tree(paths, report: Report) -> None:
    """Walk the tree and run the unreachable-statement pass over every file."""
    for root in paths:
        root = Path(root)
        if not root.exists():
            continue  # scan() reports bad-path for the same target
        for file in _iter_python_files(root):
            try:
                tree = ast.parse(file.read_text(encoding="utf-8"), filename=str(file))
            except (OSError, SyntaxError):
                continue
            _scan_unreachable(tree, file.name, report)


def scan_unreachable(paths=None) -> list[Finding]:
    """The unreachable-statement pass on its own (t62 F2) — the wrapper's proof."""
    if paths is None:
        paths = [SHIELD_DIR]
    elif isinstance(paths, (str, Path)):
        paths = [paths]
    report = Report()
    _scan_unreachable_tree(paths, report)
    return report.findings
# ----------------------------------------------------------------------- scan


def homeassistant_available() -> bool:
    """True when a real homeassistant is importable in this interpreter."""
    return importlib.util.find_spec("homeassistant") is not None


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()
    # t40/t41/t62: pure-AST checks, so they run even without homeassistant —
    # that is the whole point of a contract stubs cannot enforce (a relative
    # import needs no library to be wrong, and dead code needs no API to be
    # dead).
    _scan_entity_contracts(paths, report)
    _scan_package_imports(paths, report)
    _scan_unreachable_tree(paths, report)
    if not homeassistant_available():
        # Neither a false green nor a wall of phantom findings: with no package
        # to resolve against, every import would "fail", so the sweep reports
        # that it cannot judge and leaves the counters at zero — which trips the
        # floors below on purpose.
        report.findings.append(
            Finding(
                "<environment>",
                0,
                "no-homeassistant",
                "homeassistant is not installed in this interpreter, so nothing "
                "can be resolved; install it (uvx --with homeassistant --python "
                "3.14 ...) or run tests/test_ha_api_drift.py, which skips cleanly",
            )
        )
        return report

    for root in paths:
        root = Path(root)
        if not root.exists():
            report.findings.append(
                Finding(
                    str(root),
                    0,
                    "bad-path",
                    "path does not exist — nothing was scanned",
                )
            )
            continue
        files = _iter_python_files(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)
    violations = result.floor_violations()
    if result.findings or violations:
        for line in violations:
            print(f"FLOOR: {line}", file=sys.stderr)
        print(
            f"drift sweep FAILED ({len(result.findings)} finding(s), "
            f"{len(violations)} floor violation(s))",
            file=sys.stderr,
        )
        raise SystemExit(1)
    raise SystemExit(0)