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
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://<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


#: 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