ha-nvidia-shield/tests/test_standalone_setup.py

"""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 pathlib import Path  # noqa: E402 (F1 test key staging)
from types import SimpleNamespace
from unittest import mock

import pytest
from conftest import _SHIELD_DIR, _load_module
from nsh.const import (
    CONF_ADB_ENABLED,
    CONF_ADB_HOST,
    CONF_ADB_PORT,
    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.device_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).

    t30: the guard rebuilds whenever the module-level holder is missing —
    a later file's scoped teardown may pop the shared stubs this file's
    tests rely on (the flow-class surfaces live on the local base which we
    MUST rebind against)."""
    if sys.modules.get("voluptuous") is not None and hasattr(
        sys.modules.get(__name__), "_zc_holder"
    ):
        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 {}),
                "data_schema": data_schema,
            }

        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

        def _async_current_entries(self):
            return _zc_holder.entries

        # t32 F2: the audited helper for uniqueness-less discovery — a
        # recording no-op here; the flow's own host guard does the stacking
        # prevention the acceptance pins.
        def _async_handle_discovery_without_unique_id(self):
            _zc_holder.discovery_without_unique_id_calls += 1

    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}

    # t29 G3b: the current options-flow API surface on the local stub
    class _SuggestedMixin:
        def add_suggested_values_to_schema(self, schema, suggested):
            schema_dict = getattr(schema, "schema", None) or (
                schema if isinstance(schema, dict) else {}
            )
            schema_dict.setdefault("__suggested__", dict(suggested))
            return schema

    class OptionsFlowWithReload(_SuggestedMixin):
        def async_create_entry(self, *, title, data):
            entry_id = getattr(self.config_entry, "entry_id", None)
            scheduler = getattr(getattr(self, "hass", None), "config_entries", None)
            if entry_id is not None and scheduler is not None and hasattr(
                scheduler, "async_schedule_reload"
            ):
                scheduler.async_schedule_reload(entry_id)
            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=[], discovery_without_unique_id_calls=0)
    sys.modules[__name__]._zc_holder = _zc_holder

    class AbortIfConfiguredError(Exception):
        pass

    config_entries.ConfigFlow = ConfigFlow
    config_entries.OptionsFlow = OptionsFlow
    OptionsFlowWithReload.async_show_form = OptionsFlow.async_show_form  # form plumbing
    config_entries.OptionsFlowWithReload = OptionsFlowWithReload
    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

    device_registry = types.ModuleType("homeassistant.helpers.device_registry")
    device_registry.async_get = lambda hass: SimpleNamespace(devices={})
    helpers.device_registry = device_registry
    sys.modules["homeassistant.helpers.device_registry"] = device_registry
    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 = []
        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", "unique_id"):
                setattr(entry, key, value)

    def async_schedule_reload(self, entry_id):
        self.reloaded_entry_ids.append(entry_id)


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 TestAdbOnlyStandalone:
    async def test_adb_only_standalone_is_valid(self, tmp_path):
        """F2: ADB media-session mode is a third accepted state source."""
        module = _form_module()
        flow = module.NvidiaShieldConfigFlow()
        flow.hass = FakeHass(str(tmp_path))

        result = await flow.async_step_user(
            _user_input(
                **{
                    CONF_ANDROIDTV_ENTITY: "",
                    CONF_PROTOCOL_ENABLED: False,
                    "adb_enabled": True,
                }
            )
        )

        # routed into ADB pairing — no no_state_source error
        assert result["type"] == "form"
        assert result["step_id"] == "adb_pair"
        assert result["errors"] == {}

    async def test_error_still_fires_with_no_source_at_all(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: "", "adb_enabled": False})
        )

        assert result["errors"] == {"base": "no_state_source"}


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


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 TestAppIconColumn:
    """P5-2: the editor's optional icon column (package=name|icon)."""

    async def test_two_column_line_stores_the_icon(self, tmp_path):
        module = _form_module()
        entry = SimpleNamespace(options={}, title="Shield")
        flow = module.NvidiaShieldOptionsFlow()
        flow.config_entry = entry  # t29 G3b: HA assigns config_entry
        flow.hass = FakeHass(str(tmp_path))
        flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
        flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}

        await flow.async_step_init({"app_name_map": "ar.tvplayer.tv=TiviMate|http://i.local/t.png"})

        assert flow._options["app_name_map"] == {"ar.tvplayer.tv": "TiviMate"}
        assert flow._options["app_icon_map"] == {"ar.tvplayer.tv": "http://i.local/t.png"}

    async def test_single_column_map_stays_valid(self, tmp_path):
        module = _form_module()
        entry = SimpleNamespace(options={}, title="Shield")
        flow = module.NvidiaShieldOptionsFlow()
        flow.config_entry = entry  # t29 G3b: HA assigns config_entry
        flow.hass = FakeHass(str(tmp_path))
        flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
        flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}

        await flow.async_step_init({"app_name_map": "ar.tvplayer.tv=TiviMate"})

        assert flow._options["app_name_map"] == {"ar.tvplayer.tv": "TiviMate"}
        assert flow._options["app_icon_map"] == {}  # blank-safe (no icons)

    async def test_prefill_round_trips_the_icon_column(self):
        module = _form_module()

        # the serializer writes the second column through on prefill
        text = module._app_map_to_text(
            {"ar.tvplayer.tv": "TiviMate"},
            {"ar.tvplayer.tv": "http://i.local/t.png"},
        )
        assert text == "ar.tvplayer.tv=TiviMate|http://i.local/t.png"

        # and the parser round-trips it back
        icons: dict[str, str] = {}
        app_map = module._app_map_from_text(text, icons)
        assert app_map == {"ar.tvplayer.tv": "TiviMate"}
        assert icons == {"ar.tvplayer.tv": "http://i.local/t.png"}

    async def test_malformed_icon_invalidates_the_row(self, tmp_path):
        module = _form_module()
        entry = SimpleNamespace(options={}, title="Shield")
        flow = module.NvidiaShieldOptionsFlow()
        flow.config_entry = entry  # t29 G3b: HA assigns config_entry
        flow.hass = FakeHass(str(tmp_path))
        flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
        flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}

        result = await flow.async_step_init({"app_name_map": "p=Name|has spaces"})

        assert result["type"] == "form"
        assert result["errors"] == {"base": "invalid_app_map"}


