"""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 import ast import importlib import subprocess import sys import threading from pathlib import Path from types import SimpleNamespace 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" ), # t57 UNIT 0a: a wrong kwarg on a FLOW-DISPLAY API. `async_show_menu` takes # no `errors` in HA 2026.10.1, and the integration passed one — so every # ordinary setup path raised TypeError at the moment a user submitted the # first form, which is invisible to a stub suite whose stubs accept it. "t57_show_menu_errors.py": ( "from homeassistant import config_entries\n" "\n" "\n" "class PreFixSetupMenu(config_entries.ConfigFlow, domain='nvidia_shield'):\n" " async def async_step_user(self, user_input=None):\n" " return self.async_show_menu(\n" " step_id='menu',\n" " menu_options=['finish'],\n" " errors={\'base\': \'nope\'},\n" " )\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: the numbers live in ha_api_drift.FLOORS — one source # of truth for this wrapper and the standalone entry point (t39 F1) — so a # broken or mis-targeted walk fails here instead of looking green. violations = report.floor_violations() assert violations == [], "\n".join(violations) 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"), ("t57_show_menu_errors.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) def _run_drift_module(*paths): """Run the standalone entry point in this (HA-bearing) interpreter.""" return subprocess.run( [ sys.executable, str(Path(ha_api_drift.__file__).resolve()), *[str(path) for path in paths], ], capture_output=True, text=True, check=False, ) def test_the_standalone_entry_point_passes_on_the_production_tree(): """F1: the CLI is the sweep, with the same verdict on the real tree.""" result = _run_drift_module() assert result.returncode == 0, result.stderr assert "no Home-Assistant API drift" in result.stdout def test_the_standalone_entry_point_fails_on_an_empty_directory(tmp_path): """t39 F1: zeros must not exit 0 — the floors apply to the CLI too.""" result = _run_drift_module(tmp_path) assert result.returncode != 0 assert "0 modules" in result.stdout # the counter that used to look green assert "FLOOR:" in result.stderr assert "drift sweep FAILED" in result.stderr def test_the_standalone_entry_point_fails_on_a_missing_path(tmp_path): """A mis-specified path is a failure, not a silent zero-module sweep.""" result = _run_drift_module(tmp_path / "not-a-directory") assert result.returncode != 0 assert "bad-path" in result.stdout def test_no_floor_constants_are_duplicated_in_this_wrapper(): """F1: the floors come from ha_api_drift.FLOORS — nowhere else.""" source = Path(__file__).read_text(encoding="utf-8") tree = ast.parse(source) local_ints = [ target.id for node in tree.body if isinstance(node, ast.Assign) for target in node.targets if isinstance(target, ast.Name) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, int) ] assert local_ints == [], f"floor-like constants belong in ha_api_drift: {local_ints}" assert "floor_violations" in source def test_the_declared_python_floor_never_lags_homeassistant(): """F2: the CI guard's floor is HA's own Requires-Python, not a guess.""" _require_installed_homeassistant() import importlib.metadata as metadata declared = metadata.metadata("homeassistant")["Requires-Python"] parsed = ha_api_drift.parse_requires_python(declared) assert parsed is not None, f"unparsable Requires-Python: {declared!r}" assert tuple(parsed) <= ha_api_drift.HOMEASSISTANT_PYTHON_FLOOR, ( f"tests/ha_api_drift.py declares {ha_api_drift.HOMEASSISTANT_PYTHON_FLOOR} " f"but the installed homeassistant requires {declared} — raise the constant " f"and the setup-python version in .gitea/workflows/test.yml" ) # ------------------------------------------------- entity contract (t40, real HA) # # The stub suites cannot see this class of defect: their harness replaces # ``async_write_ha_state``, so a non-property ``extra_state_attributes`` looks # fine there. Only HA's own ``Entity.__async_calculate_state`` — which does # ``attr |= self.extra_state_attributes`` — turns it into # ``TypeError: 'method' object is not iterable`` and stops the entity from ever # rendering. These tests drive that real path. _REPO_ROOT = Path(__file__).resolve().parent.parent class _RecordingStates: """Minimal ``hass.states``: keeps what HA's entity writes.""" def __init__(self): self.registry: dict = {} self.written: list = [] def get(self, entity_id): return self.registry.get(entity_id) def async_set_internal(self, entity_id, state, attr, *args): self.written.append((entity_id, state, dict(attr))) def _real_ha_entity(module_name, class_name, *, options=None, entity_id=None): """Build a production entity under real homeassistant (no platform).""" if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) module = importlib.import_module(f"custom_components.nvidia_shield.{module_name}") states = _RecordingStates() hass = SimpleNamespace( loop_thread_id=threading.get_ident(), states=states, data={}, config=SimpleNamespace(components=set()), ) entry = SimpleNamespace( entry_id="e1", data={}, options=dict(options or {}), title="Shield" ) entity = getattr(module, class_name)(hass, entry) entity.entity_id = entity_id # HA's real translation-key paths (``unit_of_measurement``, device-class # names) need the platform's translation data; without it a translation-keyed # entity raises "cannot have a translation key ... before being added to the # entity platform". PlatformData is the real container, so the entity is # exercised with the same shape it sees in production. from homeassistant.helpers.entity_platform import PlatformData entity.platform_data = PlatformData( hass, domain=module_name, platform_name="nvidia_shield" ) # Not added through an EntityComponent: HA would warn once via # _suggest_report_issue (which needs hass.data["integrations"]) — set the # flag it uses to report that once instead of faking a whole platform. entity._no_platform_reported = True return entity, states def _write_ha_state(entity, states): """Run HA's real state write and return what reached the state machine.""" entity.async_write_ha_state() assert states.written, f"{type(entity).__name__} wrote no state" return states.written[-1] #: Every concrete entity class in the integration, exercised above against the #: REAL state machine. t40: the media player's defect was a state-write crash, so #: "the whole tree renders" is the contract under test, not a single entity. _REAL_HA_MATRIX = ( ("media_player", "NvidiaShieldMediaPlayer", None, "media_player.shield"), ("sensor", "NvidiaShieldCurrentAppSensor", None, "sensor.shield_current_app"), ( "sensor", "NvidiaShieldNowPlayingSensor", {"jellyfin_entity": "media_player.jellyfin"}, "sensor.shield_now_playing", ), ( "sensor", "NvidiaShieldAppsSensor", {"adb_enabled": True}, "sensor.shield_installed_apps", ), ("remote", "NvidiaShieldRemoteEntity", None, "remote.shield"), ("button", "NvidiaShieldPairAdbButton", None, "button.shield_pair_adb"), ) #: Classified as an entity by the sweep, but an abstract intermediate base HA #: never instantiates — its concrete subclasses are in the matrix. _INTERMEDIATE_BASES = frozenset({"NvidiaShieldSensorBase"}) @pytest.mark.parametrize( ("module_name", "class_name", "options", "entity_id"), _REAL_HA_MATRIX ) def test_entities_calculate_state_under_real_homeassistant( module_name, class_name, options, entity_id ): """t40 blocker: HA's own state write must not raise for ANY entity. ``Entity.async_write_ha_state`` → ``__async_calculate_state`` merges the extra attributes with ``attr |= ...``; the pre-fix media player raised ``TypeError: 'method' object is not iterable`` here, so it could never render. """ _require_installed_homeassistant() entity, states = _real_ha_entity( module_name, class_name, options=options, entity_id=entity_id ) written_id, state, attrs = _write_ha_state(entity, states) assert written_id == entity_id assert isinstance(state, str) assert isinstance(attrs, dict) def test_the_media_player_extra_attributes_reach_the_state_machine(): """The property's payload must survive HA's merge, not merely not raise.""" _require_installed_homeassistant() entity, states = _real_ha_entity( "media_player", "NvidiaShieldMediaPlayer", entity_id="media_player.shield" ) _entity_id, _state, attrs = _write_ha_state(entity, states) assert {"app_id", "app_name"} <= set(attrs), sorted(attrs) def test_the_real_ha_matrix_covers_every_entity_class_in_the_tree(): """A new entity class must join the matrix above, or this test fails. The sweep classifies entities from the AST; this keeps the real-HA coverage in step with that classification instead of silently shrinking. """ _require_installed_homeassistant() classified: set[str] = set() for path in sorted(ha_api_drift.SHIELD_DIR.glob("*.py")): classified |= ha_api_drift.entity_class_names( ast.parse(path.read_text(encoding="utf-8")) ) covered = {class_name for _module, class_name, _options, _entity_id in _REAL_HA_MATRIX} uncovered = classified - covered - _INTERMEDIATE_BASES assert uncovered == set(), ( f"entity classes never exercised against real HA: {sorted(uncovered)} — " f"add them to _REAL_HA_MATRIX" ) # ------------------------------------------------- t62: dead code and the floor # # Both defects pinned below shipped while every other gate was green: # connection.py carried a duplicated copy of the label-drain body AFTER its # `return cached` (valid Python, dead, invisible to this sweep), and the # call-argument floor sat at 20 while the counts had moved to ~46, so deleting a # flow-display API owner from HA_CALL_OWNERS did not trip it. #: t62 F2: the shape that shipped — a bare docstring plus the whole drain left #: after `return cached`, inside the same method. Line 6 is the dead statement. UNREACHABLE_PRE_FIX = ( "class Sweep:\n" " def labels(self, requests):\n" " cached = {}\n" " return cached\n" "\n" ' """Start at most one label fetch per watchdog cycle."""\n' " if not requests:\n" " return\n" " requests.discard(min(requests))\n" ) def test_the_sweep_flags_the_pre_fix_unreachable_body(tmp_path): """t62 F2 red proof: the duplicated drain body after `return cached` is dead.""" (tmp_path / "dead_sweep.py").write_text(UNREACHABLE_PRE_FIX, encoding="utf-8") findings = ha_api_drift.scan_unreachable(tmp_path) assert [(finding.path, finding.kind) for finding in findings] == [ ("dead_sweep.py", "unreachable-statement") ], findings assert findings[0].lineno == 6 # the dead docstring, not the return def test_every_terminal_statement_kind_is_flagged(tmp_path): """`return`, `raise`, `continue` and `break` all end a block (t62 F2).""" (tmp_path / "terminals.py").write_text( "def a():\n" " return 1\n" " x = 1\n" "\n" "\n" "def b():\n" " raise SystemExit(1)\n" " y = 2\n" "\n" "\n" "def c(items):\n" " for item in items:\n" " continue\n" " z = 3\n" " while True:\n" " break\n" " w = 4\n", encoding="utf-8", ) findings = ha_api_drift.scan_unreachable(tmp_path) assert [finding.kind for finding in findings] == ["unreachable-statement"] * 4, findings def test_the_production_tree_has_no_unreachable_statements(): """The gate itself: dead code is a finding, so it cannot ship silently.""" assert ha_api_drift.scan_unreachable() == [] def test_the_unreachable_pass_walks_real_statement_blocks(): """The floor proves "no findings" is not a pass that stopped parsing.""" report = ha_api_drift.scan() assert report.statement_blocks >= ha_api_drift.FLOORS["statement_blocks"] def test_removing_a_flow_display_owner_trips_the_call_argument_floor(monkeypatch): """t62: the floor must notice an API OWNER going away, not just a bad walk.""" _require_installed_homeassistant() intact = ha_api_drift.scan() assert intact.floor_violations() == [] reduced = dict(ha_api_drift.HA_CALL_OWNERS) del reduced["async_show_menu"] # the t57 UNIT 0a owner: 6 checked arguments monkeypatch.setattr(ha_api_drift, "HA_CALL_OWNERS", reduced) fewer = ha_api_drift.scan() assert fewer.call_args < intact.call_args assert fewer.call_args < ha_api_drift.FLOORS["call_args"], ( f"removing async_show_menu left {fewer.call_args} checked arguments, " f"still at or above the floor of {ha_api_drift.FLOORS['call_args']} — " "the flow-display owners are not load-bearing" ) assert any("call_args" in line for line in fewer.floor_violations()), ( fewer.floor_violations() )