"""t10 tests: command routing, play_media channel port, remote entity, volume.

Concrete fakes expose the ordering precisely: protocol → ADB keyevent →
legacy external entity, plus the §7 channel digit sequence (protocol digits,
~0.1 s apart, no ENTER, new sequence cancels in-flight). A scripted media
layer marks which tier served each command.
"""

import asyncio
import sys

import entity_stublib
import pytest
from entity_stublib import ENTITIES_STUB_NAMES, install, load_entity_modules
from nsh.connection import ShieldConnectionManager
from nsh.const import TIVIMATE_PACKAGE


@pytest.fixture(autouse=True)
def scoped_entity_stubs():
    """Container for the HA stubs + loaded entity modules (t7 discipline)."""
    snapshot = {name: sys.modules.get(name) for name in ENTITIES_STUB_NAMES}
    install()
    yield load_entity_modules()
    for name in ENTITIES_STUB_NAMES:
        original = snapshot[name]
        if original is None:
            sys.modules.pop(name, None)
        else:
            sys.modules[name] = original


class FakeProtocolClient:
    """Records key/launch calls; optional scripted failures."""

    def __init__(self):
        self.key_calls: list[tuple[str, str]] = []
        self.launch_calls: list[str] = []
        self.fail_keys: set[str] = set()
        self.fail_all = False

    def send_key_command(self, key, direction="SHORT"):
        self.key_calls.append((key, direction))
        if self.fail_all or key.upper() in self.fail_keys:
            raise ConnectionError("protocol transport down")

    def send_launch_app_command(self, app_link_or_app_id: str) -> None:
        self.launch_calls.append(app_link_or_app_id)
        if self.fail_all:
            raise ConnectionError("transport down")


class FakeAdbPoller:
    """Minimal ADB surface for keyevent and digit fallback paths."""

    def __init__(self):
        self.media_calls: list[str] = []
        self.digits: list[str] = []
        self.enter_calls: list[bool] = []
        self.results: dict[str, bool] = {}
        self.digit_result = True

    @property
    def connected(self):
        return True

    async def send_media_command(self, command: str) -> bool:
        self.media_calls.append(command)
        return self.results.get(command, True)

    async def send_digit_keyevent(self, digit: str) -> bool:
        self.digits.append(digit)
        return self.digit_result

    async def send_enter_keyevent(self) -> bool:
        self.enter_calls.append(True)
        return True

    async def bring_to_foreground(self, package: str) -> bool:
        return True


def make_manager(store=None, poller=None, protocol=None):
    return ShieldConnectionManager(
        hass=None,
        entry=None,
        poller=poller or FakeAdbPoller(),
        protocol=protocol or FakeProtocolClient(),
        store=store,
    )


@pytest.fixture(name="modules")
def load_modules_fixture():
    return load_entity_modules()


def make_player(modules, manager=None, jellyfin_entity=None, options=None):
    """A pre-initialised media_player entity bound to the tested manager."""
    from types import SimpleNamespace

    mp_module = modules["media_player"]
    if manager is None:
        manager = make_manager()
    entry = SimpleNamespace(options={}, data={}, title="mdarcy")
    manager.entry = entry

    def _set_option(key, value):
        entry.options[key] = value

    entry.set_option = _set_option
    player = mp_module.NvidiaShieldMediaPlayer.__new__(mp_module.NvidiaShieldMediaPlayer)
    player.hass = None
    player._entry = entry
    player._manager = manager
    player._intent = None
    player._last_channel = None
    player.hass = entity_stublib.FakeHass()
    player._entity_tasks = set()
    player._channel_task = None
    player.async_write_ha_state = lambda: None
    player._read = lambda eid: None

    served = []

    async def _cmd(service, **kwargs):
        served.append(("legacy", service))
        return None

    player._cmd = _cmd

    captured_intents = []

    def _set_intent(service, target):
        intent = SimpleNamespace(
            target_state=getattr(target, "value", target),
            command=service,
            created_at=0.0,
            prior_state=None,
            retry_count=0,
            via_adb=False,
            via_protocol=False,
            adb_command=None,
            protocol_key=None,
        )

        def _unsub():
            return None

        player._intent = intent
        captured_intents.append(intent)
        return None

    player._set_intent = _set_intent
    return player, served, captured_intents


