ha-nvidia-shield/tests/test_command_scripts.py

"""tests/test_command_scripts.py — per-command script overrides (t44).

The user's remaining universal-media-player dependency: some commands need hardware
this integration deliberately does not drive — a speaker switch that owns the
volume, an AVR input that has to move with the Shield, a mute that has to hit two
devices. One script per command is the escape hatch, and the rule is absolute:

  configured  -> that command calls the script, with its variables, and never
                 touches the protocol or ADB path;
  unset       -> the command behaves exactly as it did before this existed.

The suite is expected to stay quiet about the first case elsewhere: nothing else may
start calling services, which is why tests/test_command_routing.py asserts that no
service call happens on the native paths.
"""

from types import SimpleNamespace

import entity_stublib
import pytest
from nsh.const import (
    CONF_COMMAND_SCRIPTS,
    SCRIPT_VAR_APP_NAME,
    SCRIPT_VAR_LEVEL,
    SCRIPT_VAR_MUTED,
)
from test_command_routing import _remote_entity, make_manager, make_player

#: Commands with no data to pass — the script is called with the entity only.
PLAIN_COMMANDS = ("turn_on", "turn_off", "volume_up", "volume_down")

WATCH = "script.shield_watch"


@pytest.fixture(name="modules", autouse=True)
def scoped_modules():
    """Stub scope + loaded entity modules for one test, restored afterwards.

    The t7 discipline: without the teardown this file's `nsh.*` modules and stub
    faces survive into whichever file runs next, and the next file's flow/entity
    classes bind against the wrong faces (the t44 file-order failure).
    """
    snapshot = entity_stublib.snapshot_modules()
    entity_stublib.install()
    try:
        yield entity_stublib.load_entity_modules()
    finally:
        entity_stublib.restore_modules(snapshot)


def _services(served):
    """A hass.services stand-in that records (domain, service, data)."""

    class _RecordingServices:
        async def async_call(self, domain, service, data=None, **kwargs):
            served.append((domain, service, data))

    return _RecordingServices()


def _player(modules, scripts=None, *, is_on=None):
    player, served, _intents = make_player(modules)
    if is_on is not None:
        player._manager._on_is_on(is_on)
    if scripts is not None:
        player._entry.options[CONF_COMMAND_SCRIPTS] = scripts
    player.hass.services = _services(served)
    return player, served


def _script_call(script=WATCH, variables=None):
    data = {"entity_id": script}
    if variables:
        data["variables"] = variables
    return ("script", "turn_on", data)


class TestMappedCommands:
    """A configured script replaces the native path — for every command."""

    @pytest.mark.parametrize("command", PLAIN_COMMANDS)
    async def test_a_plain_command_calls_its_script_and_nothing_else(
        self, modules, command
    ):
        # every native path would have had work to do: ADB is up, and for the
        # power commands the state gate is open
        player, served = _player(modules, {command: WATCH}, is_on=command == "turn_off")

        await getattr(player, f"async_{command}")()

        assert served == [_script_call()]
        assert player._manager._protocol.key_calls == []
        assert player._manager._poller.media_calls == []

    async def test_volume_set_passes_the_level(self, modules):
        player, served = _player(modules, {"volume_set": WATCH})

        await player.async_set_volume_level(0.4)

        # exactly the variable name and the value as passed, nothing extra
        assert served == [_script_call(variables={SCRIPT_VAR_LEVEL: 0.4})]
        assert player._manager._protocol.key_calls == []
        assert player._manager._poller.media_calls == []

    @pytest.mark.parametrize("mute", [True, False])
    async def test_volume_mute_passes_the_requested_state(self, modules, mute):
        """A script cannot tell mute from unmute without this."""
        player, served = _player(modules, {"volume_mute": WATCH})

        await player.async_mute_volume(mute)

        assert served == [_script_call(variables={SCRIPT_VAR_MUTED: mute})]
        assert player._manager._protocol.key_calls == []
        assert player._manager._poller.media_calls == []

    async def test_select_source_passes_the_app_name(self, modules):
        player, served = _player(modules, {"select_source": WATCH})

        await player.async_select_source("TiviMate")

        assert served == [_script_call(variables={SCRIPT_VAR_APP_NAME: "TiviMate"})]

    async def test_the_remote_entity_shares_the_same_turn_on_script(self, modules):
        """There are two routes to POWER; one configuration governs both."""
        entity = _remote_entity(modules)
        served = []
        entity.hass = SimpleNamespace(services=_services(served))
        entity._entry.options[CONF_COMMAND_SCRIPTS] = {"turn_on": WATCH}
        entity._manager._on_is_on(False)  # the native path would fire

        await entity.async_turn_on()

        assert served == [_script_call()]
        assert entity._manager._protocol.key_calls == []


