ha-nvidia-shield/tests/test_init.py

"""tests/test_init.py — the t30 G6 structure test file.

Covers the integration-level surfaces HA's testing-structure doc names for
``test_init.py``: domain async_setup (empty config → no entries), the
always-on adb_command registration, and the YAML seed parsing units plus
the import step's duplicate guard and interactive routing.
"""

import asyncio
import sys
from types import SimpleNamespace

import entity_stublib
import pytest

# Scoped-stub hygiene (mirrors tests/test_config_flow_pairing.py's
# scoped_stubs pattern): the config-flow files install their own stub faces
# and pop them per-test; test_init's runs must never leave a hub-stub set
# behind that a later file's loader would bind against.
_INIT_STUB_NAMES = (
    "voluptuous",
    "homeassistant",
    "homeassistant.config_entries",
    "homeassistant.core",
    "homeassistant.helpers",
    "homeassistant.helpers.entity_registry",
    "homeassistant.helpers.device_registry",
    "homeassistant.helpers.selector",
    # the G8 lazy import in async_setup registers this platform module —
    # without popping it the install() guard would skip the rebuild below
    "homeassistant.components.media_player",
    "nsh.helpers",
    "nsh.config_flow",
    "nsh.init",
    "nsh.media_player",
)


@pytest.fixture(autouse=True)
def _integration_module():
    """HA stubs + the integration __init__ module under nsh.init."""
    snapshot = entity_stublib.snapshot_modules()
    entity_stublib.install()
    if "nsh.init" not in sys.modules:
        import conftest

        conftest._load_module("nsh.init", conftest._SHIELD_DIR / "__init__.py")
    yield sys.modules["nsh.init"]
    entity_stublib.restore_modules(snapshot)


class _SeedEntriesStub:
    """config-entries shape for flow seeding tests: entries + flow records."""

    def __init__(self, entries=()):
        self._entries = list(entries)
        self.flows = []  # (domain, context, data) records of flow.async_init

    def async_entries(self):
        return list(self._entries)

    class _Flows:
        def __init__(self, holder):
            self._holder = holder

        async def async_init(self, domain, context=None, data=None):
            self._holder.flows.append((domain, context, data))
            return {"type": "form", "items": {}}

    @property
    def flow(self):
        return self._Flows(self)


class _SeedHass(SimpleNamespace):
    """Minimal hass for async_setup: config entries, services, tasks."""

    def __init__(self, entries=()):
        config_entries = _SeedEntriesStub()
        config_entries.updating_entry_calls = []
        self._services = {("nvidia_shield", "adb_command"): None}
        #: t14: per-service response support the registration declared.
        self.response_support: dict[tuple, object] = {}
        super().__init__(
            config_entries=config_entries,
            scheduled_tasks=[],
            services=self,
        )

    def async_register(
        self, domain, service, func, schema=None, supports_response=None
    ):
        self._services[(domain, service)] = (func, schema)
        #: t14: HA is told whether the action answers with a response.
        self.response_support[(domain, service)] = supports_response

    def has_service(self, domain, service):
        return (domain, service) in self._services

    def async_create_task(self, coro):
        task = asyncio.ensure_future(coro)
        self.scheduled_tasks.append(task)
        return task


class TestDomainSetup:
    """async_setup: empty config, service registration, no entries."""

    async def test_empty_config_creates_nothing(self, _integration_module):
        hass = _SeedHass(entries=[])

        result = await _integration_module.async_setup(hass, {})

        assert result is True
        assert hass.config_entries.flows == []  # nothing seeded
        assert hass.services.has_service("nvidia_shield", "adb_command")


class TestYamlSeedBlocks:
    """_yaml_seed_blocks: the optional domain key → normalized seeds."""

    def test_block_normalized(self, _integration_module):
        from nsh.const import CONF_ADB_PORT, CONF_HOST, DEFAULT_ADB_PORT

        # t41: a LEGACY block (adb_host) still seeds — normalized onto CONF_HOST
        blocks = _integration_module._yaml_seed_blocks(
            [{"name": "Living", "adb_host": "10.0.0.31"}]
        )

        assert len(blocks) == 1
        seed = blocks[0]
        assert seed["name"] == "Living"
        assert seed[CONF_HOST] == "10.0.0.31"
        assert seed[CONF_ADB_PORT] == DEFAULT_ADB_PORT
        assert seed["adb_enabled"] is True
        assert "unique_id" not in seed  # optional key only

    def test_the_current_host_key_seeds_too(self, _integration_module):
        from nsh.const import CONF_HOST

        seed = _integration_module._yaml_seed_blocks(
            [{"name": "Desk", CONF_HOST: "shield.local"}]
        )[0]

        assert seed[CONF_HOST] == "shield.local"
        assert seed["adb_enabled"] is True  # an address names the Shield

    def test_single_block_and_protocol_hint(self, _integration_module):
        blocks = _integration_module._yaml_seed_blocks(
            {"name": "Desk", "protocol_host": "shield.local", "adb_port": 5000}
        )

        assert len(blocks) == 1
        # a legacy protocol-only block still seeds, with ADB left OFF (t32 F4)
        assert blocks[0]["host"] == "shield.local"
        assert blocks[0]["adb_port"] == 5000
        assert blocks[0]["adb_enabled"] is False

    def test_list_and_single_block_shapes(self, _integration_module):
        single = _integration_module._yaml_seed_blocks({"name": "A"})
        listed = _integration_module._yaml_seed_blocks([{"name": "B"}, {"name": "C"}])

        assert len(single) == 1 and len(listed) == 2