class TestMediaCommandRouting:
    async def test_protocol_serves_media_command(self, modules):
        player, served, _intents = make_player(modules)
        protocol = player._manager._protocol
        player._intent = None

        player.async_write_ha_state = lambda: None

        await player._media_cmd("media_play", "play", "playing")

        assert protocol.key_calls == [("KEYCODE_MEDIA_PLAY", "SHORT")]
        assert served == []

    async def test_intent_records_protocol_path_for_retries(self, modules):
        player, _served, _intents = make_player(modules)
        player._set_intent("media_play", type("S", (), {"value": "playing"})())
        player.async_write_ha_state = lambda: None
        await player._media_cmd("media_play", "play", "playing")
        # retry metadata stamped: the protocol path is recorded on the intent
        assert player._intent is not None
        assert player._intent.via_protocol is True
        assert player._intent.protocol_key == "KEYCODE_MEDIA_PLAY"


class TestChannelDigits:
    async def test_digit_sequence_digit_part_delivery(self, modules):
        player, served, _intents = make_player(modules)
        protocol = player._manager._protocol

        await player._channel_entry_sequence(["4", "0", "9", "6"])

        # protocol digits: KEYCODE_{digit} SHORT each
        assert protocol.key_calls == [
            ("KEYCODE_4", "SHORT"),
            ("KEYCODE_0", "SHORT"),
            ("KEYCODE_9", "SHORT"),
            ("KEYCODE_6", "SHORT"),
        ]
        assert served == []  # never fell through to legacy

    async def test_new_channel_cancels_inflight_sequence(self, modules):
        player, _served, _intents = make_player(modules)
        protocol = player._manager._protocol

        # start a long sequence (real path), then a new play_media replaces it
        first_task = player._track_task(player._channel_entry_sequence(["1", "2", "3"]))
        player._channel_task = first_task  # as async_play_media tracks it

        await player.async_play_media("channel", "98")
        await asyncio.sleep(0.02)

        assert first_task.cancelled()
        assert "KEYCODE_2" not in [k for k, _d in protocol.key_calls]
        player._cancel_inflight_channel()  # test cleanup

    async def test_fallback_to_adb_digits_when_protocol_down(self, modules):
        player, served, _intents = make_player(modules)
        protocol = player._manager._protocol
        protocol.fail_all = True

        await player._channel_entry_sequence(["5"])

        assert protocol.key_calls == [("KEYCODE_5", "SHORT")]
        assert player._manager._poller.digits == ["5"]  # ADB digit keyevent served
        assert served == []

    async def test_optional_enter_falls_back_to_adb_keyevent(self, modules):
        player, served, _i = make_player(modules)
        protocol = player._manager._protocol
        poller = player._manager._poller
        protocol.fail_keys.add("KEYCODE_ENTER")
        protocol.key_calls.clear()

        player._entry.options["channel_send_enter"] = True  # §7 optional ENTER on

        await player._channel_entry_sequence(["4", "1"])  # digits fine, ENTER fails

        # digits served by protocol; ENTER retried over ADB keyevent 66
        assert ("KEYCODE_4", "SHORT") in protocol.key_calls
        assert poller.digits == []  # digits were fine over the protocol
        assert poller.enter_calls == [True]
        assert served == []

    async def test_legacy_fallback_when_both_transports_down(self, modules):
        player, served, _intents = make_player(modules)
        protocol = player._manager._protocol
        protocol.fail_all = True
        player._manager._poller.digit_result = False

        await player._channel_entry_sequence(["7"])

        assert served == [("legacy", "play_media")]  # transitional path used