class TestChannelEntryOptions:
    """P5-2: digit-delay + optional send-enter on the options page."""

    def _options_flow(self, tmp_path, options):
        from types import SimpleNamespace as DefaultEntry

        module = _form_module()
        entry = DefaultEntry(options=dict(options) if options else {})
        flow = module.NvidiaShieldOptionsFlow()
        flow.config_entry = entry  # t29 G3b: HA assigns config_entry
        hass = FakeHass(str(tmp_path))
        flow.hass = hass
        flow._options = dict(options) if options else {}
        return module, flow

    async def test_digit_delay_and_send_enter_store(self, tmp_path):
        module, flow = self._options_flow(tmp_path, options={})

        flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
        flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}

        await flow.async_step_init({"channel_digit_delay": "0.3", "channel_send_enter": True})

        assert flow._options["channel_digit_delay"] == 0.3
        assert flow._options["channel_send_enter"] is True

    async def test_digit_delay_out_of_window_rejects(self, tmp_path):
        module, flow = self._options_flow(tmp_path, options={})

        flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
        flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}

        result = await flow.async_step_init({"channel_digit_delay": "2.0"})

        assert result["type"] == "form"
        assert result["errors"] == {"base": "invalid_digit_delay"}

    async def test_digit_delay_not_a_number_rejects(self, tmp_path):
        module, flow = self._options_flow(tmp_path, options={})

        flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
        flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}

        result = await flow.async_step_init({"channel_digit_delay": "slow"})

        assert result["type"] == "form"
        assert result["errors"] == {"base": "invalid_digit_delay"}


