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

import pytest
import voluptuous_mirror as vol_mirror
from conftest import _SHIELD_DIR, _load_module
from nsh.const import (
    CONF_ADB_ENABLED,
    CONF_ADB_PORT,
    CONF_ANDROIDTV_ENTITY,
    CONF_HOST,
    CONF_PROTOCOL_ENABLED,
)

_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",
)


class DiscoveryDedupAbortError(Exception):
    """Stands in for data_entry_flow.AbortFlow inside the discovery stub.

    HA's `ConfigFlow._async_handle_discovery_without_unique_id` RAISES on a
    duplicate (the flow manager turns that into an abort result); modelling it
    as a raise is what makes an un-awaited call fail loudly here.
    """

    def __init__(self, reason):
        super().__init__(reason)
        self.reason = reason


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")
    # t33 F-A: a faithful mirror — the default stays ON the key, so a schema
    # whose host default was dropped FAILS here instead of looking identical
    # to one that carries it (the old bare-key stub hid exactly that bug).
    vol_mirror.install(vol)
    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,
            description_placeholders=None,
        ):
            return {
                "type": "form",
                "step_id": step_id,
                "errors": dict(errors or {}),
                "data_schema": data_schema,
                "description_placeholders": dict(description_placeholders 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

        def _async_current_entries(self):
            return _zc_holder.entries

        def _async_in_progress(self, include_uninitialized=False, **kwargs):
            return _zc_holder.in_progress

        # t37 D3: HA's own helper, modelled faithfully — a COROUTINE that
        # dedups uniqueness-less discovery by raising. An un-awaited call
        # therefore does nothing at all, which is exactly the production bug
        # this replaces.
        async def _async_handle_discovery_without_unique_id(self):
            _zc_holder.discovery_without_unique_id_calls += 1
            if getattr(self, "_unique_id", None) is not None:
                return
            if self._async_current_entries():
                raise DiscoveryDedupAbortError("already_configured")
            await self.async_set_unique_id(_zc_holder.discovery_unique_id)
            self._abort_if_unique_id_configured()
            if self._async_in_progress(include_uninitialized=True):
                raise DiscoveryDedupAbortError("already_in_progress")

    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=[],
        in_progress=[],
        discovery_unique_id="default_discovery_unique_id",
        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 _record(sink):
    """An async_set_unique_id stand-in that records what the flow tried."""

    async def _set(unique_id):
        sink.append(unique_id)

    return _set


def _user_input(**overrides) -> dict:
    base = {
        "name": "mdarcy",
        CONF_PROTOCOL_ENABLED: False,
        CONF_HOST: "",
        CONF_ADB_ENABLED: False,
    }
    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())

        assert result["type"] == "form"
        assert result["step_id"] == "user"
        assert result["errors"] == {"base": "no_state_source"}

    async def test_the_androidtv_field_is_gone_from_the_form(self, tmp_path):
        """t41: we own the stack — no other integration is a state source."""
        module = _form_module()
        flow = module.NvidiaShieldConfigFlow()
        flow.hass = FakeHass(str(tmp_path))

        result = await flow.async_step_user(None)

        fields = vol_mirror.declared_fields(result["data_schema"])
        assert CONF_ANDROIDTV_ENTITY not in fields
        # the two native transports are what is left to choose from
        assert CONF_PROTOCOL_ENABLED in fields
        assert CONF_ADB_ENABLED in fields

    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_PROTOCOL_ENABLED: True})
        )

        assert result["type"] == "form"
        assert result["step_id"] == "protocol_pair"  # routed into pairing, no error

    async def test_entry_data_is_empty_so_options_own_the_config(self, tmp_path, monkeypatch):
        """t41: every value lives in options; `data` carries no leftovers."""
        module = _form_module()
        flow = module.NvidiaShieldConfigFlow()
        flow.hass = FakeHass(str(tmp_path))
        flow._user_data = {
            "name": "mdarcy",
            CONF_PROTOCOL_ENABLED: True,
            CONF_HOST: "shield.local",
        }

        # skip the pairing route: drive _create_entry directly
        result = flow._create_entry()

        assert result["type"] == "create_entry"
        assert 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_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(**{"adb_enabled": False})
        )

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


