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_ADB_ENABLED,
    CONF_COMMAND_SCRIPTS,
    CONF_REMOTE_COMMAND_SCRIPT,
    SCRIPT_VAR_APP_NAME,
    SCRIPT_VAR_COMMAND,
    SCRIPT_VAR_DELAY_SECS,
    SCRIPT_VAR_HOLD_SECS,
    SCRIPT_VAR_LEVEL,
    SCRIPT_VAR_MUTED,
    SCRIPT_VAR_NUM_REPEATS,
)
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 ServiceValidationError

        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(ServiceValidationError) as raised:
            await player.async_volume_up()

        # translated (t47): the command, the script and the cause are placeholders
        assert raised.value.translation_domain == "nvidia_shield"
        assert raised.value.translation_key == "command_script_failed"
        assert raised.value.translation_placeholders == {
            "action": "volume_up",
            "script": WATCH,
            "error": "speaker switch is offline",
        }
        # ...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]
    )
    # the harness's manager carries a realistic entry whose option says ADB is on
    assert make_manager().entry.options[CONF_ADB_ENABLED] is True


WATCH_SCRIPT = "script.shield_remote"


class TestRemoteCommandScript:
    """t47 B: one optional script for remote.send_command, replacing the key path.

    A remote command is a SEQUENCE (repeats, long presses, delays), so the escape
    hatch is one script for the whole call rather than a per-key override: an IR
    bridge, a macro or a second device needs the call, not the keys.
    """

    def _entity(self, modules, script=None):
        entity = _remote_entity(modules)
        served = []
        entity.hass = SimpleNamespace(services=_services(served))
        if script is not None:
            entity._entry.options[CONF_REMOTE_COMMAND_SCRIPT] = script
        return entity, served

    async def test_a_configured_script_receives_the_exact_parameters(self, modules):
        entity, served = self._entity(modules, WATCH_SCRIPT)

        await entity.async_send_command(
            ["HOME", "KEYCODE_DPAD_CENTER"],
            num_repeats=3,
            delay_secs=0.25,
            hold_secs=1.5,
        )

        assert served == [
            (
                "script",
                "turn_on",
                {
                    "entity_id": WATCH_SCRIPT,
                    "variables": {
                        SCRIPT_VAR_COMMAND: ["HOME", "KEYCODE_DPAD_CENTER"],
                        SCRIPT_VAR_NUM_REPEATS: 3,
                        SCRIPT_VAR_DELAY_SECS: 0.25,
                        SCRIPT_VAR_HOLD_SECS: 1.5,
                    },
                },
            )
        ]
        assert entity._manager._protocol.key_calls == []  # key path untouched

    async def test_the_service_defaults_are_named_too(self, modules):
        """The variable set is fixed: a script can rely on all four being present."""
        entity, served = self._entity(modules, WATCH_SCRIPT)

        await entity.async_send_command(["home"])

        variables = served[0][2]["variables"]
        assert variables == {
            SCRIPT_VAR_COMMAND: ["home"],
            SCRIPT_VAR_NUM_REPEATS: 1,
            SCRIPT_VAR_DELAY_SECS: 0.0,
            SCRIPT_VAR_HOLD_SECS: 0.0,
        }

    async def test_unset_keeps_the_native_key_path(self, modules):
        entity, served = self._entity(modules)

        await entity.async_send_command(["HOME"], num_repeats=2, hold_secs=0.0)

        assert served == []
        assert entity._manager._protocol.key_calls == [
            ("KEYCODE_HOME", "SHORT"),
            ("KEYCODE_HOME", "SHORT"),
        ]

    @pytest.mark.parametrize("stored", ["", "   ", None, 7])
    async def test_a_blank_or_hand_edited_option_means_unset(self, modules, stored):
        entity, served = self._entity(modules, stored)

        await entity.async_send_command(["HOME"])

        assert served == []
        assert entity._manager._protocol.key_calls == [("KEYCODE_HOME", "SHORT")]

    async def test_a_raising_script_is_a_translated_error(self, modules):
        from homeassistant.exceptions import ServiceValidationError

        entity, _served = self._entity(modules, WATCH_SCRIPT)

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

        entity.hass.services = _FailingServices()

        with pytest.raises(ServiceValidationError) as raised:
            await entity.async_send_command(["HOME"])

        assert raised.value.translation_domain == "nvidia_shield"
        assert raised.value.translation_key == "command_script_failed"
        assert raised.value.translation_placeholders == {
            "action": "remote.send_command",
            "script": WATCH_SCRIPT,
            "error": "IR bridge is offline",
        }
        assert entity._manager._protocol.key_calls == []  # never a silent no-op


def test_both_new_exception_keys_have_copy_in_both_files():
    """A raised translation_key with no copy shows the user a bare key."""
    import json
    from pathlib import Path

    root = Path(__file__).resolve().parent.parent / "custom_components" / "nvidia_shield"
    for name in ("strings.json", "translations/en.json"):
        exceptions = json.loads((root / name).read_text(encoding="utf-8"))["exceptions"]
        for key in ("unknown_source", "command_script_failed"):
            assert exceptions[key]["title"], (name, key)
            assert exceptions[key]["message"], (name, key)
    assert "{source}" in json.loads(
        (root / "strings.json").read_text(encoding="utf-8")
    )["exceptions"]["unknown_source"]["message"]
    assert "remote_command_script" in json.loads(
        (root / "translations/en.json").read_text(encoding="utf-8")
    )["options"]["step"]["init"]["data"]