test: command-routing and play_media(channel) orchestration — digit sequence with cancel guard, fallback ordering protocol→ADB→legacy, remote send_command holds/repeats, volume readout, is_on ladder

tests/test_command_routing.py

@@ -0,0 +1,387 @@
+"""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
+
+
+@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.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 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.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_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 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