fix(config_flow): pairing step passes test_connect's declared signature (host, port, key_path) — regression tests cover the adb_pair submit path (F1)

custom_components/nvidia_shield/config_flow.py

@@ -209,7 +209,7 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
                 )
 
                 connected = await ShieldConnectionManager.test_connect(
-                    self.hass, host, port, key_path
+                    host, port, key_path
                 )
                 if connected:
                     return self._create_entry()

tests/test_config_flow_pairing.py

@@ -0,0 +1,184 @@
+"""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 pytest
+from conftest import _SHIELD_DIR, _load_module
+from nsh.const import (
+    CONF_ADB_ENABLED,
+    CONF_ADB_HOST,
+    CONF_ADB_PORT,
+    CONF_ADB_STORE_APP_INFO,
+    CONF_ANDROIDTV_ENTITY,
+    CONF_JELLYFIN_ENTITY,
+    CONF_POWER_ENTITY,
+    CONF_RECEIVER_ENTITY,
+    CONF_RECEIVER_INPUT,
+)
+
+_KEY_PATH = "/tmp/nsh-test-adbkey"
+
+
+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_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
+    config_entries.OptionsFlow = type("OptionsFlow", (), {})
+    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
+
+    selector = types.ModuleType("homeassistant.helpers.selector")
+    selector.EntitySelector = lambda *a, **k: None
+    selector.EntitySelectorConfig = lambda *a, **k: ""
+    selector.SelectSelector = lambda *a, **k: None
+    selector.SelectSelectorConfig = lambda **k: ""
+    selector.SelectSelectorMode = types.SimpleNamespace(DROPDOWN="dropdown")
+    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_ADB_HOST: user_input_host,
+        CONF_ADB_PORT: 5555,
+        CONF_ANDROIDTV_ENTITY: "media_player.shield",
+        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
+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
+    )
+    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"}