966 lines
34 KiB
Python
966 lines
34 KiB
Python
"""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
|