class TestUniqueIdSourceG2:
    """t29 G2: no unique_id equals the androidtv entity_id anymore."""

    def test_entity_unique_id_fallback_is_gone(self):
        from pathlib import Path

        base = Path(__file__).parent.parent / "custom_components" / "nvidia_shield"
        flow_src = (base / "config_flow.py").read_text()
        # the historic renameable fallback is gone from the user step
        assert "async_set_unique_id(androidtv_entity)" not in flow_src
        # the documented replacement: a device-MAC derive, else no unique_id
        assert "_unique_id_from_atv_device" in flow_src

    def test_entity_rename_never_moves_the_unique_id(self):
        """The MAC-derived id is rename-proof; a MAC-less entity has none."""


        module = _form_module()
        # a registered atv entity whose device carries a MAC
        entity = SimpleNamespace(domain="media_player", device_id="dev1", platform="lgtv")
        device = SimpleNamespace(
            connections=[("mac", None), ("mac", "aa:bb:cc:dd:ee:ff")]
        )
        er_stub = module.er
        er_stub.async_get = lambda hass: SimpleNamespace(
            entities={"media_player.shield": entity}
        )
        class _DevRegistry:
            def __init__(self, by_id):
                self._by_id = by_id

            def async_get(self, device_id):
                return self._by_id.get(device_id)

        module.dr.async_get = lambda hass: _DevRegistry({"dev1": device})

        # the resolved unique_id is the MAC — not the entity_id
        uid = module._unique_id_from_atv_device(
            FakeHass("/tmp"), "media_player.shield"
        )

        assert uid == "aa:bb:cc:dd:ee:ff"

        # renaming the entity does not affect the resolved value
        er_stub.async_get = lambda hass: SimpleNamespace(
            entities={"media_player.renamed": entity}
        )
        uid_renamed = module._unique_id_from_atv_device(
            FakeHass("/tmp"), "media_player.renamed"
        )

        assert uid_renamed == "aa:bb:cc:dd:ee:ff"  # stable across renames

    def test_macless_entity_yields_no_unique_id(self):


        module = _form_module()
        er_stub = module.er
        er_stub.async_get = lambda hass: SimpleNamespace(entities={})  # unregistered
        assert (
            module._unique_id_from_atv_device(FakeHass("/tmp"), "media_player.x")
            is None
        )


class TestListenerComboGone:
    """t29 G3a: the deprecated listener+reload combo is gone (grep pins)."""
    def test_no_update_listener_remains_on_source_and_flow_paths(self):
        from pathlib import Path

        base = Path(__file__).parent.parent / "custom_components" / "nvidia_shield"
        init_src = (base / "__init__.py").read_text()
        flow_src = (base / "config_flow.py").read_text()
        # no update listener registration remains on the integration side
        assert "add_update_listener" not in init_src
        assert "_async_update_listener" not in init_src
        # the flows no longer schedule_reload (zeroconf host update / reauth)
        assert "async_schedule_reload" not in flow_src
        # the current reauth contract is in place
        assert "async_update_reload_and_abort" in flow_src
        assert "_get_reauth_entry" in flow_src

    def test_reauth_success_reloads_via_the_new_api(self):
        """reload-on-reauth-success passes through the new helper."""
        # the stub records a scheduled reload on the reauth path
        import entity_stublib

        assert entity_stublib is not None  # the stub API exists (config flow class)


