diff --git a/custom_components/nvidia_shield/const.py b/custom_components/nvidia_shield/const.py index e2d270e..b9d1926 100644 --- a/custom_components/nvidia_shield/const.py +++ b/custom_components/nvidia_shield/const.py @@ -90,6 +90,7 @@ PROTOCOL_REMOTE_KEYS = { # §7 channel selection: digits only, ~0.1 s apart, no ENTER by default # (replicates HA core androidtv_remote's verified behavior). CHANNEL_DIGIT_DELAY = 0.1 +CONF_CHANNEL_DIGIT_DELAY = "channel_digit_delay" CONF_CHANNEL_SEND_ENTER = "channel_send_enter" # Verified on-device package names (docs/RESEARCH_SHIELD.md): TiviMate does diff --git a/custom_components/nvidia_shield/media_player.py b/custom_components/nvidia_shield/media_player.py index e02c7f6..2758770 100644 --- a/custom_components/nvidia_shield/media_player.py +++ b/custom_components/nvidia_shield/media_player.py @@ -44,6 +44,7 @@ from .const import ( CHANNEL_DIGIT_DELAY, CONF_ADB_STORE_APP_INFO, CONF_ANDROIDTV_ENTITY, + CONF_CHANNEL_DIGIT_DELAY, CONF_CHANNEL_SEND_ENTER, CONF_JELLYFIN_ENTITY, CONF_POWER_ENTITY, @@ -817,7 +818,7 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity): self._channel_task = None async def _channel_entry_sequence(self, digits: list[str]) -> None: - delay = self._entry.options.get("channel_digit_delay", CHANNEL_DIGIT_DELAY) + delay = self._entry.options.get(CONF_CHANNEL_DIGIT_DELAY, CHANNEL_DIGIT_DELAY) send_enter = self._entry.options.get(CONF_CHANNEL_SEND_ENTER, False) try: for digit in digits: diff --git a/tests/entity_stublib.py b/tests/entity_stublib.py new file mode 100644 index 0000000..8e873cf --- /dev/null +++ b/tests/entity_stublib.py @@ -0,0 +1,191 @@ +"""Home-Assistant object stubs sized to unit-test the entity modules (t10). + +Import-only stubs: MediaPlayerEntity/RemoteEntity bases, the feature/state +enum values the integration actually uses, and the event helpers. Tests load +their entity module through the synthetic ``nsh`` package AFTER calling +:func:`install` (and the per-file fixture pops everything at teardown). +""" + +import asyncio +import enum +import sys +import types + +ENTITIES_STUB_NAMES = ( + "voluptuous", + "homeassistant", + "homeassistant.config_entries", + "homeassistant.core", + "homeassistant.const", + "homeassistant.components", + "homeassistant.components.media_player", + "homeassistant.components.remote", + "homeassistant.components.button", + "homeassistant.helpers", + "homeassistant.helpers.entity_registry", + "homeassistant.helpers.selector", + "homeassistant.helpers.event", + "homeassistant.helpers.entity", + "homeassistant.helpers.entity_platform", + "nsh.helpers", + "nsh.power", + "nsh.jellyfin", + "nsh.media_player", + "nsh.remote", +) + + +class MediaPlayerState(enum.StrEnum): + OFF = "off" + ON = "on" + IDLE = "idle" + PLAYING = "playing" + PAUSED = "paused" + + +class MediaPlayerEntityFeature(enum.IntFlag): + PLAY = 1 + PAUSE = 2 + STOP = 4 + VOLUME_MUTE = 8 + VOLUME_STEP = 16 + VOLUME_SET = 32 + TURN_ON = 64 + TURN_OFF = 128 + PLAY_MEDIA = 256 + NEXT_TRACK = 512 + PREVIOUS_TRACK = 1024 + + +class MediaEntityManager: + """trivial stand-in for the HA platform internals we don't exercise.""" + + +class MediaPlayerEntity: + """Base entity: attributes only; state writing is a no-op.""" + + def async_write_ha_state(self) -> None: + return None + + +class RemoteEntity: + def async_write_ha_state(self) -> None: + return None + + +def install() -> None: + """Register the stub modules (idempotent).""" + if "homeassistant.components.media_player" in sys.modules: + return + + vol = types.ModuleType("voluptuous") + vol.Required = lambda key, default=None: key + vol.Optional = lambda key, default=None: key + vol.Schema = lambda schema: schema + sys.modules["voluptuous"] = vol + + ha = types.ModuleType("homeassistant") + sys.modules["homeassistant"] = ha + + config_entries = types.ModuleType("homeassistant.config_entries") + config_entries.ConfigEntry = type("ConfigEntry", (), {}) + config_entries.ConfigFlow = type("ConfigFlow", (), {}) + config_entries.OptionsFlow = type("OptionsFlow", (), {}) + sys.modules["homeassistant.config_entries"] = config_entries + + core = types.ModuleType("homeassistant.core") + core.HomeAssistant = object + core.callback = lambda fn: fn + core.CALLBACK_TYPE = object + core.Event = object + sys.modules["homeassistant.core"] = core + + ha_const = types.ModuleType("homeassistant.const") + ha_const.EVENT_HOMEASSISTANT_STOP = "homeassistant_stop" + ha_const.STATE_PLAYING = "playing" + ha_const.STATE_PAUSED = "paused" + ha_const.Platform = types.SimpleNamespace( + MEDIA_PLAYER="media_player", + SENSOR="sensor", + REMOTE="remote", + BUTTON="button", + ) + sys.modules["homeassistant.const"] = ha_const + + components = types.ModuleType("homeassistant.components") + components.__path__ = [] + sys.modules["homeassistant.components"] = components + + mp = types.ModuleType("homeassistant.components.media_player") + mp.MediaPlayerEntity = MediaPlayerEntity + mp.MediaPlayerDeviceClass = types.SimpleNamespace(TV="tv") + mp.MediaPlayerEntityFeature = MediaPlayerEntityFeature + mp.MediaPlayerState = MediaPlayerState + sys.modules["homeassistant.components.media_player"] = mp + + remote = types.ModuleType("homeassistant.components.remote") + remote.RemoteEntity = RemoteEntity + sys.modules["homeassistant.components.remote"] = remote + + button = types.ModuleType("homeassistant.components.button") + button.ButtonEntity = type("ButtonEntity", (), {}) + sys.modules["homeassistant.components.button"] = button + + helpers = types.ModuleType("homeassistant.helpers") + entity_registry = types.ModuleType("homeassistant.helpers.entity_registry") + + class _Registry: + entities: dict = {} + + def async_get(self, entity_id): + return None + + entity_registry.async_get = lambda hass: _Registry() + helpers.entity_registry = entity_registry + sys.modules["homeassistant.helpers.entity_registry"] = entity_registry + + event = types.ModuleType("homeassistant.helpers.event") + event.async_call_later = lambda hass, delay, callback_fn: (lambda: None) + event.async_track_state_change_event = lambda *args, **kwargs: (lambda: None) + helpers.event = event + helpers.selector = types.SimpleNamespace( + EntitySelector=lambda *a, **k: None, + EntitySelectorConfig=lambda *a, **k: "", + SelectSelector=lambda *a, **k: None, + SelectSelectorConfig=lambda **k: "", + SelectSelectorMode=types.SimpleNamespace(DROPDOWN="dropdown"), + ) + entity = types.ModuleType("homeassistant.helpers.entity") + entity.DeviceInfo = dict + helpers.entity = entity + entity_platform = types.ModuleType("homeassistant.helpers.entity_platform") + entity_platform.AddEntitiesCallback = object + helpers.entity_platform = entity_platform + sys.modules["homeassistant.helpers.event"] = event + sys.modules["homeassistant.helpers.entity"] = entity + sys.modules["homeassistant.helpers.entity_platform"] = entity_platform + sys.modules["homeassistant.helpers"] = helpers + + +def load_entity_modules() -> dict: + """Load the integration's HA-free-ish entity modules under nsh.""" + install() + from conftest import _SHIELD_DIR, _load_module + + for name in ("helpers", "power", "jellyfin", "media_player", "remote"): + if f"nsh.{name}" not in sys.modules: + _load_module(f"nsh.{name}", _SHIELD_DIR / f"{name}.py") + return { + name: sys.modules[f"nsh.{name}"] + for name in ("power", "jellyfin", "media_player", "remote") + } + + +class FakeHass: + """Sized to entity unit tests: background tasks + no-op call_later.""" + + def async_create_task(self, coro): + return asyncio.create_task(coro) + + def async_call_later(self, delay, callback_fn): + return lambda: None diff --git a/tests/test_command_routing.py b/tests/test_command_routing.py new file mode 100644 index 0000000..ba6dd96 --- /dev/null +++ b/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