test(config_flow): zeroconf discovery (pre-fill, no-host abort, IP-change host update, same-IP skip) + apps map editor (store/parse/clear/prefill) + app_name map resolution tiers (P4-1)

2026-10-08 11:03:28 +01:00
1 parent 61c7ac99a3
commit f4d750ff92
@@ -59,7 +59,12 @@ def _install_stubs() -> 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 {})}
+            return {
+                "type": "form",
+                "step_id": step_id,
+                "errors": dict(errors or {}),
+                "data_schema": data_schema,
+            }

        def async_create_entry(self, *, title, data, options=None):
            return {"type": "create_entry", "title": title, "data": data,
@@ -74,7 +79,27 @@ def _install_stubs() -> None:
        def _abort_if_unique_id_configured(self, *, error=None, **kwargs):
            return None

+        def _async_current_entries(self):
+            return _zc_holder.entries
+
    class OptionsFlow:
+        def async_show_form(self, *, step_id, data_schema=None, errors=None):
+            return {
+                "type": "form",
+                "step_id": step_id,
+                "errors": dict(errors or {}),
+                "data_schema": data_schema,
+            }
+
+        def async_create_entry(self, *, title, data):
+            return {"type": "create_entry", "title": title, "data": data}
+
+    # Shared holder so tests can swap the current-entry list through the
+    # module object (the stub closure reads this same object).
+    _zc_holder = types.SimpleNamespace(entries=[])
+    sys.modules[__name__]._zc_holder = _zc_holder
+
+    class AbortIfConfiguredError(Exception):
        pass

    config_entries.ConfigFlow = ConfigFlow
@@ -118,12 +143,19 @@ def _install_stubs() -> None:
 class FakeConfigEntries:
    def __init__(self):
        self.updating_entry_calls: list = []
+        self.reloaded_entry_ids: list[str] = []

    def async_get_entry(self, entry_id):
        return None

    def async_update_entry(self, entry, **kwargs):
        self.updating_entry_calls.append((entry, kwargs))
+        for key, value in kwargs.items():
+            if key in ("options", "data"):
+                setattr(entry, key, value)
+
+    def async_schedule_reload(self, entry_id):
+        self.reloaded_entry_ids.append(entry_id)


 class FakeHass:
@@ -332,3 +364,196 @@ class TestMigrationKeepsOrAllowsAbsence:

        assert ok is True
        assert hass.config_entries.updating_entry_calls == []  # no rewrite needed
+
+
+def _cmd_player(modules):
+    """Standalone media_player entity via the command-routing harness."""
+    import test_command_routing as routing
+
+    player, _served, _intents = routing.make_player(modules)
+    return player, None, None
+
+
+class TestZeroconfDiscovery:
+    """P4-1a: _androidtvremote2._tcp discovery → pre-filled user step."""
+
+    def _discovery(self, host="192.168.1.50", props=None):
+        return SimpleNamespace(
+            ip_address=host, host=host, properties=props or {}
+        )
+
+    async def test_discovery_prefills_the_user_step(self, tmp_path):
+        module = _form_module()
+        flow = module.NvidiaShieldConfigFlow()
+        flow.hass = FakeHass(str(tmp_path))
+
+        result = await flow.async_step_zeroconf(self._discovery())
+
+        assert result["type"] == "form"
+        assert result["step_id"] == "user"
+        # pre-fill landed: the discovered host is recorded for the schema
+        assert flow._discovered_host == "192.168.1.50"
+        assert result["data_schema"] is not None
+
+    async def test_discovery_without_address_aborts(self, tmp_path):
+        module = _form_module()
+        flow = module.NvidiaShieldConfigFlow()
+        flow.hass = FakeHass(str(tmp_path))
+
+        result = await flow.async_step_zeroconf(
+            SimpleNamespace(ip_address=None, host=None, properties={})
+        )
+
+        assert result == {"type": "abort", "reason": "zeroconf_no_host"}
+
+    async def test_known_mac_ip_change_updates_entry_hosts(self, tmp_path):
+        module = _form_module()
+        flow = module.NvidiaShieldConfigFlow()
+        hass = FakeHass(str(tmp_path))
+        flow.hass = hass
+
+        entry = SimpleNamespace(
+            unique_id="aa:bb:cc:dd:ee:ff",
+            options={CONF_PROTOCOL_ENABLED: True, CONF_PROTOCOL_HOST: "10.0.0.1"},
+            entry_id="e1",
+            title="Shield",
+        )
+        sys.modules[__name__]._zc_holder.entries[:] = [entry]
+
+        result = await flow.async_step_zeroconf(
+            self._discovery(props={"mac": "aa:bb:cc:dd:ee:ff"})
+        )
+
+        assert result == {"type": "abort", "reason": "already_configured"}
+        # the hosts moved to the discovery address
+        assert entry.options[CONF_PROTOCOL_HOST] == "192.168.1.50"
+        assert entry.options["adb_host"] == "192.168.1.50"
+        assert hass.config_entries.reloaded_entry_ids == ["e1"]
+
+    async def test_known_mac_same_ip_skips_the_rewrite(self, tmp_path):
+        module = _form_module()
+        flow = module.NvidiaShieldConfigFlow()
+        hass = FakeHass(str(tmp_path))
+        flow.hass = hass
+
+        entry = SimpleNamespace(
+            unique_id="aa:bb:cc:dd:ee:ff",
+            options={CONF_PROTOCOL_ENABLED: True, CONF_PROTOCOL_HOST: "192.168.1.50"},
+            entry_id="e1",
+            title="Shield",
+        )
+        sys.modules[__name__]._zc_holder.entries[:] = [entry]
+
+        result = await flow.async_step_zeroconf(
+            self._discovery(props={"mac": "aa:bb:cc:dd:ee:ff"})
+        )
+
+        assert result == {"type": "abort", "reason": "already_configured"}
+        assert hass.config_entries.updating_entry_calls == []  # no IP change
+        assert hass.config_entries.reloaded_entry_ids == []
+
+
+class TestAppsMapOptionsEditor:
+    """P4-1b: the apps name/icon map editor in the options flow."""
+
+    def _options_flow(self, tmp_path, options):
+        module = _form_module()
+        entry = SimpleNamespace(options=dict(options), title="Shield")
+        flow = module.NvidiaShieldOptionsFlow(entry)
+        flow.hass = FakeHass(str(tmp_path))
+        return flow
+
+    async def test_map_text_stored_as_dict(self, tmp_path):
+        flow = self._options_flow(tmp_path, options={})
+
+        result = await flow.async_step_init(
+            {"app_name_map": "ar.tvplayer.tv=TiviMate\ncom.plex.hdx=Plex"}
+        )
+
+        assert result["type"] == "create_entry"
+        assert flow._options["app_name_map"] == {
+            "ar.tvplayer.tv": "TiviMate",
+            "com.plex.hdx": "Plex",
+        }
+
+    async def test_invalid_map_shows_form_error(self, tmp_path):
+        flow = self._options_flow(tmp_path, options={})
+
+        result = await flow.async_step_init({"app_name_map": "no equals sign here"})
+
+        assert result["type"] == "form"
+        assert result["errors"] == {"base": "invalid_app_map"}
+
+    async def test_blank_map_clears_the_entries(self, tmp_path):
+        flow = self._options_flow(
+            tmp_path, options={"app_name_map": {"ar.tvplayer.tv": "TiviMate"}}
+        )
+
+        result = await flow.async_step_init({"app_name_map": "  "})
+
+        assert result["type"] == "create_entry"
+        assert flow._options["app_name_map"] == {}
+
+    async def test_existing_map_prefills_the_text_field(self, tmp_path):
+        flow = self._options_flow(
+            tmp_path, options={"app_name_map": {"ar.tvplayer.tv": "TiviMate"}}
+        )
+
+        # first visit: the form carries the serialized current map as the
+        # field default
+        result = await flow.async_step_init(None)
+
+        assert result["type"] == "form"
+        # the editor carried the existing map through the defaults path
+        assert flow._options["app_name_map"] == {"ar.tvplayer.tv": "TiviMate"}
+
+
+class TestAppNameMapResolution:
+    """app_name resolves through the map after the entity/push tiers."""
+
+    async def test_mapped_package_resolves_the_friendly_name(self):
+        import entity_stublib
+
+        modules = entity_stublib.load_entity_modules()
+        player, _s, _i = _cmd_player(modules)
+        player._entry.options["app_name_map"] = {"ar.tvplayer.tv": "TiviMate"}
+        player._androidtv_entity = ""  # standalone
+
+        manager = player._manager
+        manager._on_current_app("ar.tvplayer.tv")  # protocol push
+        await asyncio.sleep(0.05)  # drain the call_soon_threadsafe push
+
+        assert player.app_name == "TiviMate"
+        assert player.app_id == "ar.tvplayer.tv"  # app_id stays raw
+
+    async def test_unmapped_package_stays_raw(self):
+        import entity_stublib
+
+        modules = entity_stublib.load_entity_modules()
+        player, _s, _i = _cmd_player(modules)
+        player._entry.options["app_name_map"] = {"ar.tvplayer.tv": "TiviMate"}
+        player._androidtv_entity = ""
+
+        manager = player._manager
+        manager._on_current_app("com.google.android.youtube.tv")
+        await asyncio.sleep(0.05)
+
+        assert player.app_name == "com.google.android.youtube.tv"
+
+    async def test_entity_tier_still_wins(self):
+        from types import SimpleNamespace as SimpleNS
+
+        import entity_stublib
+
+        modules = entity_stublib.load_entity_modules()
+        player, _s, _i = _cmd_player(modules)
+        player._entry.options["app_name_map"] = {"ar.tvplayer.tv": "TiviMate"}
+        player._androidtv_entity = "media_player.atv"
+        state = SimpleNS(attributes={"app_id": "ar.tvplayer.tv", "app_name": "ATV Tivi"})
+        player.hass = SimpleNS(states=SimpleNS(get=lambda eid: state))
+        player._read = lambda eid: state if eid == "media_player.atv" else None
+
+        player._manager._on_current_app("ar.tvplayer.tv")
+        await asyncio.sleep(0.05)
+
+        assert player.app_name == "ATV Tivi"  # entity tier, map untouched