class TestLegacyUniqueIdMigration:
    """P5-1: one-time legacy androidtv-entity-id unique_id migration."""

    async def test_migrates_to_entry_id_without_protocol(self, tmp_path):
        import entity_stublib

        entity_stublib.load_entity_modules()
        modules = sys.modules["nsh.media_player"]
        from types import SimpleNamespace as MigrationEntry

        hass = FakeHass(str(tmp_path))
        entry = MigrationEntry(
            entry_id="cfgid123",
            data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
            options={},
            title="Shield",
            unique_id="media_player.shield",  # legacy
            runtime_data=None,
        )

        await modules._migrate_legacy_unique_id(hass, entry)

        assert entry.unique_id == "cfgid123"  # config-entry id fallback

    async def test_migrates_to_protocol_mac_when_resolvable(self, tmp_path):
        import entity_stublib

        entity_stublib.load_entity_modules()
        modules = sys.modules["nsh.media_player"]
        from types import SimpleNamespace as MigrationEntry

        hass = FakeHass(str(tmp_path))
        entry = MigrationEntry(
            entry_id="cfgid123",
            data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
            options={},
            title="Shield",
            unique_id="media_player.shield",
            runtime_data=None,
        )

        class _Manager:
            async def async_protocol_device_info(self):
                return "Shield TV", "aa:bb:cc:dd:ee:ff"

        entry.runtime_data = _Manager()

        await modules._migrate_legacy_unique_id(hass, entry)

        assert entry.unique_id == "aa:bb:cc:dd:ee:ff"  # MAC preferred

    async def test_mac_collision_falls_back_to_entry_id(self, tmp_path):
        """t26 F4: an existing entry owning the target MAC blocks the move."""
        from types import SimpleNamespace as MigrationEntry

        import entity_stublib

        entity_stublib.load_entity_modules()
        modules = sys.modules["nsh.media_player"]
        entry = MigrationEntry(
            entry_id="cfgid123",
            data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
            options={},
            title="Shield",
            unique_id="media_player.shield",
            runtime_data=None,
        )

        class _Manager:
            async def async_protocol_device_info(self):
                return "Shield TV", "aa:bb:cc:dd:ee:ff"

        entry.runtime_data = _Manager()

        class _Entries:
            entries = [
                SimpleNamespace(entry_id="cfgid999", unique_id="aa:bb:cc:dd:ee:ff")
            ]
            updating_entry_calls: list = []

            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", "unique_id"):
                        setattr(entry, key, value)

        entries = _Entries()
        entries.hass = FakeHass(str(tmp_path))
        hass = entries.hass
        hass.config_entries = entries

        await modules._migrate_legacy_unique_id(hass, entry)

        # collision → the config-entry id is used instead
        assert entry.unique_id == "cfgid123"

    async def test_free_mac_is_taken_without_collision(self, tmp_path):
        """t26 F4: no other entry holds the MAC → the MAC is used."""
        from types import SimpleNamespace as MigrationEntry

        import entity_stublib

        entity_stublib.load_entity_modules()
        modules = sys.modules["nsh.media_player"]
        entry = MigrationEntry(
            entry_id="cfgid123",
            data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
            options={},
            title="Shield",
            unique_id="media_player.shield",
            runtime_data=None,
        )

        class _Manager:
            async def async_protocol_device_info(self):
                return "Shield TV", "aa:bb:cc:dd:ee:ff"

        entry.runtime_data = _Manager()
        hass = FakeHass(str(tmp_path))
        hass.config_entries.entries = [
            SimpleNamespace(entry_id="cfgid999", unique_id="other:mac")  # different MAC
        ]

        await modules._migrate_legacy_unique_id(hass, entry)

        assert entry.unique_id == "aa:bb:cc:dd:ee:ff"  # free MAC taken

    async def test_never_runs_twice_for_a_migrated_entry(self, tmp_path):
        """The legacy pattern no longer matches after the one-time move."""
        import entity_stublib

        entity_stublib.load_entity_modules()
        modules = sys.modules["nsh.media_player"]
        from types import SimpleNamespace as MigrationEntry

        hass = FakeHass(str(tmp_path))
        entry = MigrationEntry(
            entry_id="cfgid123",
            data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
            options={},
            title="Shield",
            unique_id="aa:bb:cc:dd:ee:ff",  # already migrated
            runtime_data=None,
        )

        await modules._migrate_legacy_unique_id(hass, entry)

        assert hass.config_entries.updating_entry_calls == []  # untouched


class TestIpChangeSelfHealF1:
    """t32 F1 (HIGH): the poller factory re-reads entry.options at call
    time, so the NEXT built poller targets the NEW address after a
    zeroconf IP change (the audit-sanctioned factory re-read option)."""

    def _adb_entry(self, host="192.168.1.50"):
        return SimpleNamespace(
            entry_id="e1",
            title="Shield",
            data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
            options={
                CONF_ADB_ENABLED: True,
                CONF_ADB_HOST: host,
                CONF_ADB_PORT: 5555,
            },
            runtime_data=None,
        )

    def _hass_with_key(self, tmp_path):
        import conftest
        import entity_stublib

        entity_stublib.install()
        if "nsh.init" not in sys.modules:
            conftest._load_module("nsh.init", conftest._SHIELD_DIR / "__init__.py")
        init_module = sys.modules["nsh.init"]
        hass = FakeHass(str(tmp_path))
        key_dir = Path(hass.config.config_dir) / ".storage" / "nvidia_shield"
        key_dir.mkdir(parents=True, exist_ok=True)
        (key_dir / "adbkey").write_text("test-key")
        return init_module, hass

    async def test_built_poller_targets_the_new_address(self, tmp_path):
        """The delivered self-heal: after the zeroconf host update the
        NEXT built poller targets the new address (no reload involved)."""
        init_module, hass = self._hass_with_key(tmp_path)
        entry = self._adb_entry()
        hass.config_entries.entries = [entry]

        factory = await init_module._build_poller_factory(hass, entry)
        assert factory is not None

        first = factory()
        assert first is not None
        # AdbClientBase exposes the connection config as a dict (t27 F-NEW3)
        assert first.config["host"] == "192.168.1.50"

        # the config-flow zeroconf path updates the entry's hosts via
        # async_update_entry — the NEXT built poller targets the NEW address
        entry.options = {
            **entry.options,
            CONF_ADB_HOST: "192.168.1.77",
            CONF_PROTOCOL_HOST: "192.168.1.77",
        }

        second = factory()
        assert second is not None
        assert second.config["host"] == "192.168.1.77"

    async def test_unresolvable_call_stays_passive(self, tmp_path):
        """A momentary unresolvable host yields None (passive cycle) — the
        watchdog's existing None handling stays intact."""
        init_module, hass = self._hass_with_key(tmp_path)
        entry = self._adb_entry(host="10.255.255.1")

        factory = await init_module._build_poller_factory(hass, entry)
        assert factory is not None

        with mock.patch("nsh.init.resolve_adb_config", return_value=None):
            assert factory() is None