class TestAppAndUrls:
    async def test_app_launch_calls_protocol(self, modules):
        player, served, _intents = make_player(modules)
        protocol = player._manager._protocol

        await player.async_play_media("app", "com.plex.hdx")

        assert protocol.launch_calls == ["com.plex.hdx"]  # C-LAUNCH auto-prefix
        assert served == []

    async def test_url_launch_passes_through_unchanged(self, modules):
        player, served, _intents = make_player(modules)
        protocol = player._manager._protocol

        await player.async_play_media("url", "plex://open")

        assert protocol.launch_calls == ["plex://open"]

    async def test_unknown_media_type_no_send(self, modules):
        player, served, _intents = make_player(modules)
        protocol = player._manager._protocol

        await player.async_play_media("spotify_uri", "spotify:track:x")

        assert protocol.launch_calls == []
        assert protocol.key_calls == []
        assert served == []


class TestRemoteEntity:
    async def test_split_command_prefixes(self, modules):
        module = modules["remote"]
        assert module._split_command("SHORT:home") == ("home", "SHORT")
        assert module._split_command("START_LONG:home") == ("home", "START_LONG")
        assert module._split_command("END_LONG:home") == ("home", "END_LONG")
        assert module._split_command("home") == ("home", "SHORT")

    async def test_key_code_name_mapping(self, modules):
        module = modules["remote"]
        assert module._key_code_name("home") == "KEYCODE_HOME"
        assert module._key_code_name("KEYCODE_HOME") == "KEYCODE_HOME"
        assert module._key_code_name("KEYCODE_HDMI_1") == "KEYCODE_HDMI_1"
        assert module._key_code_name("play") == "KEYCODE_MEDIA_PLAY"

    async def test_send_command_hits_protocol(self, modules):
        entity = _remote_entity(modules)
        protocol = entity._manager_of()._protocol

        await entity.async_send_command(["HOME"])

        assert protocol.key_calls == [("KEYCODE_HOME", "SHORT")]

    async def test_send_command_honors_hold_secs(self, modules):
        entity = _remote_entity(modules)
        protocol = entity._manager_of()._protocol

        await entity.async_send_command(["home"], hold_secs=0.05)

        assert protocol.key_calls == [
            ("KEYCODE_HOME", "START_LONG"),
            ("KEYCODE_HOME", "END_LONG"),
        ]

    async def test_send_command_honors_num_repeats(self, modules):
        entity = _remote_entity(modules)
        protocol = entity._manager_of()._protocol

        await entity.async_send_command(["home"], num_repeats=3)

        assert len(protocol.key_calls) == 3

    async def test_turn_on_off_send_power_key_respecting_is_on(self, modules):
        entity = _remote_entity(modules)
        manager = entity._manager_of()
        protocol = manager._protocol

        # push "already off" — turn_on should send POWER, turn_off should not
        manager._on_is_on(False)
        await entity.async_turn_on()
        assert protocol.key_calls == [("KEYCODE_POWER", "SHORT")]

        protocol.key_calls.clear()
        await entity.async_turn_off()
        assert protocol.key_calls == []

        # push "on" — turn_off should send POWER, turn_on should not
        manager._on_is_on(True)
        protocol.key_calls.clear()
        await entity.async_turn_off()
        assert protocol.key_calls == [("KEYCODE_POWER", "SHORT")]
        protocol.key_calls.clear()
        await entity.async_turn_on()
        assert protocol.key_calls == []


class TestAppFallback:
    async def test_app_id_prefers_the_androidtv_entity(self, modules):
        player, _s, _i = make_player(modules)
        player._androidtv_entity = "media_player.atv"
        from types import SimpleNamespace as FakeState

        state = FakeState(attributes={"app_id": "com.plex.hdx", "app_name": "Plex"})
        states = FakeState(states=FakeState(
            get=lambda eid: state if eid == "media_player.atv" else None
        ))
        player.hass = states
        player._read = lambda eid: state if eid == "media_player.atv" else None

        assert player.app_id == "com.plex.hdx"
        assert player.app_name == "Plex"

    async def test_app_falls_back_to_protocol_current_app(self, modules):
        player, _s, _i = make_player(modules)
        player._androidtv_entity = ""  # standalone setup (P3-1)
        player._manager._on_current_app("com.netflix.ninja")
        await asyncio.sleep(0.05)  # drain the call_soon_threadsafe push

        assert player.app_id == "com.netflix.ninja"
        assert player.app_name == "com.netflix.ninja"  # package is truth; map later

    async def test_app_unavailable_entity_falls_back(self, modules):
        player, _s, _i = make_player(modules)
        player._androidtv_entity = "media_player.atv"
        from types import SimpleNamespace as FakeState

        player.hass = FakeState(states=FakeState(get=lambda eid: None))  # entity gone
        player._read = lambda eid: None
        player._manager._on_current_app("com.google.android.youtube.tv")
        await asyncio.sleep(0.05)

        assert player.app_id == "com.google.android.youtube.tv"


