James Mackay
130dee4bc6
Two defects shipped with a green suite: button.py imported a helper that had been deleted (no harness ever loaded the file), and the drift sweep only judged `homeassistant.*`, so a relative import of a name that no longer exists was invisible to it. Both are now impossible to ship silently. * tests/test_module_coverage.py loads EVERY *.py file of the package, in a clean interpreter, under the stub harness — exhaustive by construction (a new platform is covered the moment the file exists; there is no list and no special case). It asserts the reported set equals the files on disk, so a silently skipped module fails too; * entity_stublib.load_entity_modules() now also loads button.py (the named gap); the sweep is what keeps that list from needing to be complete; * ha_api_drift gained pass 5: the package's own relative imports. Pure AST, no homeassistant required, with its own entry point `scan_package_imports()` for the fast suite — it checks the target file exists and defines every imported name (submodule imports still resolve), instead of trusting that the modules a harness happened to load are the ones that matter; * the file walk in both AST passes is recursive, and the non-vacuity floors moved with it (modules 10 → 20, relative_imports ≥ 100 — a shallow-glob regression trips the floor instead of looking green); * two permanent red proofs, both on a copy of the tree with the reverted `from .helpers import discover_adb_host` reference in button.py: the stub-harness sweep fails with "FAIL button.py", and the static pass fails with "button.py:1: broken-relative-import: … not defined in helpers.py". Manual runs confirm the engine's other direction on the real pre-fix tree (c588a7c: exit 1 with the t40 entity-contract finding) and exit 0 on HEAD. No Home-Assistant API drift (23 modules, 58 imports, 148 relative imports, 36 call arguments, 6 awaited HA calls). Verified: 334 passed / 2 skipped, 335 / 1 with androidtvremote2, ruff clean. Note (pre-existing, not introduced here): running a SUBSET of test files in an order the suite was not written for (e.g. test_init.py alone after test_standalone_setup.py) fails 3 import-step tests with "ModuleNotFoundError: No module named 'voluptuous'" — the per-file stub scoping assumes the suite's order. Verified identical on the pre-t41 tree (60d884b), so it is left alone here.
507 lines
18 KiB
Python
507 lines
18 KiB
Python
"""Home-Assistant object stubs sized to unit-test the entity modules (t10).
|
|
|
|
Import-only stubs: MediaPlayerEntity/RemoteEntity bases, the feature/state
|
|
enum values the integration actually uses, and the event helpers. Tests load
|
|
their entity module through the synthetic ``nsh`` package AFTER calling
|
|
:func:`install` (and the per-file fixture pops everything at teardown).
|
|
"""
|
|
|
|
import asyncio
|
|
import enum
|
|
import sys
|
|
import types
|
|
from collections.abc import Mapping
|
|
|
|
import voluptuous_mirror
|
|
|
|
ENTITIES_STUB_NAMES = (
|
|
"voluptuous",
|
|
"homeassistant",
|
|
"homeassistant.config_entries",
|
|
"homeassistant.exceptions",
|
|
"homeassistant.core",
|
|
"homeassistant.const",
|
|
"homeassistant.components",
|
|
"homeassistant.components.media_player",
|
|
"homeassistant.components.remote",
|
|
"homeassistant.components.button",
|
|
"homeassistant.helpers",
|
|
"homeassistant.helpers.entity_registry",
|
|
"homeassistant.helpers.device_registry",
|
|
"homeassistant.helpers.selector",
|
|
"homeassistant.helpers.event",
|
|
"homeassistant.helpers.redact",
|
|
"homeassistant.helpers.entity",
|
|
"homeassistant.helpers.entity_platform",
|
|
"nsh.helpers",
|
|
"nsh.power",
|
|
"nsh.jellyfin",
|
|
"nsh.media_player",
|
|
"nsh.diagnostics",
|
|
"nsh.remote",
|
|
)
|
|
|
|
|
|
class MediaPlayerState(enum.StrEnum):
|
|
OFF = "off"
|
|
ON = "on"
|
|
IDLE = "idle"
|
|
PLAYING = "playing"
|
|
PAUSED = "paused"
|
|
|
|
|
|
class MediaPlayerEntityFeature(enum.IntFlag):
|
|
PLAY = 1
|
|
PAUSE = 2
|
|
STOP = 4
|
|
VOLUME_MUTE = 8
|
|
VOLUME_STEP = 16
|
|
VOLUME_SET = 32
|
|
BROWSE_MEDIA = 4096
|
|
TURN_ON = 64
|
|
TURN_OFF = 128
|
|
PLAY_MEDIA = 256
|
|
NEXT_TRACK = 512
|
|
PREVIOUS_TRACK = 1024
|
|
|
|
|
|
class MediaClass:
|
|
"""homeassistant.components.media_player.MediaClass (values we use)."""
|
|
|
|
APP = "app"
|
|
DIRECTORY = "directory"
|
|
|
|
|
|
class MediaType:
|
|
"""homeassistant.components.media_player.MediaType (values we use)."""
|
|
|
|
APP = "app"
|
|
APPS = "apps"
|
|
|
|
|
|
class BrowseMedia:
|
|
"""The REAL media_player.BrowseMedia shape (t37 D1).
|
|
|
|
Keyword-only with the six arguments real HA requires, the optional ones it
|
|
accepts, a faithful ``as_dict`` and — crucially — NO ``domain`` parameter:
|
|
``BrowseMediaSource`` lives in
|
|
``homeassistant.components.media_source.models`` (its contract is the
|
|
``media_source://<domain>`` id) and is NOT importable from
|
|
``homeassistant.components.media_player`` at all. The old import and the
|
|
old ``domain=`` kwarg therefore fail IN-SUITE instead of hiding behind a
|
|
permissive stub.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
media_class,
|
|
media_content_id,
|
|
media_content_type,
|
|
title,
|
|
can_play,
|
|
can_expand,
|
|
children=None,
|
|
children_media_class=None,
|
|
thumbnail=None,
|
|
not_shown=0,
|
|
can_search=False,
|
|
search_media_classes=None,
|
|
):
|
|
self.media_class = media_class
|
|
self.media_content_id = media_content_id
|
|
self.media_content_type = media_content_type
|
|
self.title = title
|
|
self.can_play = can_play
|
|
self.can_expand = can_expand
|
|
self.children = children
|
|
self.children_media_class = children_media_class
|
|
self.thumbnail = thumbnail
|
|
self.not_shown = not_shown
|
|
self.can_search = can_search
|
|
self.search_media_classes = search_media_classes
|
|
|
|
def as_dict(self, *, parent=True):
|
|
"""Real payload conversion — what the frontend actually consumes."""
|
|
if self.children_media_class is None and self.children:
|
|
self.calculate_children_class()
|
|
response = {
|
|
"title": self.title,
|
|
"media_class": self.media_class,
|
|
"media_content_type": self.media_content_type,
|
|
"media_content_id": self.media_content_id,
|
|
"children_media_class": self.children_media_class,
|
|
"can_play": self.can_play,
|
|
"can_expand": self.can_expand,
|
|
"can_search": self.can_search,
|
|
"search_media_classes": self.search_media_classes,
|
|
"thumbnail": self.thumbnail,
|
|
}
|
|
if not parent:
|
|
return response
|
|
response["not_shown"] = self.not_shown
|
|
response["children"] = (
|
|
[child.as_dict(parent=False) for child in self.children]
|
|
if self.children
|
|
else []
|
|
)
|
|
return response
|
|
|
|
def calculate_children_class(self):
|
|
"""Real helper: the children's common class, else DIRECTORY."""
|
|
self.children_media_class = MediaClass.DIRECTORY
|
|
assert self.children is not None
|
|
proposed_class = self.children[0].media_class
|
|
if all(child.media_class == proposed_class for child in self.children):
|
|
self.children_media_class = proposed_class
|
|
|
|
|
|
class MediaEntityManager:
|
|
"""trivial stand-in for the HA platform internals we don't exercise."""
|
|
|
|
|
|
class MediaPlayerEntity:
|
|
"""Base entity: attributes only; state writing is a no-op."""
|
|
|
|
def async_write_ha_state(self) -> None:
|
|
return None
|
|
|
|
|
|
class RemoteEntity:
|
|
def async_write_ha_state(self) -> None:
|
|
return None
|
|
|
|
|
|
def install() -> None:
|
|
"""Register the stub modules (idempotent)."""
|
|
if "homeassistant.components.media_player" in sys.modules:
|
|
return
|
|
|
|
vol = types.ModuleType("voluptuous")
|
|
# t33 F-A: the faithful mirror keeps schema defaults visible — a bare-key
|
|
# stub cannot tell a schema WITH a default from one without.
|
|
voluptuous_mirror.install(vol)
|
|
sys.modules["voluptuous"] = vol
|
|
|
|
ha = types.ModuleType("homeassistant")
|
|
sys.modules["homeassistant"] = ha
|
|
|
|
config_entries = types.ModuleType("homeassistant.config_entries")
|
|
config_entries.ConfigEntry = type("ConfigEntry", (), {})
|
|
async def _stub_set_unique_id(self, unique_id):
|
|
# hub shape: record; a full flow test file scopes its own version
|
|
self._unique_id = unique_id
|
|
|
|
def _noop_abort(self, *, error=None, **kwargs):
|
|
return None
|
|
|
|
class _ConfigFlowStub:
|
|
"""ConfigFlow shape used by entity tests (t29 G3a API included).
|
|
|
|
Late loads under the hub stubs (e.g. test_init designs run AFTER the
|
|
other flow files) still get the base plumbing the flow modules rely
|
|
on — show/create/abort/unique-id — so the flow classes never import
|
|
against a bare base.
|
|
"""
|
|
|
|
async_set_unique_id = _stub_set_unique_id
|
|
_abort_if_unique_id_configured = _noop_abort
|
|
|
|
async def _async_handle_discovery_without_unique_id(self):
|
|
# t37 D3: the real helper is a COROUTINE and core awaits it — a
|
|
# sync stub here would turn the production `await` into a
|
|
# TypeError. The abort semantics live in
|
|
# tests/test_standalone_setup.py, whose stub owns entry/in-progress
|
|
# state; at hub scope there is nothing to abort on.
|
|
return None
|
|
|
|
def async_show_form(
|
|
self, *, step_id, data_schema=None, errors=None,
|
|
description_placeholders=None,
|
|
):
|
|
return {
|
|
"type": "form",
|
|
"step_id": step_id,
|
|
"errors": dict(errors or {}),
|
|
"data_schema": data_schema,
|
|
"description_placeholders": dict(description_placeholders or {}),
|
|
}
|
|
|
|
def async_create_entry(self, *, title, data, options=None):
|
|
return {"type": "create_entry", "title": title, "data": data}
|
|
|
|
def async_abort(self, *, reason):
|
|
return {"type": "abort", "reason": reason}
|
|
|
|
def _get_reauth_entry(self, entry=None):
|
|
if entry is not None:
|
|
return entry
|
|
return None # the caller's cached entry is used
|
|
|
|
def _get_reconfigure_entry(self, entry=None):
|
|
if entry is not None:
|
|
return entry
|
|
return None # the caller's cached entry is used
|
|
|
|
# NOTE: the real HA signature is async_update_reload_and_abort(entry, ...)
|
|
# which updates, schedules the reload and aborts; the stub records both.
|
|
|
|
def async_update_reload_and_abort(self, entry, **kwargs):
|
|
"""Real-HA semantics: update, schedule the reload, abort."""
|
|
updates = kwargs.get("options_updates") or kwargs.get("data_updates") or {}
|
|
for key, value in updates.items():
|
|
if key in ("options", "data", "unique_id"):
|
|
setattr(entry, key, value)
|
|
else:
|
|
getattr(entry, "options", {}).setdefault(key, value) # best effort
|
|
scheduler = getattr(self.hass, "config_entries", None)
|
|
if scheduler is not None and hasattr(
|
|
scheduler, "async_schedule_reload"
|
|
):
|
|
scheduler.async_schedule_reload(entry.entry_id)
|
|
return self.async_abort(reason="reauth_successful")
|
|
|
|
def _config_flow_init(self, *args, **kwargs):
|
|
self.context = getattr(self, "context", {}) or {}
|
|
self.hass = getattr(self, "hass", None)
|
|
|
|
_config_flow_base = type(
|
|
"ConfigFlow",
|
|
(_ConfigFlowStub,),
|
|
{
|
|
"__init_subclass__": classmethod(
|
|
lambda cls, domain=None, **kwargs: None
|
|
),
|
|
"__init__": _config_flow_init,
|
|
},
|
|
)
|
|
config_entries.ConfigFlow = _config_flow_base
|
|
class _OptionsFlowStub:
|
|
def __init__(self):
|
|
self.config_entry = None
|
|
|
|
def add_suggested_values_to_schema(self, schema, suggested):
|
|
# real HA merges hints; the stub records them on the schema dict
|
|
schema_dict = getattr(schema, "schema", None) or (
|
|
schema if isinstance(schema, dict) else {}
|
|
)
|
|
schema_dict.setdefault("__suggested__", dict(suggested))
|
|
return schema
|
|
|
|
class _OptionsFlowWithReload(_OptionsFlowStub):
|
|
"""Options flow that reloads after a successful save (real HA)."""
|
|
|
|
def async_create_entry(self, *, title, data, options=None):
|
|
entry_id = getattr(self.config_entry, "entry_id", None)
|
|
scheduler = getattr(getattr(self, "hass", None), "config_entries", None)
|
|
if entry_id is not None and scheduler is not None and hasattr(
|
|
scheduler, "async_schedule_reload"
|
|
):
|
|
scheduler.async_schedule_reload(entry_id)
|
|
return {"type": "create_entry", "title": title, "data": data}
|
|
|
|
config_entries.OptionsFlow = _OptionsFlowStub
|
|
config_entries.SOURCE_IMPORT = "import"
|
|
config_entries.OptionsFlowWithReload = _OptionsFlowWithReload
|
|
sys.modules["homeassistant.config_entries"] = config_entries
|
|
|
|
exceptions = types.ModuleType("homeassistant.exceptions")
|
|
|
|
class ServiceValidationErrorStubError(Exception):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args or (kwargs.get("translation_key"),))
|
|
self.translation_domain = kwargs.get("translation_domain")
|
|
self.translation_key = kwargs.get("translation_key")
|
|
|
|
exceptions.ServiceValidationError = ServiceValidationErrorStubError
|
|
sys.modules["homeassistant.exceptions"] = exceptions
|
|
|
|
core = types.ModuleType("homeassistant.core")
|
|
core.HomeAssistant = object
|
|
core.callback = lambda fn: fn
|
|
core.CALLBACK_TYPE = object
|
|
core.Event = object
|
|
sys.modules["homeassistant.core"] = core
|
|
|
|
ha_const = types.ModuleType("homeassistant.const")
|
|
ha_const.EVENT_HOMEASSISTANT_STOP = "homeassistant_stop"
|
|
ha_const.STATE_PLAYING = "playing"
|
|
ha_const.STATE_PAUSED = "paused"
|
|
ha_const.Platform = types.SimpleNamespace(
|
|
MEDIA_PLAYER="media_player",
|
|
SENSOR="sensor",
|
|
REMOTE="remote",
|
|
BUTTON="button",
|
|
)
|
|
sys.modules["homeassistant.const"] = ha_const
|
|
|
|
components = types.ModuleType("homeassistant.components")
|
|
components.__path__ = []
|
|
sys.modules["homeassistant.components"] = components
|
|
|
|
mp = types.ModuleType("homeassistant.components.media_player")
|
|
mp.MediaPlayerEntity = MediaPlayerEntity
|
|
mp.MediaPlayerDeviceClass = types.SimpleNamespace(TV="tv")
|
|
mp.MediaPlayerEntityFeature = MediaPlayerEntityFeature
|
|
mp.BrowseMedia = BrowseMedia
|
|
mp.MediaClass = MediaClass
|
|
mp.MediaType = MediaType
|
|
# NOTE (t37 D1): homeassistant.components.media_player exposes NO
|
|
# BrowseMediaSource — deliberately not stubbed, so the drifted import
|
|
# fails here exactly as it does in a real install.
|
|
mp.MediaPlayerState = MediaPlayerState
|
|
sys.modules["homeassistant.components.media_player"] = mp
|
|
|
|
remote = types.ModuleType("homeassistant.components.remote")
|
|
remote.RemoteEntity = RemoteEntity
|
|
sys.modules["homeassistant.components.remote"] = remote
|
|
|
|
button = types.ModuleType("homeassistant.components.button")
|
|
button.ButtonEntity = type("ButtonEntity", (), {})
|
|
sys.modules["homeassistant.components.button"] = button
|
|
sensor_platform = types.ModuleType("homeassistant.components.sensor")
|
|
sensor_platform.SensorEntity = type("SensorEntity", (), {})
|
|
sensor_platform.SensorDeviceClass = types.SimpleNamespace(APP="app")
|
|
sys.modules["homeassistant.components.sensor"] = sensor_platform
|
|
|
|
|
|
helpers = types.ModuleType("homeassistant.helpers")
|
|
entity_registry = types.ModuleType("homeassistant.helpers.entity_registry")
|
|
|
|
class _Registry:
|
|
entities: dict = {}
|
|
|
|
def async_get(self, entity_id):
|
|
return None
|
|
|
|
entity_registry.async_get = lambda hass: _Registry()
|
|
device_registry = types.ModuleType("homeassistant.helpers.device_registry")
|
|
|
|
class _DeviceRegistry:
|
|
def __init__(self):
|
|
self.devices: dict[str, object] = {}
|
|
|
|
def async_get(self, device_id):
|
|
return self.devices.get(device_id)
|
|
|
|
_device_registry_shared = _DeviceRegistry()
|
|
|
|
device_registry.async_get = lambda hass: _device_registry_shared
|
|
sys.modules["homeassistant.helpers.device_registry"] = device_registry
|
|
|
|
helpers.entity_registry = entity_registry
|
|
sys.modules["homeassistant.helpers.entity_registry"] = entity_registry
|
|
|
|
helpers.device_registry = device_registry
|
|
sys.modules["homeassistant.helpers.device_registry"] = device_registry
|
|
|
|
event = types.ModuleType("homeassistant.helpers.event")
|
|
event.async_call_later = lambda hass, delay, callback_fn: (lambda: None)
|
|
event.async_track_state_change_event = lambda *args, **kwargs: (lambda: None)
|
|
helpers.event = event
|
|
helpers.selector = types.SimpleNamespace(
|
|
EntitySelector=lambda *a, **k: None,
|
|
EntitySelectorConfig=lambda *a, **k: "",
|
|
SelectSelector=lambda *a, **k: None,
|
|
SelectSelectorConfig=lambda **k: "",
|
|
SelectSelectorMode=types.SimpleNamespace(DROPDOWN="dropdown"),
|
|
)
|
|
entity = types.ModuleType("homeassistant.helpers.entity")
|
|
entity.DeviceInfo = dict
|
|
entity.EntityCategory = types.SimpleNamespace(
|
|
CONFIG="config", DIAGNOSTIC="diagnostic"
|
|
)
|
|
helpers.entity = entity
|
|
entity_platform = types.ModuleType("homeassistant.helpers.entity_platform")
|
|
entity_platform.AddEntitiesCallback = object
|
|
helpers.entity_platform = entity_platform
|
|
|
|
# t37 D2: homeassistant.helpers.redact.async_redact_data — the REAL helper
|
|
# (diagnostics.py imports it at module scope; `homeassistant.util.diagnostic`
|
|
# never existed). Modelled faithfully: the same masking of keys found in
|
|
# ``to_redact``, the same nested mapping/list recursion, the same skip of
|
|
# None/empty-string values.
|
|
redact = types.ModuleType("homeassistant.helpers.redact")
|
|
redact.REDACTED = "**REDACTED**"
|
|
|
|
def _stub_async_redact_data(data, to_redact):
|
|
if not isinstance(data, (Mapping, list)):
|
|
return data
|
|
if isinstance(data, list):
|
|
return [_stub_async_redact_data(val, to_redact) for val in data]
|
|
redacted = {**data}
|
|
for key, value in redacted.items():
|
|
if value is None:
|
|
continue
|
|
if isinstance(value, str) and not value:
|
|
continue
|
|
if key in to_redact:
|
|
if isinstance(to_redact, Mapping):
|
|
redacted[key] = to_redact[key](value)
|
|
else:
|
|
redacted[key] = redact.REDACTED
|
|
elif isinstance(value, Mapping):
|
|
redacted[key] = _stub_async_redact_data(value, to_redact)
|
|
elif isinstance(value, list):
|
|
redacted[key] = [
|
|
_stub_async_redact_data(item, to_redact) for item in value
|
|
]
|
|
return redacted
|
|
|
|
redact.async_redact_data = _stub_async_redact_data
|
|
helpers.redact = redact
|
|
sys.modules["homeassistant.helpers.redact"] = redact
|
|
|
|
sys.modules["homeassistant.helpers.event"] = event
|
|
sys.modules["homeassistant.helpers.entity"] = entity
|
|
sys.modules["homeassistant.helpers.entity_platform"] = entity_platform
|
|
sys.modules["homeassistant.helpers"] = helpers
|
|
|
|
|
|
def load_entity_modules() -> dict:
|
|
"""Load the integration's HA-free-ish entity modules under nsh."""
|
|
install()
|
|
from conftest import _SHIELD_DIR, _load_module
|
|
|
|
for name in (
|
|
"helpers",
|
|
"power",
|
|
"jellyfin",
|
|
"sensor",
|
|
"media_player",
|
|
"remote",
|
|
"diagnostics",
|
|
# t41: button.py was the file nobody loaded — a deleted helper it still
|
|
# imported shipped silently. It is a platform like the others, so it
|
|
# belongs here; tests/test_module_coverage.py sweeps EVERY module, which
|
|
# is what keeps this list from having to be complete.
|
|
"button",
|
|
"init",
|
|
): # "init" maps the integration __init__.py (t29 G8 async_setup)
|
|
if name == "init" and "nsh.init" in sys.modules:
|
|
continue
|
|
if f"nsh.{name}" not in sys.modules:
|
|
file_name = "__init__.py" if name == "init" else f"{name}.py"
|
|
_load_module(f"nsh.{name}", _SHIELD_DIR / file_name)
|
|
return {
|
|
name: sys.modules[f"nsh.{name}"]
|
|
for name in (
|
|
"power",
|
|
"jellyfin",
|
|
"sensor",
|
|
"media_player",
|
|
"remote",
|
|
"diagnostics",
|
|
)
|
|
}
|
|
|
|
|
|
class FakeHass:
|
|
"""Sized to entity unit tests: background tasks + no-op call_later."""
|
|
|
|
def async_create_task(self, coro):
|
|
return asyncio.create_task(coro)
|
|
|
|
def async_call_later(self, delay, callback_fn):
|
|
return lambda: None
|