"""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.()`` 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"), # t14: THE TARGET-BEARING REGISTRATION. `ServiceRegistry.async_register` takes no # `target` (measured on 2026.10.1, and the reason the user's UI offered no Shield # to select): the frontend's picker is rendered from the service DESCRIPTION, # which `async_set_service_schema` publishes and which WINS over a services.yaml. # Both names are resolved against the INSTALLED HA at scan time, so an API that # is renamed or dropped is a finding here, not a log line on a user's install. "async_register": ("homeassistant.core", "ServiceRegistry"), # NOTE the owner is the MODULE, not a class: the description writer and the # device-expansion helper are module-level functions. That matters — the # device path first shipped as `registry.async_entries_for_device(...)`, which is # a method that does not exist; the real-HA probe caught it, and this entry makes # the API's existence a scanned fact instead of a runtime AttributeError. "async_set_service_schema": ("homeassistant.helpers", "service"), "async_entries_for_device": ("homeassistant.helpers", "entity_registry"), } #: 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 (26 modules, #: 77 imports, 282 relative imports, 65 checked arguments, 6 awaited HA calls, #: 1648 statement blocks) so a broken or mis-targeted walk fails loudly instead of #: printing zeros and exiting 0. #: #: THE MARGIN POLICY (t20/O3 — the t19 review asked for this in writing). A floor #: is a VACUITY-AND-BLINDNESS alarm, not a size ratchet: it must fail when the walk #: stops COVERING something (a broken glob, a regex that stopped matching, a module #: that stopped being parsed), and it must not fail because a legitimate change #: removed code. The margins follow from that, per counter: #: #: * counters where ONE unit is one API OWNER (modules, imports, call_args): 2. #: One owner may be refactored away; a whole CLASS of owners going unseen (the #: t62 F2 case, where deleting `async_show_menu` from HA_CALL_OWNERS left the #: sweep green) is what the 2 catches. This is the project's long-standing #: canary (t62 wrote it as "two below the count"), now stated once instead of #: being implied by each raise. #: * relative_imports: ~2.5%, currently 7. The count is dominated by package- #: relative imports that legitimately churn when files move; a collapse (an #: unparsed module, a stale glob) is orders of magnitude, not 7. #: * statement_blocks: ~2%, currently 28. This counter moves in BULK with every #: edit, which is exactly why a fixed small margin would be wrong here (the #: t19 complaint: at 1300 vs 1322 a 23-statement refactor tripped it). A 2% #: band leaves a normal refactor alone while a parse failure still shows up as #: a large drop. #: * awaited_calls: 1 (5 vs 6). Stated plainly: with six awaited HA calls a #: percentage margin is meaningless, so this floor is the weakest one here. It #: is still worth having — a walk that stopped visiting `await` sites at all #: trips it — but a deliberate removal of an awaited call MUST lower this floor #: in the same commit, and that is the reviewable act the gate exists for. #: #: The rule for changing any of them: lower a floor in the SAME commit that removes #: the code it counts, and never raise one without re-running the sweep first (a #: floor above the walk is a gate that can never pass). 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. # t17: raised 20 -> 24, just under the current 26. "modules": 24, # t15: raised from 40/100 to just under the current 76/206, because the remote # parity work is an IMPORTS-and-coupling feature: `RemoteEntityFeature`, # `ATTR_ACTIVITY`, the transport-state accessors and the shared app-directory # helpers all arrive as imports, and none of them may vanish unnoticed. # t17: raised 70 -> 75, just under the current 77 (the intents work imports # the scan/status vocabulary into the flow and the engine into the platform). "imports": 75, # t16: raised 190 -> 195 -> 235, again just under the walk. The intent # discovery module adds the package's exports to `ha_adb_common`'s surface and # is imported through it by the manager, so the relative-import count is exactly # where a new module must show up — a floor left behind would let the manager's # import of it disappear unnoticed. # t17: raised 235 -> 275, just under the current 282. The registered-intent # model, its statuses and its helpers move between modules, so this is again # exactly where a detached import must show up. "relative_imports": 275, # 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. # # t14: raised from 44 to just under the current 54, because the target-bearing # registration added two owners (`async_register`'s `schema`/`supports_response` # arguments and the module-level description writer). At 44 those owners could go # away without the gate noticing; at 52 they are load-bearing. # # t15: raised from 52 to just under the current 57 — the options-flow pairing # step is a real flow-display surface of its own (its own async_show_form with # step_id/data_schema/errors), so those call sites must not be able to vanish. # t17: raised 55 -> 63, just under the current 65 — the Intents screen adds # real flow-display call sites of its own (its menu, its three forms and the # scan's result step), none of which may vanish unnoticed. "call_args": 63, "awaited_calls": 5, # t62 F2: the unreachable-statement pass walks ~1300 statement blocks. A # floor here means a pass that silently stopped parsing cannot look green. # t15: raised 500 -> 1200 -> 1300, just under the walk, for the same reason the # imports floor moved: the exhaustion pass covers the new surfaces too. # t16: raised 1200 -> 1300 -> 1400: the parser, the cache and the manager's # scan path are new statement blocks, and the unreachable pass must keep # covering them. # t17: raised 1400 -> 1620, just under the current 1637. The screen, the seed # validator, the command builder and the button are new statement blocks, and # the unreachable pass is what keeps covering them — which is the point of the # floor: a pass that silently stopped parsing must not look green. # t20: the walk is 1648 after the protocol lifecycle fixes (the tracked # attempt, the cancel-and-await stop and the transition banner are new blocks), # so the floor is re-derived from the CURRENT walk at the 2% band above. "statement_blocks": 1615, } #: 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.()` 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.``) — 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( "", 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)