def _ladder_player(modules, foreground=None, audio=None):
    """Standalone player with a foreground (and optional audio hint) set."""
    player, _s, _i = make_player(modules)
    manager = player._manager
    manager.store.set("awake", True)
    if foreground:
        manager.store.set("foreground", foreground)
    if audio:
        manager.store.set("audio_playback", audio)
    player._androidtv_entity = ""
    player._read = lambda eid: None
    return player, manager


class TestForegroundLadderTier:
    """P3-3a: the ladder branch between the media-session tier and ON."""

    def _player(self, modules, sessions=None, foreground=None, audio=None):
        player, _s, _i = make_player(modules)
        manager = player._manager
        manager.store.set("awake", True)
        if sessions:
            manager.store.set("sessions", sessions)
        if foreground:
            manager.store.set("foreground", foreground)
        if audio:
            manager.store.set("audio_playback", audio)
        player._androidtv_entity = ""
        player._read = lambda eid: None
        return player, manager

    async def test_session_app_serves_playing_via_session_tier(self, modules):
        from nsh.ha_adb_common import MediaSessionInfo

        sessions = [
            MediaSessionInfo(package="com.netflix.ninja", playing=True, title="Movie")
        ]
        player, manager = self._player(
            modules, sessions=sessions, foreground="com.netflix.ninja"
        )

        assert player.state.value == "playing"  # session tier, not the fg tier
        np = player._get_adb_now_playing()
        assert np["source"] == "media_session"
        assert np["title"] == "Movie"

    async def test_no_session_app_with_audio_hint(self, modules):
        player, _m = self._player(
            modules,
            foreground="ar.tvplayer.tv",
            audio=["ar.tvplayer.tv"],
        )

        # TiviMate publishes no session (R2) — fg tier keeps the device ON
        assert player.state.value == "on"
        np = player._get_app_now_playing()
        assert np == {
            "state": "idle",
            "title": "Media recently active",
            "source": "app",
        }

    async def test_no_session_app_without_hint_returns_none(self, modules):
        player, _m = self._player(modules, foreground="com.spocky.projengmenu")

        assert player.state.value == "on"  # launcher foreground → still ON
        assert player._get_app_now_playing() is None

    async def test_last_known_title_beats_audio_hint(self, modules):
        from nsh.ha_adb_common import MediaSessionInfo

        player, manager = self._player(
            modules, foreground="ar.tvplayer.tv", audio=["ar.tvplayer.tv"]
        )
        manager._clock = lambda: 0.0  # stamp 0.0, clock 0.0 → inside TTL
        info = MediaSessionInfo(package="ar.tvplayer.tv", playing=True, title="Sky News")
        manager._last_known["ar.tvplayer.tv"] = (info, 0.0)

        np = player._get_app_now_playing()
        assert np == {"state": "playing", "title": "Sky News", "source": "app"}

    async def test_ttl_resurrect_guard_in_the_tier(self, modules):
        from nsh.ha_adb_common import MediaSessionInfo

        now = 1000.0
        player, manager = self._player(modules, foreground="ar.tvplayer.tv")
        player._last_channel = "101"

        info = MediaSessionInfo(package="ar.tvplayer.tv", playing=True, title="Sky News")
        manager._clock = lambda: now  # the closure reads the live `now` cell
        manager._last_known["ar.tvplayer.tv"] = (info, now)

        # inside TTL → last-known title served
        np = player._get_app_now_playing()
        assert np == {"state": "playing", "title": "Sky News", "source": "app"}

        # past STATE_STORE_TTL → 'playing' must NOT be resurrected;
        # the tier falls back to the last channel instead
        now = now + 61.0
        np = player._get_app_now_playing()
        assert np == {"state": "idle", "title": "101", "source": "app"}


