"""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:828``). 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. Both 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 either 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 (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] = { # 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, "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 relative_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.relative_imports} relative 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 _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:828``). 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 # ----------------------------------------------------------------------- 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: pure-AST checks, so they run even without homeassistant — that is # the whole point of a contract stubs cannot enforce (and a relative import # needs no library to be wrong). _scan_entity_contracts(paths, report) _scan_package_imports(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)