class TestMigrationDropsTheEntityKey:
    """t41: the androidtv key is dropped from BOTH dicts with a logged note.

    An entry that still carries it keeps working until then (nothing reads it) —
    the migration is what removes the last trace of the entity path.
    """

    async def test_v1_entry_reaches_v3_without_the_entity_key(self, tmp_path, caplog):
        module = _entry_module()
        hass = FakeHass(str(tmp_path))
        entry = SimpleNamespace(
            version=1,
            data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
            options={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
            title="mdarcy",
        )

        with caplog.at_level(logging.INFO):
            ok = await module.async_migrate_entry(hass, entry)

        assert ok is True
        assert CONF_ANDROIDTV_ENTITY not in entry.data
        assert CONF_ANDROIDTV_ENTITY not in entry.options
        assert entry.options == {CONF_PROTOCOL_ENABLED: False, CONF_HOST: ""}
        assert entry.data == {}
        # the note names what was dropped, so the log is actionable
        notes = [record.getMessage() for record in caplog.records]
        assert any("media_player.shield" in note for note in notes), notes
        assert hass.config_entries.updating_entry_calls[0][1]["version"] == 3

    async def test_an_entry_without_the_key_logs_plainly(self, tmp_path, caplog):
        module = _entry_module()
        hass = FakeHass(str(tmp_path))
        entry = SimpleNamespace(version=2, data={}, options={}, title="mdarcy")

        with caplog.at_level(logging.INFO):
            ok = await module.async_migrate_entry(hass, entry)

        assert ok is True
        notes = [record.getMessage() for record in caplog.records]
        assert any("Migrated mdarcy config entry to version 3" in note for note in notes), notes

    async def test_v2_entry_with_a_legacy_address_migrates_to_the_one_field(self, tmp_path):
        """An entry written before the unification lands on CONF_HOST."""
        module = _entry_module()
        hass = FakeHass(str(tmp_path))
        entry = SimpleNamespace(
            version=2,
            data={},
            options={CONF_PROTOCOL_ENABLED: True, "protocol_host": "shield.local"},
            title="mdarcy",
        )

        ok = await module.async_migrate_entry(hass, entry)

        assert ok is True
        assert entry.options[CONF_HOST] == "shield.local"
        assert "protocol_host" not in entry.options

    def test_the_stored_version_and_the_flow_cannot_drift(self, tmp_path):
        # _form_module owns the scoped stub install/pop — never a bare
        # entity_stublib.install() here: a shallow stub set leaks into the
        # entity-module loader of later tests.
        module = _form_module()
        from nsh.const import ENTRY_VERSION

        assert module.NvidiaShieldConfigFlow.VERSION == ENTRY_VERSION == 3


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 TestUniqueIdSource:
    """t41: no unique_id is derived from another integration, at any stage.

    t29 G2 replaced the renameable entity_id with a device-MAC derive; t41
    removes the entity path entirely, so the ONLY identity source left is the
    protocol MAC learned during pairing (``async_set_unique_id(mac)``).
    """

    def test_no_entity_derived_unique_id_remains(self):
        flow_src = (_SHIELD_DIR / "config_flow.py").read_text(encoding="utf-8")

        assert "_unique_id_from_atv_device" not in flow_src
        assert "async_set_unique_id(androidtv_entity)" not in flow_src
        # no registry lookups for another integration's devices either
        assert "device_registry" not in flow_src

    async def test_the_user_step_sets_no_unique_id(self, tmp_path):
        module = _form_module()
        flow = module.NvidiaShieldConfigFlow()
        flow.hass = FakeHass(str(tmp_path))
        flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
        recorded = []
        flow.async_set_unique_id = _record(recorded)

        await flow.async_step_user(_user_input(**{CONF_PROTOCOL_ENABLED: True}))

        assert recorded == []


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={},
            options={
                CONF_ADB_ENABLED: True,
                CONF_HOST: host,  # t41: the entry's one address
                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_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 {}
        )

    def _holder(self):
        return sys.modules[__name__]._zc_holder

    async def test_mac_less_discovery_awaits_the_helper(self, tmp_path):
        module = _form_module()
        flow = module.NvidiaShieldConfigFlow()
        flow.hass = FakeHass(str(tmp_path))
        self._holder().discovery_without_unique_id_calls = 0

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

        assert result["type"] == "form"  # prefill still offered once
        # t37 D3: only an AWAITED coroutine body runs — before the fix this
        # counter stayed at 0 and Python warned "coroutine ... was never
        # awaited", while HA's dedup never happened
        assert self._holder().discovery_without_unique_id_calls == 1
        # the helper claims HA's discovery sentinel for the flow
        assert flow._unique_id == self._holder().discovery_unique_id

    async def test_mac_less_discovery_with_entries_aborts(self, tmp_path):
        """Real contract: any entry for the handler → already_configured."""
        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_HOST: "192.168.1.50"},
            title="Shield",
        )
        self._holder().entries[:] = [entry]

        with pytest.raises(DiscoveryDedupAbortError) as raised:
            await flow.async_step_zeroconf(self._discovery())

        assert raised.value.reason == "already_configured"
        # the entries list is untouched: one-entry outcome
        assert len(self._holder().entries) == 1

    async def test_mac_less_discovery_while_another_flow_runs_aborts(self, tmp_path):
        """The other half of the real contract: one flow at a time."""
        module = _form_module()
        flow = module.NvidiaShieldConfigFlow()
        flow.hass = FakeHass(str(tmp_path))
        self._holder().in_progress[:] = [{"flow_id": "other"}]

        with pytest.raises(DiscoveryDedupAbortError) as raised:
            await flow.async_step_zeroconf(self._discovery())

        assert raised.value.reason == "already_in_progress"

    async def test_mac_less_discovery_leaves_no_un_awaited_coroutine(
        self, tmp_path
    ):
        """t37 D3: the helper must never be left un-awaited."""
        module = _form_module()
        flow = module.NvidiaShieldConfigFlow()
        flow.hass = FakeHass(str(tmp_path))

        with warnings.catch_warnings(record=True) as caught:
            warnings.simplefilter("always")
            await flow.async_step_zeroconf(self._discovery())
            del flow
            gc.collect()

        never_awaited = [
            str(w.message) for w in caught if "never awaited" in str(w.message)
        ]
        assert never_awaited == []


