333 lines
11 KiB
Python
333 lines
11 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
|
|
|
|
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,
|
|
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.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", (), {})
|
|
class _ConfigFlowStub:
|
|
"""ConfigFlow shape used by flow tests (t29 G3a API included)."""
|
|
|
|
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.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
|
|
|
|
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
|
|
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",
|
|
"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", "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
|