feat(config_flow): standalone setup — androidtv entity optional with no_state_source validation, entry data omits the field for standalone entries, sensor tolerates its absence; migration keeps v1 field / v2 absence valid (P3-1a,d)

tests/test_standalone_setup.py

@@ -0,0 +1,300 @@
+"""P3-1 tests: standalone setup — optional androidtv entity + app fallback.
+
+The integration no longer hard-requires an external androidtv media_player:
+the protocol current-app push can be the only state source. Covers the flow
+validation (no state source → form error), the standalone pairing route,
+entry data omitting the field, media_player app_id/app_name fallback, and
+migration validity for v2 entries without the field.
+"""
+
+import asyncio  # noqa: F401  (async steps need the event loop)
+import sys
+import types
+from types import SimpleNamespace
+
+import pytest
+from conftest import _SHIELD_DIR, _load_module
+from nsh.const import (
+    CONF_ANDROIDTV_ENTITY,
+    CONF_PROTOCOL_ENABLED,
+    CONF_PROTOCOL_HOST,
+)
+
+_STUB_NAMES = (
+    "voluptuous",
+    "homeassistant",
+    "homeassistant.config_entries",
+    "homeassistant.core",
+    "homeassistant.const",
+    "homeassistant.helpers",
+    "homeassistant.helpers.entity_registry",
+    "homeassistant.helpers.selector",
+    "nsh.helpers",
+    "nsh.config_flow",
+    "nsh.entry",
+)
+
+
+def _install_stubs() -> None:
+    """Import-time stand-ins (idempotent; same shape as the pairing tests)."""
+    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)
+            super().__init_subclass__(**kwargs)
+
+        def async_show_form(self, *, step_id, data_schema=None, errors=None):
+            return {"type": "form", "step_id": step_id, "errors": dict(errors or {})}
+
+        def async_create_entry(self, *, title, data, options=None):
+            return {"type": "create_entry", "title": title, "data": data,
+                    "options": options}
+
+        def async_abort(self, *, reason):
+            return {"type": "abort", "reason": reason}
+
+        async def async_set_unique_id(self, unique_id):
+            self._unique_id = unique_id
+
+        def _abort_if_unique_id_configured(self, *, error=None, **kwargs):
+            return None
+
+    class OptionsFlow:
+        pass
+
+    config_entries.ConfigFlow = ConfigFlow
+    config_entries.OptionsFlow = OptionsFlow
+    config_entries.ConfigEntry = type("ConfigEntry", (), {})
+    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
+
+    ha_const = types.ModuleType("homeassistant.const")
+    ha_const.EVENT_HOMEASSISTANT_STOP = "homeassistant_stop"
+
+    class _Platform:
+        MEDIA_PLAYER = "media_player"
+        SENSOR = "sensor"
+        REMOTE = "remote"
+        BUTTON = "button"
+
+    ha_const.Platform = _Platform
+    sys.modules["homeassistant.const"] = ha_const
+
+    helpers = types.ModuleType("homeassistant.helpers")
+    entity_registry = types.ModuleType("homeassistant.helpers.entity_registry")
+    entity_registry.async_get = lambda hass: SimpleNamespace(entities={})
+    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 FakeConfigEntries:
+    def __init__(self):
+        self.updating_entry_calls: list = []
+
+    def async_get_entry(self, entry_id):
+        return None
+
+    def async_update_entry(self, entry, **kwargs):
+        self.updating_entry_calls.append((entry, kwargs))
+
+
+class FakeHass:
+    def __init__(self, tmp_dir: str, states: dict | None = None):
+        self.config = SimpleNamespace(config_dir=tmp_dir)
+        self.config_entries = FakeConfigEntries()
+        self._states = states or {}
+
+        class _States:
+            def get(self, entity_id):
+                return self._states.get(entity_id)
+
+        self._states_ref = _States()
+        self._states_ref._states = self._states
+        self.states = self._states_ref
+
+    @staticmethod
+    async def async_add_executor_job(fn, *args):
+        return fn(*args)
+
+
+@pytest.fixture(autouse=True)
+def scoped_stubs():
+    """Keep the import stubs inside this module's tests (t7 discipline)."""
+    snapshot = {name: sys.modules.get(name) for name in _STUB_NAMES}
+    _install_stubs()
+    yield
+    for name in _STUB_NAMES:
+        original = snapshot[name]
+        if original is None:
+            sys.modules.pop(name, None)
+        else:
+            sys.modules[name] = original
+
+
+def _form_module():
+    _install_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 _entry_module():
+    _install_stubs()
+    if "nsh.entry" not in sys.modules:
+        _load_module("nsh.entry", _SHIELD_DIR / "__init__.py")
+    return sys.modules["nsh.entry"]
+
+
+def _user_input(**overrides) -> dict:
+    base = {
+        "name": "mdarcy",
+        CONF_ANDROIDTV_ENTITY: "media_player.shield",
+        CONF_PROTOCOL_ENABLED: False,
+        CONF_PROTOCOL_HOST: "",
+    }
+    base.update(overrides)
+    return base
+
+
+class TestStandaloneValidation:
+    async def test_neither_source_shows_form_error(self, tmp_path):
+        module = _form_module()
+        flow = module.NvidiaShieldConfigFlow()
+        flow.hass = FakeHass(str(tmp_path))
+
+        result = await flow.async_step_user(_user_input(**{CONF_ANDROIDTV_ENTITY: ""}))
+
+        assert result["type"] == "form"
+        assert result["step_id"] == "user"
+        assert result["errors"] == {"base": "no_state_source"}
+
+    async def test_androidtv_entity_alone_is_a_valid_source(self, tmp_path):
+        module = _form_module()
+        flow = module.NvidiaShieldConfigFlow()
+        flow.hass = FakeHass(str(tmp_path))
+
+        result = await flow.async_step_user(_user_input())
+
+        assert result["type"] == "create_entry"
+        assert result["data"] == {CONF_ANDROIDTV_ENTITY: "media_player.shield"}
+
+    async def test_standalone_protocol_goes_to_pairing(self, tmp_path, monkeypatch):
+        module = _form_module()
+        flow = module.NvidiaShieldConfigFlow()
+        flow.hass = FakeHass(str(tmp_path))
+        flow._user_data = {}
+        from nsh.connection import ShieldConnectionManager
+
+        class _Client:
+            async def async_generate_cert_if_missing(self):
+                return True
+
+            async def async_start_pairing(self):
+                return "8421"
+
+        monkeypatch.setattr(
+            ShieldConnectionManager, "build_protocol_client",
+            lambda *a, **k: _Client(),
+        )
+        monkeypatch.setattr(
+            module, "protocol_cert_paths", lambda hass: ("/tmp/a.pem", "/tmp/a.key")
+        )
+
+        result = await flow.async_step_user(
+            _user_input(**{CONF_ANDROIDTV_ENTITY: "", CONF_PROTOCOL_ENABLED: True})
+        )
+
+        assert result["type"] == "form"
+        assert result["step_id"] == "protocol_pair"  # routed into pairing, no error
+
+    async def test_standalone_entry_data_omits_the_field(self, tmp_path, monkeypatch):
+        """No androidtv entity → the key is absent from entry data (not "")."""
+        module = _form_module()
+        flow = module.NvidiaShieldConfigFlow()
+        flow.hass = FakeHass(str(tmp_path))
+        flow._user_data = {
+            "name": "mdarcy",
+            CONF_ANDROIDTV_ENTITY: "",
+            CONF_PROTOCOL_ENABLED: True,
+            CONF_PROTOCOL_HOST: "shield.local",
+        }
+
+        # skip the pairing route: drive _create_entry directly
+        result = flow._create_entry()
+
+        assert result["type"] == "create_entry"
+        assert CONF_ANDROIDTV_ENTITY not in result["data"]
+
+
+class TestMigrationKeepsOrAllowsAbsence:
+    async def test_v1_to_v2_keeps_the_androidtv_field(self, tmp_path):
+        module = _entry_module()
+        hass = FakeHass(str(tmp_path))
+        entry = SimpleNamespace(
+            version=1,
+            data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
+            options={},
+            title="mdarcy",
+        )
+
+        ok = await module.async_migrate_entry(hass, entry)
+
+        assert ok is True
+        assert entry.data[CONF_ANDROIDTV_ENTITY] == "media_player.shield"  # kept
+        # options were rewritten with the protocol defaults; the data dict —
+        # including the androidtv field — was never touched
+        assert hass.config_entries.updating_entry_calls == [
+            (
+                entry,
+                {
+                    "options": {CONF_PROTOCOL_ENABLED: False, CONF_PROTOCOL_HOST: ""},
+                    "version": 2,
+                },
+            )
+        ]
+
+    async def test_v2_without_the_field_is_valid(self, tmp_path):
+        """Standalone entries born v2 must pass through migration untouched."""
+        module = _entry_module()
+        hass = FakeHass(str(tmp_path))
+        entry = SimpleNamespace(
+            version=2,
+            data={},
+            options={CONF_PROTOCOL_ENABLED: True, CONF_PROTOCOL_HOST: "shield.local"},
+            title="mdarcy",
+        )
+
+        ok = await module.async_migrate_entry(hass, entry)
+
+        assert ok is True
+        assert hass.config_entries.updating_entry_calls == []  # no rewrite needed