class TestSeeding:
    """async_setup seeds SOURCE_IMPORT flows for shaped blocks only."""

    async def test_shaped_block_seeds_a_flow(self, _integration_module):
        hass = _SeedHass(entries=[])

        await _integration_module.async_setup(
            hass, {"nvidia_shield": [{"name": "R", "adb_host": "10.0.0.5"}]}
        )
        if hass.scheduled_tasks:
            await asyncio.gather(*hass.scheduled_tasks, return_exceptions=True)

        assert len(hass.config_entries.flows) == 1
        domain, context, data = hass.config_entries.flows[0]
        assert domain == "nvidia_shield"
        assert context["source"] == "import"
        assert data["host"] == "10.0.0.5"  # t41: one address field


class TestSeedingCurrentKey:
    """The t41 `host` key seeds exactly like the legacy one."""

    async def test_host_key_seeds_a_flow(self, _integration_module):
        hass = _SeedHass(entries=[])

        await _integration_module.async_setup(
            hass, {"nvidia_shield": [{"name": "H", "host": "10.0.0.7"}]}
        )
        if hass.scheduled_tasks:
            await asyncio.gather(*hass.scheduled_tasks, return_exceptions=True)

        assert len(hass.config_entries.flows) == 1
        assert hass.config_entries.flows[0][2]["host"] == "10.0.0.7"

    async def test_hostless_block_never_seeds(self, _integration_module):
        hass = _SeedHass(entries=[])

        result = await _integration_module.async_setup(
            hass, {"nvidia_shield": [{"name": "X"}]}
        )

        assert result is True
        assert hass.config_entries.flows == []  # hostless → warning only


class TestImportStep:
    """async_step_import: duplicate guards + interactive routing."""

    def _flow(self, entries):
        # the flow module loads under these stubs and the base plumbing is
        # injected locally — importing test_standalone_setup here would
        # leak its local stub set into later test files.
        from nsh.config_flow import NvidiaShieldConfigFlow

        hass = SimpleNamespace(
            config_entries=_SeedEntriesStub(entries),
        )
        hass.config_entries.flows = []
        flow = NvidiaShieldConfigFlow()
        flow.hass = hass
        flow.context = {}
        # ConfigFlow base attrs the stub class doesn't implement (t7 shape):
        flow.async_set_unique_id = None  # set below
        flow._collected_unique_id = None

        async def set_unique_id(uid):
            flow._collected_unique_id = uid

        flow.async_set_unique_id = set_unique_id

        def abort_if_unique_id_configured(*args, **kwargs):
            for entry in hass.config_entries.async_entries():
                if entry.unique_id == flow._collected_unique_id:
                    from nsh.config_flow import AbortIfConfiguredError

                    raise AbortIfConfiguredError(f"{entry.unique_id}")

        flow._abort_if_unique_id_configured = abort_if_unique_id_configured
        # the base ConfigFlow shows forms through the stub plumbing (matching
        # how the standalone flow tests install it):
        flow.async_show_form = lambda **kwargs: {
            "type": "form",
            **kwargs,
        }
        flow.async_create_entry = lambda **kwargs: {
            "type": "create_entry",
            **kwargs,
        }
        flow.async_abort = lambda **kwargs: {"type": "abort", **kwargs}
        flow._async_current_entries = lambda: list(
            hass.config_entries.async_entries()
        )
        return flow

    async def test_ui_configured_host_never_re_seeded(self):
        from nsh.const import CONF_HOST

        existing = SimpleNamespace(
            entry_id="ui1",
            unique_id=None,
            options={CONF_HOST: "10.0.0.31"},
        )
        # options live on the entry (conf zeroconf guard reads them)
        flow = self._flow([existing])

        result = await flow.async_step_import(
            {"name": "A", "adb_host": "10.0.0.31"}
        )

        assert result == {"type": "abort", "reason": "already_configured"}

    async def test_virtual_unique_id_blocks_reseed(self):
        flow_holder = SimpleNamespace(
            entry_id="seed1",
            unique_id="yaml:Living:10.0.0.31",
            options={},
        )
        flow = self._flow([flow_holder])

        result = await flow.async_step_import(
            {"name": "Living", "adb_host": "10.0.0.31"}
        )

        assert result == {"type": "abort", "reason": "already_configured"}

    async def test_fresh_block_routes_into_the_user_flow(self):
        flow = self._flow([])

        result = await flow.async_step_import(
            {"name": "Fresh", "adb_host": "10.0.0.99"}
        )

        # interactive pairing — the pre-filled user step, one visit
        assert result["type"] == "form"
        assert result["step_id"] == "user"