class TestChannelLeakGuard:
    """F1: the channel number never leaks to a non-channel foreground app."""

    async def test_launcher_never_shows_the_channel_number(self, modules):
        player, _m = _ladder_player(
            modules, foreground="com.spocky.projengmenu"
        )
        player._last_channel = "101"  # entered in TiviMate earlier

        np = player._get_app_now_playing()

        assert np is None  # the leak is gone

    async def test_channel_app_still_gets_the_channel_title(self, modules):
        player, _m = _ladder_player(modules, foreground=TIVIMATE_PACKAGE)

        player._last_channel = "101"
        np = player._get_app_now_playing()

        assert np == {"state": "idle", "title": "101", "source": "app"}

    async def test_audio_member_can_carry_the_channel_title(self, modules):
        player, manager = _ladder_player(
            modules, foreground="ar.tvplayer.tv", audio=["ar.tvplayer.tv"]
        )

        player._last_channel = "205"
        np = player._get_app_now_playing()

        assert np["title"] == "205"


class TestPrivacyOrdering:
    """F3: privacy mode outranks the channel fallback — identity stays the app."""

    async def test_channel_number_never_leaks_in_privacy_mode(self, modules):
        player, manager = _ladder_player(modules, foreground=TIVIMATE_PACKAGE)
        player._entry.options["adb_store_app_info"] = False  # privacy on

        player._last_channel = "101"
        np = player._get_app_now_playing()

        assert np["title"] == TIVIMATE_PACKAGE  # app identity, not the channel
        assert np["title"] != "101"

    async def test_channel_number_never_leaks_in_privacy_mode_with_last_known(self, modules):
        from nsh.ha_adb_common import MediaSessionInfo

        player, manager = _ladder_player(modules, foreground=TIVIMATE_PACKAGE)
        player._entry.options["adb_store_app_info"] = False
        manager._clock = lambda: 0.0
        info = MediaSessionInfo(
            package=TIVIMATE_PACKAGE, playing=True, title="Sky News"
        )
        manager._last_known[TIVIMATE_PACKAGE] = (info, 0.0)

        player._last_channel = "101"
        np = player._get_app_now_playing()

        assert np["state"] == "playing"  # playback truth survives privacy
        assert np["title"] == TIVIMATE_PACKAGE  # but identity stays the app


class TestVolumeAndPowerLadder:
    async def test_volume_readout_from_protocol_push(self, modules):
        player, served, _intents = make_player(modules)
        manager = player._manager
        manager._on_volume_info({"level": 7, "max": 15, "muted": True})

        assert player.volume_level == pytest.approx(7 / 15)
        assert player.is_volume_muted is True

    async def test_volume_set_is_not_claimed(self, modules):
        """C-VOL: VOLUME_SET stays out of supported features (readout only)."""
        from nsh.media_player import _BASE_FEATURES  # noqa: PLC0415

        assert not (_BASE_FEATURES & (1 << 5))  # VOLUME_SET bit off

    async def test_is_on_ladder_protocol_first(self, modules):
        player, _served, _intents = make_player(modules)
        manager = player._manager

        # no protocol push, no ADB wake data: fall to the plug/other logic
        assert manager.is_on is None

        manager._on_is_on(True)
        assert player._is_power_on() is True

        manager._on_is_on(False)
        assert player._is_power_on() is False

    async def test_adb_wake_used_when_protocol_silent(self, modules):
        player, _served, _intents = make_player(modules)
        manager = player._manager
        manager.store.set("awake", False)
        # protocol silent (is_on None) → ADB wake is the decision
        assert manager.is_on is None
        assert player._is_power_on() is False


# ── shared helpers ──


