James Mackay
c588a7c66b
t37 made MAC-less zeroconf discovery honour Home Assistant's own `_async_handle_discovery_without_unique_id` contract — the helper is a coroutine and its dedup (abort `already_configured` when ANY entry exists for this domain, abort `already_in_progress` when another flow runs) only happens when it is awaited. That is stricter than the accidental host-equality guard it replaced, so it changes what a user can expect: * with a MAC, discovery still aborts duplicates and updates a moved device's hosts; * without a MAC, a second (or third) Shield must be added MANUALLY — discovery will abort as "already configured". Someone with two Shields would otherwise read that abort as a bug. The README now has a "Discovery (zeroconf)" section under Configuration stating the rule and the manual route, and the drift sweep's docstring carries the same note next to the D3 entry, so the decision is discoverable from wherever you meet it. Verified: 289 passed, 2 skipped; ruff clean.
515 lines
19 KiB
Python
515 lines
19 KiB
Python
"""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.
|
|
|
|
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"),
|
|
"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"),
|
|
}
|
|
|
|
#: 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 (13 modules,
|
|
#: 60 imports, 35 checked arguments, 7 awaited HA calls) so a broken or
|
|
#: mis-targeted walk fails loudly instead of printing zeros and exiting 0.
|
|
FLOORS: dict[str, int] = {
|
|
"modules": 10,
|
|
"imports": 40,
|
|
"call_args": 20,
|
|
"awaited_calls": 5,
|
|
}
|
|
|
|
#: 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
|
|
call_args: int = 0
|
|
awaited_calls: 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.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)
|
|
|
|
|
|
# ------------------------------------------------------ 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("."))
|
|
|
|
|
|
# ----------------------------------------------------------------------- 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()
|
|
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 = 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)
|
|
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)
|