From 824d6192a1e68f988f84474b4b661ac4bf1e65a4 Mon Sep 17 00:00:00 2001 From: James Mackay Date: Sat, 10 Oct 2026 21:20:28 +0100 Subject: [PATCH] =?UTF-8?q?test(ci):=20D1/D2/D3=20=E2=80=94=20repo-owned?= =?UTF-8?q?=20HA=20API-drift=20sweep=20with=20a=20CI=20job=20(t37)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The unit suite validates production modules against hand-written stubs, so a call site that no longer matches real Home Assistant stays green — that is how D1, D2 and D3 reached master. This adds the check that would have caught them. `tests/ha_api_drift.py` (stdlib-only, documented inline) parses every module of custom_components/nvidia_shield 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 package import submodule` means) — catches D1/D2 and any future move; 2. every keyword argument passed to a known HA API (HA_CALL_OWNERS: the ConfigFlow/ConfigEntries APIs we call) must exist in the real signature — catches the `options_updates` class of bug; 3. every `self.()` call in a class that inherits from an HA class is compared with the real definition: a coroutine must be awaited and an awaited call must be a coroutine — catches D3. `tests/test_ha_api_drift.py` runs it against the production tree, asserts the sweep is not vacuous (module/import/argument/await counters), and pins the red proof with the four pre-fix forms, plus the green proof for the fixed forms. It skips cleanly when homeassistant is absent, so the stdlib-only jobs are unaffected. `.gitea/workflows/test.yml` gains the `ha-api-drift` job — the one place Home Assistant is installed — guarded by an explicit Python >= 3.13 check (Home Assistant 2026.x requires it) and the same PEP 668 retry the test job uses. Red proof against the real pre-fix revisions (svn-style checkout of `git show HEAD:…` into a temp dir, and the pre-t35 config_flow.py for the kwarg): media_player.py:965: broken-import: from homeassistant.components.media_player import BrowseMediaSource: …BrowseMediaSource does not exist diagnostics.py:93: broken-import: from homeassistant.util.diagnostic import async_redact_data: module 'homeassistant.util.diagnostic' is not importable config_flow.py:425: missing-await: self._async_handle_discovery_without_unique_id() is a coroutine on homeassistant.config_entries.ConfigFlow config_flow.py:700: unknown-kwarg: async_update_reload_and_abort() has no 'options_updates' parameter (exit 1 on that tree, exit 0 on this one) Verified: variant 1 289 passed / 2 skipped; variant 2 (androidtvremote2) 290 passed / 1 skipped; ruff clean; `UV_PYTHON=3.13 uvx --with homeassistant --with pytest-asyncio pytest tests/test_ha_api_drift.py -q` → 4 passed (the bare form of that command needs the pin here only because this sandbox's default Python is 3.15 and pycares has no wheel for it / no CMake to build it; CI pins the interpreter explicitly). --- .gitea/workflows/test.yml | 43 +++- tests/ha_api_drift.py | 405 +++++++++++++++++++++++++++++++++++++ tests/test_ha_api_drift.py | 130 ++++++++++++ 3 files changed, 577 insertions(+), 1 deletion(-) create mode 100644 tests/ha_api_drift.py create mode 100644 tests/test_ha_api_drift.py diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml index b61e146..1ac6001 100644 --- a/.gitea/workflows/test.yml +++ b/.gitea/workflows/test.yml @@ -1,4 +1,4 @@ -# Quality gate: lint + unit tests. +# Quality gate: lint + unit tests + the HA API-drift sweep. # # The test suite is stdlib-only (parsers, intents, running-apps) and runs the # vendored ha_adb_common helpers without installing Home Assistant — see @@ -6,6 +6,11 @@ # volume/TYPED-dict tests execute (tests/test_protocol_transport.py). Real # firmware captures live in tests/fixtures # (how to capture more: docs/RESEARCH_SHIELD.md §8). +# +# The `ha-api-drift` job is the ONE place Home Assistant is installed: every +# other job validates our code against hand-written stubs, which stay green +# even when a call site no longer matches the real API. That is how three +# drifts reached master (t36 D1/D2/D3) — see tests/ha_api_drift.py. name: Test @@ -48,3 +53,39 @@ jobs: - name: Unit tests run: python3 -m pytest tests -q + + # Resolve every Home-Assistant API use in our production modules against a + # real install: imports, keyword arguments and await/sync mismatches. The + # sweep is repo-owned (tests/ha_api_drift.py) and skips cleanly when + # Home Assistant is absent, so it is inert in the job above and live here. + ha-api-drift: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Check the runner Python can host Home Assistant + run: | + set -euo pipefail + # Home Assistant 2026.x requires Python >= 3.13; a newer runner just + # works. Fail loudly with the fix instead of a confusing resolver + # error when the image is older. + if ! python3 -c 'import sys; sys.exit(0 if sys.version_info >= (3, 13) else 1)'; then + echo "Home Assistant needs Python >= 3.13 but this runner has $(python3 -V)." + echo "Add 'actions/setup-python@v5' with python-version '3.13' to this job." + exit 1 + fi + + - name: Install Home Assistant + test tooling + run: | + set -euo pipefail + # Same PEP 668 dance as the test job (see the note there). The + # version floor tracks what we validate against; a resolver failure + # here is a real signal (Python too old), not something to swallow. + PKGS="pytest~=9.1 pytest-asyncio~=1.4 'homeassistant>=2026.10'" + if ! eval python3 -m pip install --quiet $PKGS; then + echo "system Python is externally managed; retrying with --break-system-packages" + eval python3 -m pip install --quiet --break-system-packages $PKGS + fi + + - name: HA API-drift sweep + run: python3 -m pytest tests/test_ha_api_drift.py -q diff --git a/tests/ha_api_drift.py b/tests/ha_api_drift.py new file mode 100644 index 0000000..75aae24 --- /dev/null +++ b/tests/ha_api_drift.py @@ -0,0 +1,405 @@ +"""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. + +Known limits (deliberate, and why the sweep still catches the real classes): +``**kwargs`` splats cannot be checked statically, calls on objects other than +``self`` (``self.hass.config_entries.…``) are only keyword-checked, and a +method our own class overrides is not resolved against HA. Those gaps are +visible in the ``Report`` counters, which the pytest wrapper asserts on so the +sweep cannot silently become vacuous. + +Without homeassistant installed nothing can be resolved; the pytest wrapper +(``tests/test_ha_api_drift.py``) skips cleanly in that case, and CI runs one job +with homeassistant installed for exactly this check. +""" + +from __future__ import annotations + +import ast +import importlib +import inspect +from dataclasses import dataclass, field +from pathlib import Path + +SHIELD_DIR = ( + Path(__file__).resolve().parent.parent / "custom_components" / "nvidia_shield" +) + +#: Attribute name -> (module, class) that owns the REAL signature. Resolved with +#: getattr at scan time, so a renamed or dropped API is a finding, not a crash. +HA_CALL_OWNERS: dict[str, tuple[str, str]] = { + "async_update_reload_and_abort": ("homeassistant.config_entries", "ConfigFlow"), + "async_show_form": ("homeassistant.config_entries", "ConfigFlow"), + "async_create_entry": ("homeassistant.config_entries", "ConfigFlow"), + "async_abort": ("homeassistant.config_entries", "ConfigFlow"), + "async_set_unique_id": ("homeassistant.config_entries", "ConfigFlow"), + "_abort_if_unique_id_configured": ( + "homeassistant.config_entries", + "ConfigFlow", + ), + "async_update_entry": ("homeassistant.config_entries", "ConfigEntries"), + "async_schedule_reload": ("homeassistant.config_entries", "ConfigEntries"), + "async_get_entry": ("homeassistant.config_entries", "ConfigEntries"), + "async_forward_entry_setups": ("homeassistant.config_entries", "ConfigEntries"), + "async_unload_platforms": ("homeassistant.config_entries", "ConfigEntries"), + "async_add_executor_job": ("homeassistant.core", "HomeAssistant"), +} + +_MISSING = object() + + +@dataclass(frozen=True) +class Finding: + """One drifted API use.""" + + path: str + lineno: int + kind: str + detail: str + + def __str__(self) -> str: + return f"{self.path}:{self.lineno}: {self.kind}: {self.detail}" + + +@dataclass +class Report: + """Findings plus the counters that prove the sweep actually walked things.""" + + findings: list[Finding] = field(default_factory=list) + modules: int = 0 + imports: int = 0 + call_args: int = 0 + awaited_calls: int = 0 + + def __str__(self) -> str: + if not self.findings: + return ( + "no Home-Assistant API drift " + f"({self.modules} modules, {self.imports} imports, " + f"{self.call_args} call arguments, " + f"{self.awaited_calls} awaited HA calls)" + ) + return "\n".join(str(finding) for finding in self.findings) + + +# --------------------------------------------------------------------- imports + + +def _check_import(module_name: str, attr: str | None) -> str | None: + """Return a reason string when the import target is missing, else None.""" + try: + module = importlib.import_module(module_name) + except Exception as err: # noqa: BLE001 - the error IS the finding + return ( + f"module {module_name!r} is not importable " + f"({type(err).__name__}: {err})" + ) + + if attr is None or hasattr(module, attr): + return None + + # `from package import submodule` resolves through the import system even + # when the attribute is not visible on the package yet. + try: + importlib.import_module(f"{module_name}.{attr}") + except Exception: # noqa: BLE001 - the error IS the finding + return f"{module_name}.{attr} does not exist" + return None + + +def _scan_imports(tree: ast.AST, path: str, report: Report) -> None: + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + module_name = node.module or "" + if not module_name.startswith("homeassistant"): + continue + for alias in node.names: + report.imports += 1 + reason = _check_import(module_name, alias.name) + if reason: + report.findings.append( + Finding( + path, + node.lineno, + "broken-import", + f"from {module_name} import {alias.name}: {reason}", + ) + ) + elif isinstance(node, ast.Import): + for alias in node.names: + if not alias.name.startswith("homeassistant"): + continue + report.imports += 1 + reason = _check_import(alias.name, None) + if reason: + report.findings.append( + Finding( + path, + node.lineno, + "broken-import", + f"import {alias.name}: {reason}", + ) + ) + + +# -------------------------------------------------------------------- keywords + + +def _real_signature(module_name: str, owner: str, attr: str): + """(signature, reason) for an HA API; both None when uninspectable.""" + try: + module = importlib.import_module(module_name) + except Exception as err: # noqa: BLE001 - the error IS the finding + return None, f"{module_name} is not importable ({type(err).__name__})" + + owner_obj = getattr(module, owner, None) + if owner_obj is None: + return None, f"{module_name}.{owner} does not exist" + + api = getattr(owner_obj, attr, None) + if api is None: + return None, f"{module_name}.{owner}.{attr} does not exist" + + try: + return inspect.signature(api), None + except (TypeError, ValueError): # pragma: no cover - C callables + return None, None + + +def _scan_call_keywords(tree: ast.AST, path: str, report: Report) -> None: + for node in ast.walk(tree): + if not isinstance(node, ast.Call) or not node.keywords: + continue + func = node.func + if not isinstance(func, ast.Attribute): + continue + owner = HA_CALL_OWNERS.get(func.attr) + if owner is None: + continue + + signature, reason = _real_signature(*owner, func.attr) + if reason: + report.findings.append( + Finding(path, node.lineno, "unknown-ha-api", reason) + ) + continue + if signature is None: # pragma: no cover - uninspectable + continue + if any( + param.kind is inspect.Parameter.VAR_KEYWORD + for param in signature.parameters.values() + ): + continue + + for keyword in node.keywords: + if keyword.arg is None: # **splat: not statically checkable + continue + report.call_args += 1 + if keyword.arg not in signature.parameters: + report.findings.append( + Finding( + path, + keyword.lineno, + "unknown-kwarg", + f"{func.attr}() has no {keyword.arg!r} parameter; " + f"real signature: {signature}", + ) + ) + + +# ---------------------------------------------------------------- await/sync + + +def _module_imports(tree: ast.AST) -> dict[str, tuple[str, str | None]]: + """Local name -> (module, attribute-or-None) for the module's HA imports.""" + imports: dict[str, tuple[str, str | None]] = {} + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and (node.module or "").startswith( + "homeassistant" + ): + for alias in node.names: + imports[alias.asname or alias.name] = (node.module, alias.name) + elif isinstance(node, ast.Import): + for alias in node.names: + if alias.name.startswith("homeassistant"): + imports[alias.asname or alias.name.split(".")[0]] = ( + alias.name, + None, + ) + return imports + + +def _resolve_base(expr: ast.AST, imports: dict[str, tuple[str, str | None]]): + """Resolve a class base expression to (module, attribute) when it is HA's.""" + if isinstance(expr, ast.Attribute) and isinstance(expr.value, ast.Name): + target = imports.get(expr.value.id) + if target is None: + return None + module_name, attr = target + if attr is None: + # `import homeassistant.config_entries as config_entries` + return module_name, expr.attr + if module_name == "homeassistant": + # `from homeassistant import config_entries` binds the SUBMODULE, + # so `config_entries.ConfigFlow` is that module's attribute. + return f"{module_name}.{attr}", expr.attr + return None + if isinstance(expr, ast.Name): + target = imports.get(expr.id) + if target is not None and target[1] is not None: + return target + return None + + +def _ha_bases( + class_node: ast.ClassDef, imports: dict[str, tuple[str, str | None]] +) -> list[type]: + bases: list[type] = [] + for expr in class_node.bases: + target = _resolve_base(expr, imports) + if target is None: + continue + try: + candidate = getattr(importlib.import_module(target[0]), target[1]) + except Exception: # noqa: BLE001 - unresolvable base, skip it + continue + if isinstance(candidate, type): + bases.append(candidate) + return bases + + +def _method_on_bases(bases: list[type], attr: str): + for base in bases: + found = getattr(base, attr, _MISSING) + if found is not _MISSING: + return found, base + return None, None + + +class _AwaitVisitor(ast.NodeVisitor): + """Flags awaited/not-awaited mismatches for `self.()` calls.""" + + def __init__(self, bases: list[type], path: str, report: Report) -> None: + self._bases = bases + self._path = path + self._report = report + self._in_await = 0 + + def visit_Await(self, node: ast.Await) -> None: + self._in_await += 1 + self.generic_visit(node) + self._in_await -= 1 + + def visit_Call(self, node: ast.Call) -> None: + func = node.func + if ( + isinstance(func, ast.Attribute) + and isinstance(func.value, ast.Name) + and func.value.id == "self" + ): + api, owner = _method_on_bases(self._bases, func.attr) + if api is not None: + is_coro = inspect.iscoroutinefunction(api) + awaited = self._in_await > 0 + where = f"{owner.__module__}.{owner.__qualname__}" + if is_coro and not awaited: + self._report.findings.append( + Finding( + self._path, + node.lineno, + "missing-await", + f"self.{func.attr}() is a coroutine on {where} — " + f"core awaits it; without `await` its body never runs", + ) + ) + elif awaited and not is_coro: + self._report.findings.append( + Finding( + self._path, + node.lineno, + "awaits-sync-api", + f"self.{func.attr}() is not a coroutine on {where}, " + f"so awaiting it raises TypeError", + ) + ) + if awaited: + self._report.awaited_calls += 1 + self.generic_visit(node) + + +def _scan_awaits(tree: ast.AST, path: str, report: Report) -> None: + imports = _module_imports(tree) + for node in ast.walk(tree): + if not isinstance(node, ast.ClassDef): + continue + bases = _ha_bases(node, imports) + if not bases: + continue + for statement in node.body: + # visit() walks the statement itself: a flat ast.walk would lose the + # `await` nesting and report every awaited call as un-awaited. + _AwaitVisitor(bases, path, report).visit(statement) + + +# ----------------------------------------------------------------------- scan + + +def scan(paths=None) -> Report: + """Resolve every Home-Assistant API use in the production tree.""" + if paths is None: + paths = [SHIELD_DIR] + elif isinstance(paths, (str, Path)): + paths = [paths] + + report = Report() + for root in paths: + root = Path(root) + files = sorted(root.glob("*.py")) if root.is_dir() else [root] + for file in files: + report.modules += 1 + tree = ast.parse(file.read_text(encoding="utf-8"), filename=str(file)) + name = file.name + _scan_imports(tree, name, report) + _scan_call_keywords(tree, name, report) + _scan_awaits(tree, name, report) + return report + + +if __name__ == "__main__": # pragma: no cover - manual red-proof entry point + import sys + + targets = sys.argv[1:] or None + result = scan(targets) + print(result) + raise SystemExit(1 if result.findings else 0) diff --git a/tests/test_ha_api_drift.py b/tests/test_ha_api_drift.py new file mode 100644 index 0000000..0552419 --- /dev/null +++ b/tests/test_ha_api_drift.py @@ -0,0 +1,130 @@ +"""Run the Home-Assistant API-drift sweep (t37). + +Skips cleanly when homeassistant is not installed — every other test file runs +against hand-written stubs, and this is the one check that must see the real +package. CI's `ha-api-drift` job installs it: + + uvx --with homeassistant --with pytest-asyncio pytest tests/test_ha_api_drift.py -q + +The sweep itself lives in ``tests/ha_api_drift.py`` (documented there). +""" + +from __future__ import annotations + +from pathlib import Path + +import ha_api_drift +import pytest + +pytest.importorskip( + "homeassistant", + reason="the API-drift sweep resolves our call sites against a real install", +) + +#: The three t36 drifts plus the `options_updates` kwarg t35 removed — the +#: sweep must flag every one of them, or it cannot protect us from the next. +PRE_FIX_SOURCES = { + "d1_browse_media_source.py": ( + "from homeassistant.components.media_player import BrowseMediaSource\n" + ), + "d2_util_diagnostic.py": ( + "from homeassistant.util.diagnostic import async_redact_data\n" + ), + "d3_unawaited_discovery.py": ( + "from homeassistant import config_entries\n" + "\n" + "\n" + "class PreFixFlow(config_entries.ConfigFlow, domain='nvidia_shield'):\n" + " async def async_step_zeroconf(self, discovery_info):\n" + " self._async_handle_discovery_without_unique_id()\n" + ), + "t35_options_updates.py": ( + "from homeassistant import config_entries\n" + "\n" + "\n" + "class PreFixReauthFlow(config_entries.ConfigFlow, domain='nvidia_shield'):\n" + " async def async_step_reauth_confirm(self, user_input=None):\n" + " self.async_update_reload_and_abort(\n" + " self.entry, options_updates={'protocol_enabled': True}\n" + " )\n" + " return self.async_abort(reason='reauth_successful')\n" + ), +} + + +def _require_installed_homeassistant(): + """Skip when sys.modules holds a test stub rather than an install.""" + import sys + + module = sys.modules["homeassistant"] + path = getattr(module, "__file__", None) + if path is None or not Path(path).exists(): + pytest.skip("homeassistant in sys.modules is a test stub, not an install") + + +def test_no_api_drift_in_the_production_tree(): + """The gate: every HA import, kwarg and await matches the real package.""" + _require_installed_homeassistant() + + report = ha_api_drift.scan() + + assert report.findings == [], "\n" + str(report) + + +def test_the_sweep_is_not_vacuous(): + """The counters prove the walk really covered the production tree.""" + _require_installed_homeassistant() + + report = ha_api_drift.scan() + + # Floors, not targets: they sit below today's walk (13 modules, 60 imports, + # 35 checked arguments, 7 awaited HA calls) so a broken walker shows up as a + # failure instead of a silently green sweep. + assert report.modules >= 10, report + assert report.imports >= 40, report + assert report.call_args >= 20, report + assert report.awaited_calls >= 5, report + + +def test_the_sweep_flags_the_pre_fix_defects(tmp_path): + """Red proof: the t36 drifts and the options_updates kwarg are all caught.""" + _require_installed_homeassistant() + + for name, source in PRE_FIX_SOURCES.items(): + (tmp_path / name).write_text(source, encoding="utf-8") + + report = ha_api_drift.scan(tmp_path) + + found = {(finding.path, finding.kind) for finding in report.findings} + assert found == { + ("d1_browse_media_source.py", "broken-import"), + ("d2_util_diagnostic.py", "broken-import"), + ("d3_unawaited_discovery.py", "missing-await"), + ("t35_options_updates.py", "unknown-kwarg"), + }, "\n" + str(report) + + +def test_the_sweep_accepts_the_fixed_forms(tmp_path): + """The same call sites written the way production writes them are clean.""" + _require_installed_homeassistant() + + (tmp_path / "fixed.py").write_text( + "from homeassistant.components.media_player import BrowseMedia\n" + "from homeassistant.helpers.redact import async_redact_data\n" + "from homeassistant import config_entries\n" + "\n" + "\n" + "class FixedFlow(config_entries.ConfigFlow, domain='nvidia_shield'):\n" + " async def async_step_zeroconf(self, discovery_info):\n" + " await self._async_handle_discovery_without_unique_id()\n" + "\n" + " def finish(self):\n" + " return self.async_update_reload_and_abort(\n" + " self.entry, options={'protocol_enabled': True}\n" + " )\n", + encoding="utf-8", + ) + + report = ha_api_drift.scan(tmp_path) + + assert report.findings == [], "\n" + str(report)