class TestSeedBlockShapeF4:
    """t32 F4 (rider): protocol-only seed blocks don't force ADB on."""

    def test_protocol_only_block_keeps_adb_off(self):
        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"]

        blocks = init_module._yaml_seed_blocks(
            {"name": "Proto", "protocol_host": "shield.local"}
        )

        assert blocks[0]["adb_enabled"] is False
        assert blocks[0]["host"] == "shield.local"  # t41: one address field

    def test_adb_block_keeps_adb_on(self):
        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"]

        blocks = init_module._yaml_seed_blocks(
            {"name": "Adb", "adb_host": "10.0.0.31"}
        )

        assert blocks[0]["adb_enabled"] is True

    async def test_protocol_only_import_prefills_adb_off(self):
        module = _form_module()
        flow = module.NvidiaShieldConfigFlow()
        flow.hass = FakeHass("/tmp/nsh-f4")
        flow._async_current_entries = lambda: []

        result = await flow.async_step_import(
            {"name": "Proto", "protocol_host": "shield.local"}
        )

        assert result["type"] == "form"
        assert flow._user_data["adb_enabled"] is False


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):
        """t33 F-A: the discovered host is the DEFAULT OF THE SCHEMA KEYS.

        The old check (the attribute was stored) stayed green while the
        rendered form was blank — this one reads the schema itself.
        """
        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"
        assert flow._discovered_host == "192.168.1.50"
        defaults = vol_mirror.defaults_by_key(result["data_schema"])
        assert defaults[CONF_HOST] == "192.168.1.50"
        # ...and a discovery-driven form says the address is already known
        note = result["description_placeholders"]["discovery_note"]
        assert "Discovered Shield at 192.168.1.50" in note

    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_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"}
        # t41: the ONE address field moved to the discovery address
        assert entry.options[CONF_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

        # the address already holds the discovered value: nothing changes
        entry = SimpleNamespace(
            unique_id="aa:bb:cc:dd:ee:ff",
            options={CONF_PROTOCOL_ENABLED: True, CONF_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_the_per_host_heal_cases_collapsed_into_one_rule(self):
        """t41: the blank / self-heal / deliberate-override cases are gone.

        They only existed because the address was stored twice (t21 F2). With one
        field the discovery address simply IS the entry's address, so this pins
        the simplification: a second address field cannot quietly come back.
        """
        source = (_SHIELD_DIR / "config_flow.py").read_text(encoding="utf-8")

        assert "CONF_ADB_HOST" not in source
        assert "CONF_PROTOCOL_HOST" not in source

    async def test_an_explicit_address_follows_the_device(self, tmp_path):
        """t41: one Shield, one address — discovery is its current source.

        The old rule kept a "deliberately different ADB host" untouched; that
        override is precisely what the single-field ruling removes.
        """
        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_HOST: "10.0.0.9"},  # stale
            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[CONF_HOST] == "192.168.1.50"
        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"}

        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"}

        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_cached_manifest_label_is_the_second_tier(self):
        """P5-1: below the explicit map sits the device manifest label."""
        import entity_stublib

        modules = entity_stublib.load_entity_modules()
        player, _s, _i = _cmd_player(modules)
        player._entry.options["app_name_map"] = {}  # no override
        manager = player._manager
        manager._app_labels["ar.tvplayer.tv"] = ("TiviMate", manager._clock())

        manager._on_current_app("ar.tvplayer.tv")
        await asyncio.sleep(0.05)

        assert player.app_name == "TiviMate"