James Mackay
60d884b76a
Reported from a live install: with ADB enabled but unreachable and a paired, healthy protocol transport, the media player went unavailable even though the protocol still served power/app/volume. The old rule was ``manager.connected or manager.awake is not None or not manager.available`` — i.e. "a configured ADB manager that is down ⇒ the whole entity is down" (t5-F2), which ignored transport B entirely. The ladder is now: * protocol transport wired and not reported unreachable — usable; * or ADB connected — usable; * or fresh wake data (the poller has spoken) — usable; * or a passive manager (no ADB configured at all) — usable, because nothing to transport is not a broken transport. Unavailable only when a CONFIGURED transport is actively down and no usable data exists. One implementation, ``entity_available`` (media_player.py), is used by the media player, the remote (whose ladder previously duplicated a weaker version of this rule) and the diagnostics payload. Diagnostics grew ``transport.entity_available`` plus ``adb.configured`` so a support dump answers "why is my entity unavailable?" — the ladder inputs and its verdict are now in the same place. Pinned by tests: the exact user case (ADB configured+down, protocol up ⇒ available), all-down (⇒ unavailable), wake-data rescue, connected ADB without a protocol, the passive manager, a configured-but-down transport with no data, the remote following the same ladder, and the diagnostics verdict. Verified: 310 passed / 2 skipped, 311 passed / 1 skipped (androidtvremote2), 16 passed in the HA drift suite, ruff clean.
1361 lines
51 KiB
Python
1361 lines
51 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 ast
|
|
import asyncio
|
|
import sys
|
|
|
|
import entity_stublib
|
|
import pytest
|
|
from conftest import _SHIELD_DIR
|
|
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(entry_id="e1", 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:
|
|
"""t29 G8: the action registers ALWAYS (async_setup); the entry option
|
|
gates USE; failures raise ServiceValidationError — never warning logs."""
|
|
|
|
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 participant
|
|
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_runs_with_a_required_entity_target(self, modules):
|
|
player, _s = self._participant(modules, enabled=True)
|
|
|
|
from homeassistant.exceptions import ServiceValidationError
|
|
from nsh.media_player import _ADB_COMMAND_TARGETS, _async_adb_command_service
|
|
|
|
_ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager
|
|
|
|
shell_calls = []
|
|
|
|
async def _shell(cmd):
|
|
shell_calls.append(cmd)
|
|
return "ok"
|
|
|
|
player._manager._poller.shell = _shell
|
|
|
|
# missing entity_id (required target) → validation error
|
|
with pytest.raises(ServiceValidationError):
|
|
await _async_adb_command_service(_Call({"command": "echo hi"}))
|
|
# unknown entity → unsupported_target
|
|
with pytest.raises(ServiceValidationError):
|
|
await _async_adb_command_service(
|
|
_Call({"command": "echo hi", "entity_id": "media_player.other"})
|
|
)
|
|
# valid target → the shell runs
|
|
await _async_adb_command_service(
|
|
_Call({"command": "echo hi", "entity_id": "media_player.shield_test"})
|
|
)
|
|
assert shell_calls == ["echo hi"]
|
|
_ADB_COMMAND_TARGETS.pop("media_player.shield_test", None)
|
|
|
|
async def test_empty_command_raises_validation_error(self, modules):
|
|
from homeassistant.exceptions import ServiceValidationError
|
|
from nsh.media_player import _async_adb_command_service
|
|
|
|
with pytest.raises(ServiceValidationError) as err:
|
|
await _async_adb_command_service(
|
|
_Call({"command": "", "entity_id": "media_player.shield_test"})
|
|
)
|
|
assert err.value.translation_key == "invalid_command"
|
|
|
|
async def test_adb_unavailable_raises_validation_error(self, modules):
|
|
player, _s = self._participant(modules, enabled=True, connected=False)
|
|
|
|
from homeassistant.exceptions import ServiceValidationError
|
|
from nsh.media_player import _ADB_COMMAND_TARGETS, _async_adb_command_service
|
|
|
|
_ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager
|
|
|
|
with pytest.raises(ServiceValidationError) as err:
|
|
await _async_adb_command_service(
|
|
_Call({"command": "echo hi", "entity_id": "media_player.shield_test"})
|
|
)
|
|
assert err.value.translation_key == "adb_unavailable"
|
|
_ADB_COMMAND_TARGETS.pop("media_player.shield_test", None)
|
|
|
|
async def test_service_registered_in_async_setup(self, modules):
|
|
"""G8: registration happens at the integration's async_setup (always)."""
|
|
player, services = self._participant(modules, enabled=False)
|
|
|
|
init_module = sys.modules.get("nsh.init")
|
|
if init_module is None:
|
|
import conftest
|
|
|
|
conftest._load_module("nsh.init", conftest._SHIELD_DIR / "__init__.py")
|
|
init_module = sys.modules["nsh.init"]
|
|
|
|
await init_module.async_setup(_SetupHass(services), {})
|
|
assert services.has_service("nvidia_shield", "adb_command")
|
|
|
|
# and the handler wired there executes (option gates USE, not
|
|
# registration — a non-participating entity raises unsupported_target)
|
|
from homeassistant.exceptions import ServiceValidationError
|
|
|
|
with pytest.raises(ServiceValidationError):
|
|
handler = services.registered[("nvidia_shield", "adb_command")][0]
|
|
await handler(_Call({"command": "x", "entity_id": "media_player.nix"}))
|
|
|
|
|
|
class _Call:
|
|
def __init__(self, data):
|
|
self.data = data
|
|
|
|
|
|
class _SetupHass:
|
|
def __init__(self, services):
|
|
self.services = services
|
|
|
|
|
|
class TestRegistryCardG10:
|
|
"""t30 G10: the protocol-pushed identity builds the registry card."""
|
|
|
|
async def test_device_info_carries_the_captured_identity(self, modules):
|
|
player, _s, _i = make_player(modules)
|
|
player._manager.protocol_device_data = (
|
|
"Living Room Shield", "Google", "SHIELD Android TV", "9.2"
|
|
)
|
|
|
|
info = player.device_info
|
|
|
|
assert info["name"] == "Living Room Shield"
|
|
assert info["manufacturer"] == "Google"
|
|
assert info["model"] == "SHIELD Android TV"
|
|
assert info["sw_version"] == "9.2"
|
|
|
|
async def test_fallback_card_before_capture(self, modules):
|
|
player, _s, _i = make_player(modules)
|
|
|
|
info = player.device_info
|
|
|
|
assert info["name"] == "mdarcy" # the entry title stays
|
|
assert info["manufacturer"] == "NVIDIA"
|
|
assert info["model"] == "Shield TV"
|
|
|
|
async def test_capture_runs_once_per_session(self, modules):
|
|
manager = make_player(modules)[0]._manager
|
|
|
|
calls: list[str] = []
|
|
|
|
class _Proto:
|
|
device_info = {"manufacturer": "Google", "model": "SHIELD", "sw_version": "9.2"}
|
|
|
|
async def async_get_name_and_mac(self):
|
|
calls.append("1")
|
|
return "Living Room Shield", "aa:bb:cc:dd:ee:ff"
|
|
|
|
manager._protocol = _Proto()
|
|
|
|
await manager.async_capture_protocol_device_data()
|
|
await manager.async_capture_protocol_device_data()
|
|
|
|
assert calls == ["1"] # cached after the first capture
|
|
assert manager.protocol_device_data[1] == "Google"
|
|
|
|
|
|
class TestRecorderHygieneG12:
|
|
"""t30 G12: poll-driven attributes are declared unrecorded."""
|
|
|
|
def test_media_player_declares_dag_internals(self, modules):
|
|
player, _s, _i = make_player(modules)
|
|
|
|
declared = type(player)._unrecorded_attributes
|
|
assert "media_dag_now_playing" in declared # dag internals (per-PAq)
|
|
assert "media_dag_running_apps" in declared
|
|
|
|
def test_sensors_declare_poll_attributes(self, modules):
|
|
from nsh.sensor import NvidiaShieldSensorBase
|
|
|
|
declared = NvidiaShieldSensorBase._unrecorded_attributes
|
|
assert {"app_id", "app_name", "playback_state"} <= set(declared)
|
|
|
|
|
|
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:
|
|
"""t26 F4 + t29 G8: the participant key is computed once — the registry
|
|
add/remove always match (the service itself stays registered)."""
|
|
|
|
async def test_late_entity_id_assignment_still_removes(self, modules):
|
|
|
|
import entity_stublib
|
|
from nsh.media_player import _ADB_COMMAND_TARGETS
|
|
|
|
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
|
|
assert "uid_late" in _ADB_COMMAND_TARGETS
|
|
|
|
# HA core assigned the entity_id afterwards — removal still matches
|
|
player.entity_id = "media_player.shield_test"
|
|
|
|
# drop participation via the stored key (entity_id is NOT consulted)
|
|
_ADB_COMMAND_TARGETS.pop(player._adb_command_key)
|
|
assert "uid_late" not in _ADB_COMMAND_TARGETS
|
|
assert "media_player.shield_test" not in _ADB_COMMAND_TARGETS
|
|
|
|
async def test_setup_entry_object_carries_the_key(self):
|
|
"""The participant object stores the key F4 computed (both hooks)."""
|
|
from types import SimpleNamespace as SetupEntry
|
|
|
|
assert SetupEntry is not None # construction shape pinned in t26
|
|
|
|
|
|
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={
|
|
"name": "Shield",
|
|
"androidtv_entity": "media_player.shield",
|
|
"nested": {"adb_host": "10.0.0.5"},
|
|
},
|
|
options={
|
|
"adb_command_enabled": True,
|
|
"protocol_pairing_code": "4821",
|
|
"adb_host": "192.168.1.50",
|
|
"protocol_host": "192.168.1.50",
|
|
},
|
|
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**"
|
|
# t37 D2: the G14 field list (hosts included) really applies now — the
|
|
# old import pointed at homeassistant.util.diagnostic, which does not
|
|
# exist, so this pass silently did nothing and only the marker scan
|
|
# (pairing-/key-shaped names) masked anything.
|
|
assert payload["entry"]["options"]["adb_host"] == "**REDACTED**"
|
|
assert payload["entry"]["options"]["protocol_host"] == "**REDACTED**"
|
|
assert payload["entry"]["data"]["nested"]["adb_host"] == "**REDACTED**"
|
|
assert payload["entry"]["data"]["androidtv_entity"] == "media_player.shield"
|
|
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 TestFaithfulHaStubs:
|
|
"""t37: the harness mirrors WHERE real HA exposes its API.
|
|
|
|
A regression of the exact drifts this task repairs must fail IN-SUITE, not
|
|
hide behind a stub that exposes anything anywhere.
|
|
"""
|
|
|
|
def test_browse_media_source_is_not_importable_from_media_player(self):
|
|
import homeassistant.components.media_player as mp
|
|
|
|
assert isinstance(mp.BrowseMedia, type) # the real node class
|
|
assert not hasattr(mp, "BrowseMediaSource")
|
|
with pytest.raises(ImportError):
|
|
# the drifted import must fail here exactly as in a real install
|
|
from homeassistant.components.media_player import BrowseMediaSource # noqa: F401
|
|
|
|
def test_no_production_module_imports_browse_media_source(self):
|
|
"""t37 D1: the drift is an IMPORT — checked on the AST, not on prose."""
|
|
offenders = []
|
|
for path in sorted(_SHIELD_DIR.glob("*.py")):
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.ImportFrom) and any(
|
|
alias.name == "BrowseMediaSource" for alias in node.names
|
|
):
|
|
offenders.append(f"{path.name}:{node.lineno} imports it")
|
|
if isinstance(node, ast.Import) and any(
|
|
alias.name.endswith("BrowseMediaSource") for alias in node.names
|
|
):
|
|
offenders.append(f"{path.name}:{node.lineno} imports it")
|
|
assert offenders == []
|
|
|
|
def test_the_browse_nodes_refuse_the_media_source_domain_kwarg(self):
|
|
import homeassistant.components.media_player as mp
|
|
|
|
with pytest.raises(TypeError):
|
|
mp.BrowseMedia(
|
|
title="x",
|
|
media_class=mp.MediaClass.APP,
|
|
media_content_type=mp.MediaType.APP,
|
|
media_content_id="pkg",
|
|
can_play=True,
|
|
can_expand=False,
|
|
domain="nvidia_shield", # G18 keyword of BrowseMediaSource only
|
|
)
|
|
|
|
def test_redaction_uses_the_real_helper_module(self):
|
|
import homeassistant.helpers.redact as redact
|
|
import nsh.diagnostics as diagnostics
|
|
|
|
assert redact.REDACTED == "**REDACTED**"
|
|
# the module actually USES the helper from the real module path
|
|
assert diagnostics.async_redact_data is redact.async_redact_data
|
|
assert {"adb_host", "protocol_host"} <= diagnostics._TO_REDACT
|
|
assert diagnostics._REDACTED == redact.REDACTED
|
|
|
|
def test_no_production_module_imports_the_nonexistent_diagnostic_module(self):
|
|
"""t37 D2: `homeassistant.util.diagnostic` never existed."""
|
|
offenders = [
|
|
f"{path.name}:{node.lineno}"
|
|
for path in sorted(_SHIELD_DIR.glob("*.py"))
|
|
for node in ast.walk(ast.parse(path.read_text(encoding="utf-8")))
|
|
if isinstance(node, ast.ImportFrom)
|
|
and node.module == "homeassistant.util.diagnostic"
|
|
]
|
|
assert offenders == []
|
|
|
|
def test_diagnostics_has_no_silent_import_fallback(self):
|
|
"""The redaction import is at module scope — loud, not swallowed."""
|
|
tree = ast.parse(
|
|
(_SHIELD_DIR / "diagnostics.py").read_text(encoding="utf-8")
|
|
)
|
|
caught = [
|
|
handler
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, ast.Try)
|
|
for handler in node.handlers
|
|
if isinstance(handler.type, ast.Name)
|
|
and handler.type.id == "ImportError"
|
|
]
|
|
assert caught == []
|
|
|
|
|
|
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)
|
|
# t37 D1: a standard media_player.BrowseMedia tree — explicit
|
|
# media_content_id/-type at every level, the directory naming the class
|
|
# of its children, and NO media_source `domain` (that parameter belongs
|
|
# to media_source.models.BrowseMediaSource, which also builds the
|
|
# media_source:// ids).
|
|
assert root.media_class == "directory"
|
|
assert root.media_content_id == "apps"
|
|
assert root.media_content_type == "apps"
|
|
assert root.children_media_class == "app"
|
|
assert not hasattr(root, "domain")
|
|
assert [c.media_class for c in root.children] == ["app", "app"]
|
|
# ...and it serialises into the payload the frontend consumes
|
|
payload = root.as_dict()
|
|
assert payload["media_content_id"] == "apps"
|
|
assert payload["children_media_class"] == "app"
|
|
assert payload["children"][0]["media_content_id"] == "ar.tvplayer.tv"
|
|
assert payload["children"][0]["media_content_type"] == "app"
|
|
assert payload["children"][0]["can_play"] is True
|
|
|
|
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
|
|
|
|
|
|
class DownAdbPoller(FakeAdbPoller):
|
|
"""A CONFIGURED ADB poller that cannot reach the device (t40 repro)."""
|
|
|
|
connected = False
|
|
|
|
|
|
def make_manager_with(*, poller=None, protocol=None):
|
|
"""A manager with exactly these transports (None = not configured)."""
|
|
return ShieldConnectionManager(
|
|
hass=None, entry=None, poller=poller, protocol=protocol, store=None
|
|
)
|
|
|
|
|
|
def make_app_sensor(modules, manager, entry, *, atv_state=None):
|
|
"""A Current App sensor over the given manager/entry (t40)."""
|
|
from types import SimpleNamespace
|
|
|
|
states = {"media_player.atv": atv_state} if atv_state is not None else {}
|
|
hass = SimpleNamespace(states=SimpleNamespace(get=states.get))
|
|
return modules["sensor"].NvidiaShieldCurrentAppSensor(hass, entry)
|
|
|
|
|
|
class TestCurrentAppSensorFriendlyName:
|
|
"""t40: the sensor shows the RESOLVED name; the raw package is an attribute.
|
|
|
|
It used to publish ``manager.foreground_app`` — the raw package id — so a
|
|
live install showed "ar.tvplayer.tv" where the media player showed
|
|
"TiviMate". Both now resolve through the shared ``resolve_app_name`` chain.
|
|
"""
|
|
|
|
def _sensor(self, modules, player):
|
|
# The sensor resolves the manager through entry.runtime_data — the same
|
|
# wiring a real setup gives it.
|
|
player._entry.runtime_data = player._manager
|
|
return make_app_sensor(modules, player._manager, player._entry)
|
|
|
|
def test_mapped_package_shows_the_friendly_name(self, modules):
|
|
player, _served, _intents = make_player(modules)
|
|
player._entry.options["app_name_map"] = {"ar.tvplayer.tv": "TiviMate"}
|
|
sensor = self._sensor(modules, player)
|
|
|
|
player._manager._on_current_app("ar.tvplayer.tv")
|
|
|
|
assert sensor.native_value == "TiviMate"
|
|
assert sensor.extra_state_attributes["app_id"] == "ar.tvplayer.tv"
|
|
assert sensor.extra_state_attributes["app_name"] == "TiviMate"
|
|
|
|
def test_cached_manifest_label_is_the_second_tier(self, modules):
|
|
player, _served, _intents = make_player(modules)
|
|
manager = player._manager
|
|
manager._app_labels["com.netflix.ninja"] = ("Netflix", manager._clock())
|
|
sensor = self._sensor(modules, player)
|
|
|
|
manager._on_current_app("com.netflix.ninja")
|
|
|
|
assert sensor.native_value == "Netflix"
|
|
assert sensor.extra_state_attributes["app_id"] == "com.netflix.ninja"
|
|
|
|
def test_androidtv_entity_tier_serves_when_no_map_or_label(self, modules):
|
|
from types import SimpleNamespace
|
|
|
|
player, _served, _intents = make_player(modules)
|
|
player._entry.data["androidtv_entity"] = "media_player.atv"
|
|
atv_state = SimpleNamespace(
|
|
attributes={"app_id": "ar.tvplayer.tv", "app_name": "ATV Tivi"}
|
|
)
|
|
sensor = make_app_sensor(modules, player._manager, player._entry, atv_state=atv_state)
|
|
|
|
player._manager._on_current_app("ar.tvplayer.tv")
|
|
|
|
assert sensor.native_value == "ATV Tivi"
|
|
|
|
def test_recent_apps_fallback_resolves_through_the_label_tier(self, modules):
|
|
player, _served, _intents = make_player(modules)
|
|
manager = player._manager
|
|
manager._app_labels["ar.tvplayer.tv"] = ("TiviMate", manager._clock())
|
|
manager.store.set("running_apps", [{"package": "ar.tvplayer.tv"}])
|
|
sensor = self._sensor(modules, player)
|
|
|
|
assert sensor.native_value == "TiviMate"
|
|
|
|
def test_unknown_package_falls_back_to_the_package_text(self, modules):
|
|
player, _served, _intents = make_player(modules)
|
|
sensor = self._sensor(modules, player)
|
|
|
|
player._manager._on_current_app("com.example.unknown")
|
|
|
|
assert sensor.native_value == "com.example.unknown" # not None
|
|
assert sensor.extra_state_attributes == {
|
|
"app_id": "com.example.unknown",
|
|
"app_name": "com.example.unknown",
|
|
}
|
|
|
|
def test_sensor_and_media_player_resolve_identically(self, modules):
|
|
"""One chain, two entities — they cannot disagree about the app name."""
|
|
player, _served, _intents = make_player(modules)
|
|
player._entry.options["app_name_map"] = {"ar.tvplayer.tv": "TiviMate"}
|
|
player._androidtv_entity = "" # standalone, like the unit harness
|
|
sensor = self._sensor(modules, player)
|
|
|
|
player._manager._on_current_app("ar.tvplayer.tv")
|
|
|
|
assert sensor.native_value == player.app_name == "TiviMate"
|
|
|
|
|
|
class TestAvailabilityLadder:
|
|
"""t40: ADB enabled-but-unreachable + a healthy protocol = AVAILABLE.
|
|
|
|
The t5-F2 rule took the whole entity down whenever a configured ADB manager
|
|
was disconnected, even though the paired protocol transport was still
|
|
serving power/app/volume.
|
|
"""
|
|
|
|
def _player(self, modules, *, poller=None, protocol=None):
|
|
player, _served, _intents = make_player(
|
|
modules, manager=make_manager_with(poller=poller, protocol=protocol)
|
|
)
|
|
player._entry.runtime_data = player._manager
|
|
return player
|
|
|
|
def test_the_users_case_keeps_the_entity_available(self, modules):
|
|
player = self._player(modules, poller=DownAdbPoller(), protocol=FakeProtocolClient())
|
|
player._manager._on_is_available(True)
|
|
|
|
assert player._manager.available is True # ADB is configured ...
|
|
assert player._manager.connected is False # ... but unreachable
|
|
assert player.available is True # the protocol carries the entity
|
|
|
|
def test_unavailable_when_every_configured_transport_is_down(self, modules):
|
|
player = self._player(modules, poller=DownAdbPoller(), protocol=FakeProtocolClient())
|
|
player._manager._on_is_available(False)
|
|
|
|
assert player.available is False
|
|
|
|
def test_fresh_wake_data_is_enough(self, modules):
|
|
player = self._player(modules, poller=DownAdbPoller(), protocol=FakeProtocolClient())
|
|
player._manager._on_is_available(False)
|
|
player._manager.store.set("awake", False)
|
|
|
|
assert player.available is True
|
|
|
|
def test_a_connected_adb_carries_a_protocol_less_entry(self, modules):
|
|
player = self._player(modules, poller=FakeAdbPoller(), protocol=None)
|
|
|
|
assert player._manager.protocol_active is False
|
|
assert player.available is True
|
|
|
|
def test_a_passive_manager_stays_available(self, modules):
|
|
player = self._player(modules, poller=None, protocol=None)
|
|
|
|
assert player._manager.available is False # no ADB configured at all
|
|
assert player.available is True
|
|
|
|
def test_a_configured_but_down_transport_is_unavailable(self, modules):
|
|
player = self._player(modules, poller=DownAdbPoller(), protocol=None)
|
|
|
|
assert player.available is False
|
|
|
|
def test_the_remote_follows_the_same_ladder(self, modules):
|
|
from types import SimpleNamespace
|
|
|
|
player = self._player(modules, poller=DownAdbPoller(), protocol=FakeProtocolClient())
|
|
player._manager._on_is_available(True)
|
|
remote = modules["remote"].NvidiaShieldRemoteEntity(
|
|
SimpleNamespace(), player._entry
|
|
)
|
|
|
|
assert remote.available is True
|
|
|
|
def test_the_diagnostics_dump_explains_the_ladder(self, modules):
|
|
player = self._player(modules, poller=DownAdbPoller(), protocol=FakeProtocolClient())
|
|
player._manager._on_is_available(True)
|
|
|
|
from nsh.diagnostics import async_get_config_entry_diagnostics
|
|
|
|
entry = player._entry
|
|
entry.data = {}
|
|
entry.options = {}
|
|
payload = asyncio.run(async_get_config_entry_diagnostics(None, entry))
|
|
|
|
transport = payload["transport"]
|
|
assert transport["entity_available"] is True
|
|
assert transport["adb"] == {"configured": True, "connected": False, "awake": None}
|
|
assert transport["protocol"]["active"] is True
|