NestQuest/tests/conftest.py

"""Shared pytest fixtures for the NestQuest test suite.

pytest-homeassistant-custom-component could not be adopted: it pins
``pytest==9.0.3`` and ``voluptuous==0.13.1``, which conflicts with the dev
group's ``voluptuous>=0.14`` (used directly by the options flow's schema and
its tests), and its back-releases raise the floor to Python >=3.12 while this
project supports >=3.11.  The documented fallback is therefore used:
``pytest-asyncio`` plus a standard-shaped ``hass`` fixture below.

conftest.py is the single place where ``homeassistant`` modules are provided:
it first tries to import the real packages (importlib in a try/except) and
only when they are absent installs MagicMock stand-ins via ``sys.modules`` and
wires the parent-package attributes.  All of that happens here, before any
test module imports the integration, so no test module needs its own
``sys.modules`` setup.  It also provides the ``hass`` fixture plus shared
helpers used by the behavior tests (config-entry factory, listener registry).
"""

from __future__ import annotations

import asyncio
import importlib
import inspect
import sys
from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest

# ---------------------------------------------------------------------------
# Homeassistant module provisioning (mock-only fallback, real import first).
# ---------------------------------------------------------------------------
_HA_MODULES = (
    "homeassistant",
    "homeassistant.core",
    "homeassistant.config_entries",
    "homeassistant.data_entry_flow",
)

_HA_AVAILABLE: dict[str, bool] = {}
for _mod in _HA_MODULES:
    try:
        importlib.import_module(_mod)
        _HA_AVAILABLE[_mod] = True
    except ImportError:
        _HA_AVAILABLE[_mod] = False

if not all(_HA_AVAILABLE.values()):
    # Mock-only branch: install MagicMock stand-ins for the missing modules.
    for _mod in _HA_MODULES:
        if not _HA_AVAILABLE[_mod]:
            sys.modules.setdefault(_mod, MagicMock())

    # Wire parent package attributes so "from homeassistant import
    # config_entries" resolves to the mocked submodule, mirroring real HA's
    # package layout.
    for _parent, _child in (
        ("homeassistant", "config_entries"),
        ("homeassistant", "data_entry_flow"),
        ("homeassistant", "core"),
    ):
        if not _HA_AVAILABLE[_parent]:
            setattr(sys.modules[_parent], _child, sys.modules[f"{_parent}.{_child}"])


def _ha_mock(name: str):
    """Return a module object for a mocked homeassistant module."""
    return sys.modules[name]


def options_flow_base():
    """Return the active OptionsFlowWithConfigEntry class (real or mocked)."""
    return getattr(sys.modules["homeassistant.config_entries"], "OptionsFlowWithConfigEntry")


class _ConfigFlowBase:
    """Stand-in for homeassistant.config_entries.ConfigFlow.

    A superset covering both the config-flow and options-flow tests: it must
    satisfy every consumer of the mocked surface.
    """

    VERSION = 1

    def __init_subclass__(cls, domain=None, **kwargs):
        super().__init_subclass__(**kwargs)
        cls._domain = domain

    def __init__(self):
        self.context = {}

    def async_show_form(self, *, step_id, **kwargs):
        return {"type": "form", "step_id": step_id, **kwargs}

    def async_abort(self, *, reason, **kwargs):
        return {"type": "abort", "reason": reason}

    def async_create_entry(self, *, title, data, **kwargs):
        result = {"type": "create_entry", "title": title, "data": data}
        hass = getattr(self, "hass", None)
        manager = getattr(hass, "config_entries", None) if hass is not None else None
        add = getattr(manager, "async_add", None)
        if add is not None:
            add(SimpleNamespace(domain=self._domain, title=title, data=data))
        return result

    async def async_set_unique_id(self, unique_id, raise_on_progress=True):
        self.context["unique_id"] = unique_id
        return None

    def _async_current_entries(self):
        return self.hass.config_entries.async_entries(self._domain)

    def _async_in_progress(self):
        return []

    @staticmethod
    def async_get_options_flow(config_entry):
        raise NotImplementedError


class _OptionsFlowWithConfigEntryBase:
    """Stand-in mirroring HA 2024.6 config_entries.OptionsFlowWithConfigEntry."""

    def __init__(self, config_entry=None):
        self.config_entry = config_entry
        self.hass = None
        self.options = dict(config_entry.options) if config_entry is not None else {}

    def async_show_form(self, *, step_id, **kwargs):
        return {"type": "form", "step_id": step_id, **kwargs}

    def async_create_entry(self, *, title, data, **kwargs):
        return {"type": "create_entry", "title": title, "data": data}


def _callback_decorator(fn):
    """Mirror homeassistant.core.callback: flags the fn and returns it unchanged."""
    fn.__ha_callback__ = True
    return fn