class TestMacLessDiscoveryF2:
    """t32 F2 (rider): MAC-less zeroconf discovery records the
    uniqueness-less discovery and can't stack a duplicate prefill."""

    def _discovery(self, host="192.168.1.50", props=None):
        return SimpleNamespace(
            ip_address=host, host=host, properties=props or {}
        )

    async def test_mac_less_records_the_helper(self, tmp_path):
        module = _form_module()
        flow = module.NvidiaShieldConfigFlow()
        flow.hass = FakeHass(str(tmp_path))
        sys.modules[__name__]._zc_holder.discovery_without_unique_id_calls = 0

        result = await flow.async_step_zeroconf(self._discovery())

        assert result["type"] == "form"  # prefill still offered once
        assert (
            sys.modules[__name__]._zc_holder.discovery_without_unique_id_calls
            == 1
        )

    async def test_mac_less_rediscovery_does_not_stack(self, tmp_path):
        """A second MAC-less discovery of an already-configured host
        aborts — one-entry outcome."""
        module = _form_module()
        flow = module.NvidiaShieldConfigFlow()
        flow.hass = FakeHass(str(tmp_path))
        entry = SimpleNamespace(
            entry_id="e1",
            unique_id=None,  # MAC-less entry
            options={CONF_PROTOCOL_HOST: "192.168.1.50"},
            title="Shield",
        )
        sys.modules[__name__]._zc_holder.entries[:] = [entry]

        result = await flow.async_step_zeroconf(self._discovery())

        assert result == {"type": "abort", "reason": "already_configured"}
        # the entries list is untouched: one-entry outcome
        assert len(sys.modules[__name__]._zc_holder.entries) == 1


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 == []  # update-only (G3a)

    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

        # both hosts already hold the discovered address: nothing changes
        entry = SimpleNamespace(
            unique_id="aa:bb:cc:dd:ee:ff",
            options={
                CONF_PROTOCOL_ENABLED: True,
                CONF_PROTOCOL_HOST: "192.168.1.50",
                "adb_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 == []

    async def test_blank_adb_host_self_heals_from_the_discovery_address(self, tmp_path):
        """F2: a blank (auto-detect) adb host follows a same-IP rediscovery."""
        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"}
        # only the adb host was filled (self-heal), protocol already current:
        update = hass.config_entries.updating_entry_calls[0][1]["options"]
        assert update["adb_host"] == "192.168.1.50"
        assert update["protocol_host"] == "192.168.1.50"
        assert hass.config_entries.reloaded_entry_ids == []  # update-only (G3a)

    async def test_deliberately_different_adb_host_is_never_clobbered(self, tmp_path):
        """F2: an explicit ADB override survives a protocol-host IP change."""
        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",  # stale (IP moved)
                "adb_host": "10.0.0.9",  # deliberate ADB override
            },
            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"}
        update = hass.config_entries.updating_entry_calls[0][1]["options"]
        assert update["protocol_host"] == "192.168.1.50"  # protocol followed
        assert update["adb_host"] == "10.0.0.9"  # override untouched
        assert hass.config_entries.reloaded_entry_ids == []  # update-only (G3a)


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()
        flow.config_entry = entry  # t29 G3b: HA assigns config_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_serves_when_no_higher_tier_matches(self):
        """P5-1: the entity tier is below map/manifest — it serves when the
        package has no explicit override and no manifest label yet."""
        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"] = {}  # no override
        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)

        # no cached label yet → falls through to the entity tier
        assert player.app_name == "ATV Tivi"