def _remote_entity(modules):
    from types import SimpleNamespace

    remote_module = modules["remote"]
    entry = SimpleNamespace(
        options={},
        data={},
        title="mdarcy",
        entry_id="e1",
    )
    manager = make_manager()
    entry.runtime_data = manager
    entity = remote_module.NvidiaShieldRemoteEntity.__new__(
        remote_module.NvidiaShieldRemoteEntity
    )
    entity.hass = None
    entity._entry = entry
    entity._manager = manager
    entity._attr_unique_id = None
    entity._manager_unregistered = None
    return entity


async def _noop_async(*args, **kwargs):
    return None


class FakeServices:
    """hass.services stand-in recording registrations."""

    def __init__(self):
        self.registered: dict[tuple, object] = {}
        self.service_calls: list[tuple] = []

    def has_service(self, domain, service):
        return (domain, service) in self.registered

    def async_register(self, domain, service, func, schema=None):
        self.registered[(domain, service)] = (func, schema)

    def async_remove(self, domain, service):
        self.registered.pop((domain, service), None)


class ServiceHarness:
    """FakeHass extension: `.services` + direct service invocation."""

    def attach(self, player):
        player.hass.services = FakeServices()


class TestAdbCommandService:
    """P4-2a: option-gated raw shell service on the entry."""

    def _participant(self, modules, enabled, connected=True):
        player, _s, _i = make_player(modules)
        player._entry.options["adb_command_enabled"] = enabled
        if not connected:
            player._manager._poller = None
        player.entity_id = "media_player.shield_test"
        player._attr_unique_id = "uid_shield"
        kind = entity_stublib.FakeHass()
        player.hass = kind
        kind.services = FakeServices()
        return player, kind.services

    async def test_option_off_does_not_participate(self, modules):
        from nsh.media_player import _ADB_COMMAND_TARGETS

        before = dict(_ADB_COMMAND_TARGETS)
        player, _services = self._participant(modules, enabled=False)
        # the add hook consults the entry option — off → no registration
        if player._entry.options.get("adb_command_enabled", False):
            _ADB_COMMAND_TARGETS.setdefault(player.entity_id, player._manager)
        assert before == _ADB_COMMAND_TARGETS

    async def test_option_on_registers_and_runs(self, modules):
        player, services = self._participant(modules, enabled=True)

        from nsh.media_player import _ADB_COMMAND_TARGETS, _async_adb_command_service
        _ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager

        class Call:
            data = {"command": "dumpsys power | head -5"}

        shell_calls = []

        async def _shell(cmd):
            shell_calls.append(cmd)
            return "ok"

        player._manager._poller.shell = _shell
        await _async_adb_command_service(Call())

        assert shell_calls == ["dumpsys power | head -5"]
        _ADB_COMMAND_TARGETS.pop("media_player.shield_test", None)

    async def test_rejects_when_adb_unavailable(self, modules):
        player, _s = self._participant(modules, enabled=True, connected=False)

        from nsh.media_player import _ADB_COMMAND_TARGETS, _async_adb_command_service
        _ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager

        class Call:
            data = {"command": "echo hi"}

        # no poller → the command is not sent and nothing raises
        player._manager._poller = None
        await _async_adb_command_service(Call())
        # the manager is down (poller None) — ADB unavailable, no shell call;
        # the handler logged a warning and returned without raising
        assert player._manager._poller is None
        _ADB_COMMAND_TARGETS.pop("media_player.shield_test", None)

    async def test_unregister_on_last_participant_unload(self, modules):
        from nsh.media_player import (
            _ADB_COMMAND_TARGETS,
            _adb_command_service_ensure,
            _adb_command_service_maybe_remove,
        )

        mods = load_entity_modules()
        player, _s, _i = make_player(mods)
        player._entry.options["adb_command_enabled"] = True
        player.entity_id = "media_player.shield_test"
        player.hass.services = FakeServices()

        _adb_command_service_ensure(player.hass)
        _ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager
        assert player.hass.services.has_service("nvidia_shield", "adb_command")

        _ADB_COMMAND_TARGETS.pop("media_player.shield_test")
        _adb_command_service_maybe_remove(player.hass)
        assert not player.hass.services.has_service("nvidia_shield", "adb_command")