if not all(_HA_AVAILABLE.values()):
    # The concrete 2024.6-shaped surface the integration is written against
    # (mirrors the shapes the tests previously installed per module).
    _config_entries_mock = _ha_mock("homeassistant.config_entries")
    _core_mock = _ha_mock("homeassistant.core")
    _data_entry_flow_mock = _ha_mock("homeassistant.data_entry_flow")

    _config_entries_mock.ConfigFlow = _ConfigFlowBase
    _config_entries_mock.OptionsFlowWithConfigEntry = _OptionsFlowWithConfigEntryBase
    _core_mock.callback = _callback_decorator
    _data_entry_flow_mock.RESULT_TYPE_FORM = "form"
    _data_entry_flow_mock.RESULT_TYPE_CREATE_ENTRY = "create_entry"
    _data_entry_flow_mock.RESULT_TYPE_ABORT = "abort"
    _data_entry_flow_mock.FlowResult = dict


# ---------------------------------------------------------------------------
# Shared helpers for the behavior tests.
# ---------------------------------------------------------------------------
class ListenerRegistry:
    """Models HA's update-listener bookkeeping: appends to a list per entry, with per-listener removers."""

    def __init__(self, hass=None):
        self._hass = hass
        self._listeners: dict[str, list[object]] = {}
        self.reloaded: list[str] = []
        self._loop = asyncio.new_event_loop()

    def add(self, entry_id: str, listener) -> object:
        def _remove():
            self._listeners.get(entry_id, []).remove(listener)

        self._listeners.setdefault(entry_id, []).append(listener)
        return _remove

    def dispatch_options_update(self, entry) -> None:
        """Synchronous dispatch for sync tests (drives a private event loop)."""
        for listener in list(self._listeners.get(entry.entry_id, [])):
            result = listener(self._hass, entry)
            if inspect.isawaitable(result):
                self._loop.run_until_complete(result)

    async def async_dispatch_options_update(self, entry) -> None:
        """Async dispatch for pytest-asyncio tests (awaits listeners in the running loop)."""
        for listener in list(self._listeners.get(entry.entry_id, [])):
            result = listener(self._hass, entry)
            if inspect.isawaitable(result):
                await result

    @property
    def size(self) -> int:
        return sum(len(listeners) for listeners in self._listeners.values())


def make_config_entry(
    entry_id: str = "test_entry",
    options: dict | None = None,
    data: dict | None = None,
):
    """Create a stub config entry shaped like HA's ConfigEntry for tests."""
    entry = SimpleNamespace(
        entry_id=entry_id,
        options=dict(options or {}),
        data=dict(data or {}),
    )
    entry.add_update_listener = None
    return entry


def wire_entry_to_registry(entry, registry: ListenerRegistry):
    """Attach an update-listener registration function to the entry."""
    entry.add_update_listener = lambda listener: registry.add(entry.entry_id, listener)
    return entry


def make_hass() -> tuple:
    """Create a hass stub plus its listener registry, mirroring HA's shape."""
    hass = SimpleNamespace(
        data={},
        config=MagicMock(),
        config_entries=SimpleNamespace(async_reload=None),
    )
    registry = ListenerRegistry(hass)

    async def _async_reload(entry_id: str) -> None:
        registry.reloaded.append(entry_id)

    hass.config_entries.async_reload = MagicMock(side_effect=_async_reload)
    hass.config_entries._entries = []
    hass.config_entries.async_add = (
        lambda entry: hass.config_entries._entries.append(entry)
    )

    def _async_entries(domain: str):
        return [e for e in hass.config_entries._entries if e.domain == domain]

    hass.config_entries.async_entries = _async_entries
    return hass, registry


@pytest.fixture
def hass():
    """Provide a standard-shaped hass fixture.

    Mirrors the surface of Home Assistant's ``hass`` fixture: a ``config``
    namespace, a ``config_entries`` manager wired to a listener registry, a
    ``data`` dict for integration state, and the running event ``loop``.
    """
    _hass, registry = make_hass()
    _hass.registry = registry
    _hass.loop = asyncio.new_event_loop()
    try:
        yield _hass
    finally:
        _hass.loop.close()


@pytest.fixture
def make_entry():
    """Provide the config-entry factory as a fixture."""

    def _factory(entry_id: str = "test_entry", options: dict | None = None, data: dict | None = None):
        return make_config_entry(entry_id=entry_id, options=options, data=data)

    return _factory


@pytest.fixture
def listener_registry():
    """Provide a fresh listener registry for manual wiring."""

    def _factory(hass=None):
        return ListenerRegistry(hass)

    return _factory


@pytest.fixture
def make_flow():
    """Provide a factory for a stubbed NestQuestConfigFlow bound to hass."""

    def _factory(hass=None, existing_entries=None, in_progress=None):
        from custom_components.nestquest import config_flow

        flow = config_flow.NestQuestConfigFlow()
        flow.hass = hass if hass is not None else make_hass()[0]
        for entry in existing_entries or []:
            if getattr(entry, "domain", None) == flow._domain:
                flow.hass.config_entries.async_add(entry)
        if in_progress:
            flow._async_in_progress = lambda: in_progress
        return flow

    return _factory


@pytest.fixture
def make_config_flow_entry():
    """Provide an entry stub carrying a domain, for current-entries checks."""

    def _factory(domain: str, entry_id: str = "existing"):
        return SimpleNamespace(domain=domain, entry_id=entry_id)

    return _factory