ha-nvidia-shield/tests/entity_stublib.py

"""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

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.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")
    # 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

        def _async_handle_discovery_without_unique_id(self):
            # t32 F2 (audited helper): recording no-op at hub scope
            pass

        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 = 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