class TestChannelDigitDelayConsumed:
    """t26 F3: a non-default digit delay reaches the sequence's sleep."""

    async def test_non_default_delay_reaches_the_sequence_sleep(self, modules):
        import asyncio

        player, _s = _ladder_player(modules)
        player._entry.options["channel_digit_delay"] = 0.3  # non-default

        sleeps: list[float] = []
        real_sleep = asyncio.sleep

        mp_module = sys.modules["nsh.media_player"]
        real_module_sleep = mp_module.asyncio.sleep

        async def spying_sleep(delay, *args):
            sleeps.append(delay)
            await real_sleep(0)

        mp_module.asyncio.sleep = spying_sleep
        try:
            await player.async_play_media("channel", "123")
            if player._channel_task is not None:
                await player._channel_task  # the sequence runs to completion
        finally:
            mp_module.asyncio.sleep = real_module_sleep

        # three digits → three sleeps, each with the option's 0.3 s
        assert sleeps == [0.3, 0.3, 0.3]


class TestTextEntry:
    """P5-2: text entry — protocol IME first, ADB input-text fallback."""

    def _player_with_texts(self, modules):
        player, _s, _i = make_player(modules)
        protocol = player._manager._protocol
        protocol.text_calls = []
        protocol.send_text = lambda t: protocol.text_calls.append(t) or True
        shell_texts = []

        async def _shell(cmd):
            shell_texts.append(cmd)
            return "ok"

        player._manager._poller.shell = _shell
        return player, protocol, shell_texts

    async def test_protocol_ime_sends_first(self, modules):
        player, protocol, shell_texts = self._player_with_texts(modules)

        await player.async_play_media("text", "hello world")

        assert protocol.text_calls == ["hello world"]
        assert shell_texts == []  # ADB fallback not used

    async def test_adb_fallback_when_the_protocol_send_fails(self, modules):
        player, protocol, shell_texts = self._player_with_texts(modules)

        def _fail(text):
            protocol.text_calls.append(text)
            raise RuntimeError("ime closed")

        protocol.send_text = _fail
        await player.async_play_media("text", "hello 'world'")

        import shlex

        sample = "hello " + chr(39) + "world" + chr(39)
        assert shell_texts == [f"input text {shlex.quote(sample)}"]

    async def test_empty_text_is_ignored(self, modules):
        player, protocol, shell_texts = self._player_with_texts(modules)

        await player.async_play_media("text", "")

        assert protocol.text_calls == [] and shell_texts == []
        player._manager._protocol.send_text = lambda t: True
        await player.async_play_media("text", "")
        assert protocol.text_calls == []


class TestIconOverride:
    """P5-2: optional icon column → CONF_APP_ICON_MAP → entity_picture."""

    def _player_with_icon(self, modules, icon_map):
        player, _s, _i = make_player(modules)
        player._entry.options["app_icon_map"] = icon_map
        return player

    async def test_icon_serves_when_no_jellyfin_picture(self, modules):
        import asyncio

        player = self._player_with_icon(
            modules, {"ar.tvplayer.tv": "http://icons.local/tivi.png"}
        )
        manager = player._manager
        manager._on_current_app("ar.tvplayer.tv")
        await asyncio.sleep(0.05)

        assert player.media_image_url == "http://icons.local/tivi.png"

    async def test_no_icon_for_an_unmatched_package(self, modules):
        import asyncio

        player = self._player_with_icon(
            modules, {"com.netflix.ninja": "http://icons.local/netflix.png"}
        )
        player._manager._on_current_app("ar.tvplayer.tv")
        await asyncio.sleep(0.05)

        assert player.media_image_url is None

    async def test_jellyfin_picture_wins_over_the_icon_map(self, modules):

        player = self._player_with_icon(
            modules, {"ar.tvplayer.tv": "http://icons.local/tivi.png"}
        )
        player._jf_active = lambda: True
        player._jf_attr = lambda attr: "http://jf.local/pic.jpg"

        assert player.media_image_url == "http://jf.local/pic.jpg"


