"""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 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.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: APP = "app" DIRECTORY = "directory" class BrowseMediaSource: """HA BrowseMedia shape used by the app directory tests.""" def __init__( self, *, title, media_class, media_content_type, media_content_id, can_play, can_expand, domain=None, # t30 G18 (2026-05-20 blog contract) children=None, thumbnail=None, ): self.title = title self.media_class = media_class self.media_content_type = media_content_type self.media_content_id = media_content_id self.can_play = can_play self.can_expand = can_expand self.domain = domain self.children = children or [] self.thumbnail = thumbnail 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") vol.Required = lambda key, default=None: key vol.Optional = lambda key, default=None: key vol.Schema = lambda schema: schema 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 def async_show_form(self, *, step_id, data_schema=None, errors=None): return { "type": "form", "step_id": step_id, "errors": dict(errors or {}), "data_schema": data_schema, } 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 # 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 = BrowseMediaSource # factory used as a constructor mp.BrowseMediaSource = BrowseMediaSource mp.MediaClass = MediaClass 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 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", "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