test(init): tests/test_init.py lands — domain async_setup units + YAML seed parsing/normalization, flow-init seeding records, hostless drop, duplicate guards (UI host + virtual unique_id) and single-visit interactive routing; scoped stub hygiene mirrors scoped_stubs (t30 G6/U1)

tests/test_init.py

@@ -0,0 +1,240 @@
+"""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
+
+
+@pytest.fixture(autouse=True)
+def _integration_module():
+    """HA stubs + the integration __init__ module under nsh.init."""
+    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"]
+    sys.modules.pop("nsh.init", None)
+
+
+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}
+        super().__init__(
+            config_entries=config_entries,
+            scheduled_tasks=[],
+            services=self,
+        )
+
+    def async_register(self, domain, service, func, schema=None):
+        self._services[(domain, service)] = (func, schema)
+
+    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_HOST, CONF_ADB_PORT, DEFAULT_ADB_PORT
+
+        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_ADB_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_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
+        # protocol fields are prefill hints ONLY (pairing stays interactive)
+        assert blocks[0]["protocol_host"] == "shield.local"
+        assert blocks[0]["adb_port"] == 5000
+
+    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["adb_host"] == "10.0.0.5"
+
+    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}
+        return flow
+
+    async def test_ui_configured_host_never_re_seeded(self):
+        from nsh.const import CONF_ADB_HOST
+
+        existing = SimpleNamespace(
+            entry_id="ui1",
+            unique_id=None,
+            options={CONF_ADB_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"