class TestServiceRegistryKey:
    """F4: the participant key is computed once — add/remove always match."""

    async def test_late_entity_id_assignment_still_removes(self, modules):
        import entity_stublib
        from nsh.media_player import (
            _ADB_COMMAND_TARGETS,
            _adb_command_service_ensure,
            _adb_command_service_maybe_remove,
        )

        mods = load_entity_modules()
        player, _s, _i = make_player(mods)
        player._entry.options["adb_command_enabled"] = True
        player._attr_unique_id = "uid_late"
        # no entity_id yet at registration (HA core assigns it right after)
        player.hass = entity_stublib.FakeHass()
        player.hass.services = FakeServices()
        if hasattr(player, "_adb_command_key"):
            del player._adb_command_key  # start clean

        # add path — the key is computed once from unique_id
        key = (
            getattr(player, "entity_id", None) or player._attr_unique_id
        )
        player._adb_command_key = key
        _ADB_COMMAND_TARGETS[key] = player._manager
        _adb_command_service_ensure(player.hass)
        assert "uid_late" in _ADB_COMMAND_TARGETS

        # HA core assigned the entity_id afterwards — removal must still match
        player.entity_id = "media_player.shield_test"
        _adb_command_service_maybe_remove(player.hass)
        assert "uid_late" in _ADB_COMMAND_TARGETS  # service stays until unload

        # removal uses the stored key (does not consult the new entity_id)
        _ADB_COMMAND_TARGETS.pop(player._adb_command_key)
        _adb_command_service_maybe_remove(player.hass)
        assert not player.hass.services.has_service("nvidia_shield", "adb_command")


class TestEntryDiagnostics:
    """P4-2b: redacted entry diagnostics with store + transport health."""

    async def test_sensitive_keys_redacted_and_health_present(self, modules):
        import entity_stublib

        mods = entity_stublib.load_entity_modules()
        player, _s, _i = make_player(mods)
        from types import SimpleNamespace as DiagEntry

        entry = DiagEntry(
            data={},
            options={
                "adb_command_enabled": True,
                "protocol_pairing_code": "4821",
            },
            title="Shield",
        )
        manager = player._manager
        manager.store.set("awake", True)
        entry.runtime_data = manager

        from nsh.diagnostics import async_get_config_entry_diagnostics

        payload = await async_get_config_entry_diagnostics(None, entry)

        assert payload["entry"]["options"]["protocol_pairing_code"] == "**REDACTED**"
        assert payload["entry"]["options"]["adb_command_enabled"] is True
        assert "awake" in payload["store"]
        assert payload["transport"]["adb"]["connected"] is not None
        assert payload["transport"]["protocol"]["active"] is not None


class TestBrowseMediaDirectory:
    """P4-2c: the app directory root and its launch path."""

    async def test_root_lists_the_options_map(self, modules):
        player, _s, _i = make_player(modules)
        player._entry.options["app_name_map"] = {
            "ar.tvplayer.tv": "TiviMate",
            "com.netflix.ninja": "Netflix",
        }

        root = await player.async_browse_media()

        assert root.title == "Apps"
        assert root.can_play is False and root.can_expand is True
        ids = [(c.media_content_id, c.title) for c in root.children]
        assert ids == [
            ("ar.tvplayer.tv", "TiviMate"),
            ("com.netflix.ninja", "Netflix"),
        ]
        assert all(c.media_content_type == "app" for c in root.children)

    async def test_root_falls_back_to_the_curated_default(self, modules):
        player, _s, _i = make_player(modules)  # no apps map configured

        root = await player.async_browse_media()

        ids = [c.media_content_id for c in root.children]
        assert "ar.tvplayer.tv" in ids  # TiviMate
        assert "com.github.damontecres.wholphin" in ids  # Wholphin
        assert "com.netflix.ninja" in ids  # Netflix
        assert "com.google.android.youtube.tv" in ids  # YouTube

    async def test_selection_launches_via_play_media_app(self, modules):
        player, _s, _i = make_player(modules)
        protocol = player._manager._protocol
        root = await player.async_browse_media()
        selected = root.children[0]

        await player.async_play_media("app", selected.media_content_id)

        assert protocol.launch_calls == [selected.media_content_id]  # C-LAUNCH