ha-nvidia-shield/tests/test_config_flow_pairing.py

"""F1 regression: the adb_pair submit path calls
ShieldConnectionManager.test_connect with its declared signature
(host, port, key_path) — nothing extra, arity exact.

config_flow.py imports Home Assistant; it is exercised here against small
module stubs (import-only; no HA behaviour needed on this path) with the
actual ADB handshake mocked on the manager.
"""

import sys
import types
from unittest.mock import AsyncMock

import entity_stublib
import pytest
from conftest import _SHIELD_DIR, _load_module
from nsh.const import (
    CONF_ADB_ENABLED,
    CONF_ADB_PORT,
    CONF_ADB_STORE_APP_INFO,
    CONF_HOST,
    CONF_JELLYFIN_ENTITY,
    CONF_POWER_ENTITY,
    CONF_RECEIVER_ENTITY,
    CONF_RECEIVER_INPUT,
)

_KEY_PATH = "/tmp/nsh-test-adbkey"


_STUB_NAMES = (
    "voluptuous",
    "homeassistant",
    "homeassistant.config_entries",
    "homeassistant.core",
    "homeassistant.helpers",
    "homeassistant.helpers.entity_registry",
    "homeassistant.helpers.device_registry",
    "homeassistant.helpers.selector",
    # Loaded execution runs beneath the stubs; never escape this file either.
    "nsh.helpers",
    "nsh.config_flow",
)


def _install_import_stubs() -> None:
    """Register import-time stand-ins for HA/voluptuous (idempotent)."""
    if sys.modules.get("voluptuous") is not None:
        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")

    class ConfigFlow:
        VERSION = 1

        def __init_subclass__(cls, **kwargs):
            kwargs.pop("domain", None)  # class statement passes domain=DOMAIN
            super().__init_subclass__(**kwargs)

        def async_show_form(self, *, step_id, data_schema=None, errors=None):
            return _form(step_id, errors or {})

        def async_show_menu(self, *, step_id=None, menu_options, sort=False,
                            description_placeholders=None):
            # t56: the real MENU result shape; the flow manager dispatches the
            # chosen `next_step_id` to `async_step_<choice>(None)`.
            return {
                "type": "menu",
                "step_id": step_id,
                "menu_options": list(menu_options),
                "description_placeholders": dict(
                    description_placeholders or {}
                ),
            }

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

    def _form(step_id, errors):
        return {"type": "form", "step_id": step_id, "errors": dict(errors)}

    config_entries.ConfigFlow = ConfigFlow
    class _BaseOptionsFlow:
        pass

    # t29 G3b: the current options-flow API surface on the local stub
    class _SuggestedMixin:
        def add_suggested_values_to_schema(self, schema, suggested):
            schema_dict = getattr(schema, "schema", None) or (
                schema if isinstance(schema, dict) else {}
            )
            schema_dict.setdefault("__suggested__", dict(suggested))
            return schema

    class OptionsFlowWithReload(_SuggestedMixin):
        def async_create_entry(self, *, title, data):
            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 = _BaseOptionsFlow
    config_entries.OptionsFlowWithReload = OptionsFlowWithReload
    sys.modules["homeassistant.config_entries"] = config_entries

    core = types.ModuleType("homeassistant.core")
    core.HomeAssistant = object
    core.callback = lambda fn: fn
    sys.modules["homeassistant.core"] = core

    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()
    helpers.entity_registry = entity_registry
    sys.modules["homeassistant.helpers.entity_registry"] = entity_registry
    sys.modules["homeassistant.helpers"] = helpers

    device_registry = types.ModuleType("homeassistant.helpers.device_registry")
    device_registry.async_get = lambda hass: types.SimpleNamespace(devices={})
    helpers.device_registry = device_registry
    sys.modules["homeassistant.helpers.device_registry"] = device_registry
    selector = types.ModuleType("homeassistant.helpers.selector")
    entity_stublib.install_selector_faces(selector)
    sys.modules["homeassistant.helpers.selector"] = selector


class _Config:
    config_dir = "/tmp/nsh-test-config"


class _FakeHass:
    config = _Config


    @staticmethod
    async def async_add_executor_job(fn, *args):
        return fn(*args)


def _config_flow_module():
    _install_import_stubs()
    if "nsh.helpers" not in sys.modules:
        _load_module("nsh.helpers", _SHIELD_DIR / "helpers.py")
    if "nsh.config_flow" not in sys.modules:
        _load_module("nsh.config_flow", _SHIELD_DIR / "config_flow.py")
    return sys.modules["nsh.config_flow"]


def _flow(user_input_host: str, monkeypatch=None):
    module = _config_flow_module()
    flow = module.NvidiaShieldConfigFlow()
    flow.hass = _FakeHass
    flow._user_data = {
        "name": "Nvidia Shield",
        CONF_ADB_ENABLED: True,
        CONF_HOST: user_input_host,
        CONF_ADB_PORT: 5555,
        CONF_JELLYFIN_ENTITY: "",
        CONF_RECEIVER_ENTITY: "",
        CONF_RECEIVER_INPUT: "",
        CONF_POWER_ENTITY: "",
        CONF_ADB_STORE_APP_INFO: True,
    }
    if monkeypatch is not None:
        # Patch module-level resolution — the call site is a plain global.
        monkeypatch.setattr(
            module, "ensure_adb_key", lambda _config_dir: _KEY_PATH
        )
    return flow


@pytest.fixture(autouse=True)
def scoped_stubs():
    """Contain the import stubs to this module's tests (t7 INFO).

    Snapshots every injected sys.modules entry and restores or removes it on
    teardown, so later tests never observe the fake homeassistant/voluptuous
    (nor the stub-loaded nsh.helpers/nsh.config_flow) modules.
    """
    snapshot = entity_stublib.snapshot_modules()
    _install_import_stubs()
    yield
    entity_stublib.restore_modules(snapshot)


@pytest.fixture
def mocked_test_connect(monkeypatch):
    from nsh.connection import ShieldConnectionManager

    mock = AsyncMock(return_value=True)
    monkeypatch.setattr(ShieldConnectionManager, "test_connect", mock)
    # Preload + patch the lazy-loaded config-flow module's ensure_adb_key
    # (no key material — and no adb_shell keygen — during tests).
    module = _config_flow_module()
    monkeypatch.setattr(module, "ensure_adb_key", lambda _config_dir: _KEY_PATH)
    return mock


async def test_pairing_submit_uses_declared_signature(mocked_test_connect):
    """F1 (HIGH): exactly (host, port, key_path) — no inherited hass arg."""
    flow = _flow("shield.local")

    result = await flow.async_step_adb_pair({"submit": True})

    mocked_test_connect.assert_awaited_once_with(
        "shield.local", 5555, _KEY_PATH
    )
    # t56: the pairing hands over to the setup menu; Finish creates the entry
    assert result["type"] == "menu"
    result = await flow.async_step_finish(None)
    assert result["type"] == "create_entry"


async def test_pairing_submit_reports_adb_failed_on_false(mocked_test_connect):
    flow = _flow("shield.local")
    mocked_test_connect.return_value = False

    result = await flow.async_step_adb_pair({"submit": True})

    mocked_test_connect.assert_awaited_once()
    assert result["type"] == "form"
    assert result["errors"] == {"base": "adb_failed"}


async def test_pairing_submit_without_host_reports_error(mocked_test_connect):
    flow = _flow("")  # no host resolvable

    result = await flow.async_step_adb_pair({"submit": True})

    mocked_test_connect.assert_not_awaited()
    assert result["errors"] == {"base": "adb_no_host"}