"""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 SELECT_SOURCE = 2048 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://`` 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 #: Every face install() registers. The guard below checks the WHOLE set, not one #: sentinel: test files scope their stubs per file (snapshot + pop at teardown, the #: t7 discipline), so one file can leave `homeassistant.components.media_player` #: behind while another name it popped — e.g. `voluptuous` — is gone. A sentinel #: guard then made install() a no-op in the next file, and whichever module needed #: the missing face failed with ModuleNotFoundError depending on file order (t44). #: The faces install() provides, derived from the list every test file scopes on. #: The ``nsh.*`` entries there are the integration modules the harness loads, not #: stub faces, so they are filtered out here. _STUB_FACES = tuple( name for name in ENTITIES_STUB_NAMES if not name.startswith("nsh.") ) def install() -> None: """Register the stub modules (idempotent, and self-healing). Callers must be able to rely on this regardless of what a previous test file popped: a missing face rebuilds the whole set, so any subset or file order works. A complete set is left alone — a file that installed richer faces of its own (tests/test_config_flow_ux.py's voluptuous mirror, the discovery stub) keeps them. """ if all(name in sys.modules for name in _STUB_FACES): 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 HomeAssistantErrorStubError(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") class ServiceValidationErrorStubError(HomeAssistantErrorStubError): """Real HA: ServiceValidationError derives from HomeAssistantError (t44).""" # t44: media_player raises HomeAssistantError when a configured command script # fails, so both names have to exist — and with the real inheritance. exceptions.HomeAssistantError = HomeAssistantErrorStubError 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 # t44: config_flow imports `selector` from this package, and the per-file stub # installers register it as its own face. Installing it here too keeps # `_STUB_FACES` honest: install() then really does provide every face a test # file may have popped, which is what its guard checks. selector = types.ModuleType("homeassistant.helpers.selector") selector.EntitySelector = lambda *args, **kwargs: None selector.EntitySelectorConfig = lambda *args, **kwargs: "" selector.SelectSelector = lambda *args, **kwargs: None selector.SelectSelectorConfig = lambda **kwargs: "" selector.SelectSelectorMode = types.SimpleNamespace(DROPDOWN="dropdown") helpers.selector = selector sys.modules["homeassistant.helpers.selector"] = selector sys.modules["homeassistant.helpers"] = helpers #: Namespaces the stub harness owns. Anything under these prefixes that a test #: file creates is harness state, so a file's teardown must remove it: a module #: left behind keeps the stub objects it captured (an exception class, the redact #: helper the diagnostics module holds) after the next file rebuilt them, which is #: how a file-order-dependent failure looked (t44). _TRACKED_PREFIXES = ("homeassistant", "voluptuous", "nsh") def _tracked(name: str) -> bool: return name.split(".", 1)[0] in _TRACKED_PREFIXES def snapshot_modules() -> dict: """Every stub/module name live right now — for a leak-free teardown.""" return {name: module for name, module in sys.modules.items() if _tracked(name)} def restore_modules(snapshot: dict) -> None: """Undo everything the current test file did to those namespaces. Two halves, and the first is what a hand-written list of names kept getting wrong: drop every tracked name the file created (so the next file loads it fresh, against ITS stubs), then put back every object it replaced. """ for name in [name for name in sys.modules if _tracked(name) and name not in snapshot]: del sys.modules[name] sys.modules.update(snapshot) 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