class TestUnmappedCommands:
    """With nothing configured, every command keeps its native path."""

    async def test_no_service_call_happens_at_all(self, modules):
        player, served = _player(modules, {}, is_on=False)

        await player.async_turn_on()
        await player.async_volume_up()

        assert served == []  # nothing left the integration

    async def test_turn_on_still_sends_the_protocol_power_key(self, modules):
        player, _served = _player(modules, {}, is_on=False)

        await player.async_turn_on()

        assert player._manager._protocol.key_calls == [("KEYCODE_POWER", "SHORT")]

    async def test_volume_up_still_uses_the_native_protocol_key(self, modules):
        from nsh.const import PROTOCOL_REMOTE_KEYS

        player, _served = _player(modules, {})

        await player.async_volume_up()

        assert player._manager._protocol.key_calls == [
            (PROTOCOL_REMOTE_KEYS["volume_up"], "SHORT")
        ]

    async def test_volume_up_still_falls_through_to_adb(self, modules):
        """The second native tier, reached exactly when the protocol is down."""
        player, _served = _player(modules, {})
        player._manager._protocol = None

        await player.async_volume_up()

        assert player._manager._poller.media_calls == ["volume_up"]

    async def test_volume_set_without_a_script_stays_a_no_op(self, modules):
        """Absolute volume exists ONLY through a script (the protocol has no
        VOLUME_SET key), so unconfigured it sends nothing — never a crash."""
        player, served = _player(modules, {})

        await player.async_set_volume_level(0.4)

        assert served == []
        assert player._manager._protocol.key_calls == []
        assert player._manager._poller.media_calls == []

    async def test_other_commands_are_unaffected_by_a_mapped_one(self, modules):
        from nsh.const import PROTOCOL_REMOTE_KEYS

        player, served = _player(modules, {"volume_up": WATCH})

        await player.async_volume_down()

        assert served == []
        assert player._manager._protocol.key_calls == [
            (PROTOCOL_REMOTE_KEYS["volume_down"], "SHORT")
        ]


class TestScriptFailures:
    """A mapped command that cannot run must be visible, never a silent no-op."""

    async def test_a_raising_script_surfaces_a_clean_error(self, modules):
        from homeassistant.exceptions import HomeAssistantError

        player, _served = _player(modules, {"volume_up": WATCH}, is_on=False)

        class _FailingServices:
            async def async_call(self, domain, service, data=None, **kwargs):
                raise RuntimeError("speaker switch is offline")

        player.hass.services = _FailingServices()

        with pytest.raises(HomeAssistantError) as raised:
            await player.async_volume_up()

        message = str(raised.value)
        assert "volume_up" in message
        assert WATCH in message
        assert "speaker switch is offline" in message
        # ...and it did NOT quietly fall back to the native path
        assert player._manager._protocol.key_calls == []
        assert player._manager._poller.media_calls == []


class TestOptionShape:
    """The stored option is read through one normalizer, defensively."""

    @pytest.mark.parametrize(
        "stored",
        [
            "script.watch",  # not a dict at all
            {"turn_on": 5},  # not an entity id
            {"turn_on": "   "},  # empty
            {"bogus_command": WATCH},  # unknown command
            None,
        ],
    )
    async def test_a_hand_edited_option_never_reaches_the_service_layer(
        self, modules, stored
    ):
        player, served = _player(modules, stored, is_on=False)

        await player.async_turn_on()

        assert served == []
        assert player._manager._protocol.key_calls == [("KEYCODE_POWER", "SHORT")]

    async def test_a_padded_entity_id_is_trimmed(self, modules):
        player, served = _player(modules, {"turn_on": f"  {WATCH}  "}, is_on=False)

        await player.async_turn_on()

        assert served == [_script_call()]


class TestFeatureSet:
    """A script-only command still has to be reachable, so it is advertised."""

    def _features(self, player):
        return int(player.supported_features)

    def test_volume_set_appears_only_when_configured(self, modules):
        """volume_set has no native path at all, so the script is the only reach."""
        from homeassistant.components.media_player import MediaPlayerEntityFeature

        player, _served = _player(modules, {})
        baseline = self._features(player)
        assert baseline & int(MediaPlayerEntityFeature.VOLUME_SET) == 0

        player._entry.options[CONF_COMMAND_SCRIPTS] = {"volume_set": WATCH}
        with_volume = self._features(player)

        assert with_volume & int(MediaPlayerEntityFeature.VOLUME_SET)
        # ...and nothing else moved
        assert with_volume & ~int(MediaPlayerEntityFeature.VOLUME_SET) == baseline

    def test_select_source_is_advertised_once_there_is_something_to_offer(
        self, modules
    ):
        """t47: either a script OR a source list is enough — the native path exists
        now, so an unconfigured integration still has sources to offer."""
        from homeassistant.components.media_player import MediaPlayerEntityFeature

        player, _served = _player(modules, {})
        flagged = self._features(player)

        assert flagged & int(MediaPlayerEntityFeature.SELECT_SOURCE)
        # a configured script changes nothing about the flag
        player._entry.options[CONF_COMMAND_SCRIPTS] = {"select_source": WATCH}
        assert self._features(player) & int(MediaPlayerEntityFeature.SELECT_SOURCE)

    def test_a_script_for_an_already_advertised_command_changes_nothing(self, modules):
        player, _served = _player(modules, {})
        baseline = self._features(player)

        player._entry.options[CONF_COMMAND_SCRIPTS] = {"turn_on": WATCH}

        assert self._features(player) == baseline


def test_the_remote_entry_options_are_the_media_player_options(modules):
    """Both entities read the same option key — pinned so they cannot drift."""
    player, _served = _player(modules, {"turn_on": WATCH})
    entity = _remote_entity(modules)
    entity._entry.options[CONF_COMMAND_SCRIPTS] = {"turn_on": WATCH}

    assert (
        player._entry.options[CONF_COMMAND_SCRIPTS]
        == entity._entry.options[CONF_COMMAND_SCRIPTS]
    )
    assert make_manager().entry is None  # the harness's manager starts unbound