engineer
eb62011ded
The live report: "Browse media seems to spin and never finish loading". The cause is in what async_browse_media HANDS BACK, not in the method — it returns with no awaits, so nothing inside it can hang. The root node's media_content_id was None, and HA core never serves a node without a string id: across the 43 BrowseMedia constructions in the installed core media_player platforms, zero pass None (the roots use a literal/module constant — arcam_fmj's "root", "apps", BROWSE_ROOT, TUNER_PRESETS_ROOT — or "" for a pure container). The media-browser frontend keys the node it is loading on that id, so a null id left the dialog re-requesting the same node for ever. BROWSE_ROOT_ID = "root" is now what the entity serves for EVERY requested id — the root id, None, or anything a stale frontend asks for — so a repeat request is idempotent rather than a loop. The type was reassessed as well: MediaType.APPS (t61's choice) is what core pairs with children=[...] — a folder OF apps — so on a zero-child node it invited the frontend to treat it as an unloaded collection. No MediaType member means "a directory with nothing in it", so the node carries the class it already declares (MediaClass.DIRECTORY -> "directory"), which is exactly what can_expand=False and children=[] say. Verified over the real path the frontend uses — websocket_browse_media -> player.async_browse_media -> payload.as_dict() — by driving HA's own decorated handler against a real entity (tests/test_ha_api_drift.py, which the CI drift job runs with Home Assistant installed): the root is a genuine BrowseMedia, its as_dict() survives json.dumps, the id is the string "root", and browse(root_id) == browse(None) in shape. Red proof on the pre-fix source: the new real-HA test fails at `assert isinstance(payload["media_content_id"], str)` -> `isinstance(None, str)`. The frontend bundle is not installed here (no hass_frontend) and there is no browser to drive, so the rendered dialog is NOT exercised: this pins the backend half of that path, end to end through the handler's decorators.
2131 lines
81 KiB
Python
2131 lines
81 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
|
|
from types import SimpleNamespace
|
|
|
|
import entity_stublib
|
|
import pytest
|
|
from conftest import _SHIELD_DIR
|
|
from entity_stublib import install, load_entity_modules
|
|
from nsh.connection import ShieldConnectionManager
|
|
from nsh.const import APP_LABEL_EXIT_MARKER, TIVIMATE_PACKAGE
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def scoped_entity_stubs():
|
|
"""Container for the HA stubs + loaded entity modules (t7 discipline)."""
|
|
snapshot = entity_stublib.snapshot_modules()
|
|
install()
|
|
yield load_entity_modules()
|
|
entity_stublib.restore_modules(snapshot)
|
|
|
|
|
|
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._intent_timer_unsub = None # mirrors __init__ (intent self-heal)
|
|
player._last_channel = None
|
|
player.hass = entity_stublib.FakeHass()
|
|
player._entity_tasks = set()
|
|
player._channel_task = None
|
|
player.async_write_ha_state = lambda: None
|
|
# t61: FakeHass carries the real `states` shape, so the entity's own `_read`
|
|
# is the one under test — an entity no test set still reads as None.
|
|
# t61: `__init__` gives the entity an (empty) power-sensor map; the harness
|
|
# mirrors it so the attribute payload can be read here too.
|
|
player._power_sensors = {}
|
|
|
|
# t41: nothing may leave this integration, so an HA service call is now the
|
|
# failure signal rather than a supported tier. `served` stays empty unless a
|
|
# regression reintroduces the external-entity delegation.
|
|
served = []
|
|
|
|
class _RecordingServices:
|
|
async def async_call(self, domain, service, data=None, **kwargs):
|
|
# t44: the payload matters now (command scripts pass HA script
|
|
# variables), so record it too — every existing assertion is `== []`.
|
|
served.append((domain, service, data))
|
|
|
|
player.hass.services = _RecordingServices()
|
|
|
|
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"])
|
|
|
|
# t41: both transports down is a no-op — no other integration is called
|
|
assert served == []
|
|
|
|
|
|
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 TestIntentRetryWithoutATransport:
|
|
"""t41: an intent no transport served surrenders instead of delegating."""
|
|
|
|
def _intent(self, **overrides):
|
|
from types import SimpleNamespace
|
|
|
|
intent = SimpleNamespace(
|
|
target_state="playing",
|
|
command="media_play",
|
|
created_at=0.0,
|
|
prior_state=None,
|
|
retry_count=0,
|
|
via_adb=False,
|
|
via_protocol=False,
|
|
adb_command=None,
|
|
protocol_key=None,
|
|
)
|
|
for key, value in overrides.items():
|
|
setattr(intent, key, value)
|
|
return intent
|
|
|
|
async def test_unserved_intent_surrenders_without_a_service_call(self, modules):
|
|
player, served, _intents = make_player(modules)
|
|
player.async_write_ha_state = lambda: None
|
|
player._intent = self._intent()
|
|
|
|
player._on_intent_timeout(0.0) # async_call_later passes the timestamp
|
|
|
|
assert player._intent is None # surrendered to the truth
|
|
assert served == [] # and nothing left this integration
|
|
|
|
async def test_protocol_served_intent_still_retries_over_the_protocol(self, modules):
|
|
player, served, _intents = make_player(modules)
|
|
player.async_write_ha_state = lambda: None
|
|
protocol = player._manager._protocol
|
|
player._intent = self._intent(
|
|
via_protocol=True, protocol_key="KEYCODE_MEDIA_PLAY"
|
|
)
|
|
|
|
player._on_intent_timeout(0.0) # async_call_later passes the timestamp
|
|
await asyncio.sleep(0.05)
|
|
|
|
assert protocol.key_calls == [("KEYCODE_MEDIA_PLAY", "SHORT")]
|
|
assert served == []
|
|
|
|
|
|
class TestAppFallback:
|
|
"""t41: the app identity comes from OUR transports' store only."""
|
|
|
|
async def test_app_id_comes_from_the_manager_store(self, modules):
|
|
player, _s, _i = make_player(modules)
|
|
player._manager._on_current_app("com.plex.hdx")
|
|
await asyncio.sleep(0.05) # drain the call_soon_threadsafe push
|
|
|
|
assert player.app_id == "com.plex.hdx"
|
|
assert player.app_name == "com.plex.hdx" # package is truth; map later
|
|
|
|
async def test_app_follows_the_latest_protocol_push(self, modules):
|
|
player, _s, _i = make_player(modules)
|
|
player._manager._on_current_app("com.netflix.ninja")
|
|
await asyncio.sleep(0.05)
|
|
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._read = lambda eid: None
|
|
return player, manager
|
|
|
|
|
|
class TestForegroundLadderTier:
|
|
"""P3-3a/t55: the ladder branch between the media-session tier and IDLE."""
|
|
|
|
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._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) — an open app, nothing playing:
|
|
# the t55 ladder reports IDLE, never ON ("On" is reserved for no data)
|
|
assert player.state.value == "idle"
|
|
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 == "idle" # launcher open, nothing playing
|
|
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 _MpState:
|
|
"""A HA State-shaped stand-in for the power entity / Jellyfin entity."""
|
|
|
|
def __init__(self, state, attributes=None):
|
|
self.state = state
|
|
self.attributes = attributes or {}
|
|
|
|
|
|
def _state_player(
|
|
modules,
|
|
*,
|
|
awake=None,
|
|
sessions=None,
|
|
foreground=None,
|
|
protocol_on=None,
|
|
power_entity_state=None,
|
|
jellyfin_state=None,
|
|
jellyfin_configured=False,
|
|
):
|
|
"""A player whose power/media sources are exactly what the test sets.
|
|
|
|
``None`` means "this source has not reported" — the distinction the t55
|
|
ladder turns on, so every test spells out which sources spoke.
|
|
``jellyfin_configured`` covers the entity being configured while ``_read``
|
|
still answers None (the entity is gone).
|
|
"""
|
|
player, _served, _intents = make_player(modules)
|
|
manager = player._manager
|
|
if awake is not None:
|
|
manager.store.set("awake", awake)
|
|
if sessions is not None:
|
|
manager.store.set("sessions", sessions)
|
|
if foreground is not None:
|
|
manager.store.set("foreground", foreground)
|
|
if protocol_on is not None:
|
|
manager.store.set("is_on", protocol_on)
|
|
|
|
entry = player._entry
|
|
entry.options = dict(entry.options or {})
|
|
if power_entity_state is not None:
|
|
entry.options["power_entity"] = "switch.shield_plug"
|
|
if jellyfin_state is not None or jellyfin_configured:
|
|
entry.options["jellyfin_entity"] = "media_player.jellyfin"
|
|
|
|
def _read(entity_id):
|
|
if entity_id == "switch.shield_plug" and power_entity_state is not None:
|
|
return _MpState(power_entity_state)
|
|
if entity_id == "media_player.jellyfin" and jellyfin_state is not None:
|
|
return _MpState(jellyfin_state, {"media_title": "Movie"})
|
|
return None
|
|
|
|
player._read = _read
|
|
return player, manager
|
|
|
|
|
|
class TestStateModelT55:
|
|
"""t55: exactly {playing, paused, buffering, idle, off} + ON for no data.
|
|
|
|
The live defect: ``state`` returned ON for "powered on with an open app but
|
|
nothing playing", which users read as "On overrides my playing states" and
|
|
which sat ABOVE the media tiers in one branch. Every source that reports
|
|
playing/paused/buffering must win, and ON must only describe the case where
|
|
no source reported anything at all.
|
|
"""
|
|
|
|
# ── the five states ──
|
|
|
|
async def test_no_data_at_all_is_the_only_on(self, modules):
|
|
player, _m = _state_player(modules)
|
|
|
|
assert player.state.value == "on"
|
|
|
|
async def test_protocol_off_is_off(self, modules):
|
|
player, _m = _state_player(modules, protocol_on=False)
|
|
|
|
assert player.state.value == "off"
|
|
|
|
async def test_power_entity_off_is_off_even_with_a_session(self, modules):
|
|
"""Explicit OFF outranks a stale media source."""
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, _m = _state_player(
|
|
modules,
|
|
awake=True,
|
|
power_entity_state="off",
|
|
sessions=[MediaSessionInfo(package="com.netflix.ninja", playing=True)],
|
|
)
|
|
|
|
assert player.state.value == "off"
|
|
|
|
async def test_open_app_with_nothing_playing_is_idle(self, modules):
|
|
"""The headline fix: ON no longer masks "on but nothing playing"."""
|
|
player, _m = _state_player(
|
|
modules, awake=True, foreground="ar.tvplayer.tv", protocol_on=True
|
|
)
|
|
|
|
assert player.state.value == "idle"
|
|
|
|
async def test_powered_on_with_no_app_is_idle(self, modules):
|
|
player, _m = _state_player(modules, protocol_on=True)
|
|
|
|
assert player.state.value == "idle"
|
|
|
|
async def test_power_entity_on_is_idle_not_on(self, modules):
|
|
"""A plug reporting 'on' IS a source — ON is reserved for no data."""
|
|
player, _m = _state_player(modules, power_entity_state="on")
|
|
|
|
assert player.state.value == "idle"
|
|
|
|
async def test_adb_standby_is_idle(self, modules):
|
|
player, _m = _state_player(modules, awake=False)
|
|
|
|
assert player.state.value == "idle"
|
|
|
|
# ── each source outranks the idle/on fallbacks ──
|
|
|
|
async def test_jellyfin_playing_outranks_the_idle_fallback(self, modules):
|
|
player, _m = _state_player(
|
|
modules, awake=True, foreground="ar.tvplayer.tv", jellyfin_state="playing"
|
|
)
|
|
|
|
assert player.state.value == "playing"
|
|
|
|
async def test_jellyfin_playing_outranks_no_power_data(self, modules):
|
|
"""No transport, no power entity — Jellyfin still knows better than ON."""
|
|
player, _m = _state_player(modules, jellyfin_state="playing")
|
|
|
|
assert player.state.value == "playing"
|
|
|
|
async def test_media_session_buffering_outranks_the_idle_fallback(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, _m = _state_player(
|
|
modules,
|
|
awake=True,
|
|
foreground="com.netflix.ninja",
|
|
sessions=[
|
|
MediaSessionInfo(package="com.netflix.ninja", buffering=True)
|
|
],
|
|
)
|
|
|
|
assert player.state.value == "buffering"
|
|
|
|
async def test_media_session_playing_outranks_the_idle_fallback(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, _m = _state_player(
|
|
modules,
|
|
awake=True,
|
|
foreground="com.netflix.ninja",
|
|
sessions=[MediaSessionInfo(package="com.netflix.ninja", playing=True)],
|
|
)
|
|
|
|
assert player.state.value == "playing"
|
|
|
|
async def test_media_session_paused_outranks_the_idle_fallback(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, _m = _state_player(
|
|
modules,
|
|
awake=True,
|
|
foreground="com.netflix.ninja",
|
|
sessions=[MediaSessionInfo(package="com.netflix.ninja", paused=True)],
|
|
)
|
|
|
|
assert player.state.value == "paused"
|
|
|
|
async def test_intent_playing_outranks_the_idle_fallback(self, modules):
|
|
player, _m = _state_player(modules, awake=True, foreground="ar.tvplayer.tv")
|
|
player._intent = SimpleNamespace(target_state="playing")
|
|
|
|
assert player.state.value == "playing"
|
|
|
|
async def test_live_session_beats_a_stale_cached_one(self, modules):
|
|
"""Session state 3/4 keep behaving as before: live truth, then cache."""
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, manager = _state_player(
|
|
modules,
|
|
awake=True,
|
|
foreground="com.netflix.ninja",
|
|
sessions=[MediaSessionInfo(package="com.netflix.ninja", paused=True)],
|
|
)
|
|
info = MediaSessionInfo(package="com.netflix.ninja", playing=True, title="Live")
|
|
manager._last_known["com.netflix.ninja"] = (info, manager._clock())
|
|
|
|
assert player.state.value == "paused" # the live session, not the cache
|
|
|
|
# ── vocabulary guard: no other value can be produced ──
|
|
|
|
def test_the_vocabulary_is_exactly_the_five_states_plus_on(self, modules):
|
|
from pathlib import Path
|
|
|
|
from nsh import media_player as mp
|
|
|
|
# Checked on the AST (the t37 idiom), not on prose: every state the
|
|
# module can return is in the documented vocabulary, and the
|
|
# media-tier values really are all used.
|
|
tree = ast.parse(Path(mp.__file__).read_text(encoding="utf-8"))
|
|
used = {
|
|
node.attr
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, ast.Attribute)
|
|
and isinstance(node.value, ast.Name)
|
|
and node.value.id == "MediaPlayerState"
|
|
}
|
|
assert used <= {"PLAYING", "PAUSED", "BUFFERING", "IDLE", "OFF", "ON"}
|
|
assert {"PLAYING", "PAUSED", "BUFFERING", "IDLE", "OFF"} <= used
|
|
# The documented vocabulary itself, as the user reads it on the card.
|
|
assert {s.lower() for s in used} == {
|
|
"playing",
|
|
"paused",
|
|
"buffering",
|
|
"idle",
|
|
"off",
|
|
"on",
|
|
}
|
|
|
|
async def test_on_never_masks_a_reported_power_state(self, modules):
|
|
"""Every power-reporting shape must be off or idle — never on."""
|
|
shapes = [
|
|
{"protocol_on": True},
|
|
{"protocol_on": False},
|
|
{"awake": True},
|
|
{"awake": False},
|
|
{"power_entity_state": "on"},
|
|
{"power_entity_state": "off"},
|
|
]
|
|
for kwargs in shapes:
|
|
player, _m = _state_player(modules, **kwargs)
|
|
assert player.state.value != "on", kwargs
|
|
|
|
# ── BUFFERING: state, controls, now-playing ──
|
|
|
|
async def test_buffering_advertises_the_playback_controls(self, modules):
|
|
from entity_stublib import MediaPlayerEntityFeature
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, _m = _state_player(
|
|
modules,
|
|
awake=True,
|
|
foreground="com.netflix.ninja",
|
|
sessions=[MediaSessionInfo(package="com.netflix.ninja", buffering=True)],
|
|
)
|
|
|
|
features = player.supported_features
|
|
assert features & MediaPlayerEntityFeature.PLAY
|
|
assert features & MediaPlayerEntityFeature.PAUSE
|
|
assert features & MediaPlayerEntityFeature.STOP
|
|
|
|
async def test_no_session_still_means_base_features_only(self, modules):
|
|
from entity_stublib import MediaPlayerEntityFeature
|
|
|
|
player, _m = _state_player(modules, awake=True, foreground="ar.tvplayer.tv")
|
|
|
|
assert not (player.supported_features & MediaPlayerEntityFeature.PLAY)
|
|
|
|
async def test_buffering_now_playing_is_named_buffering(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, _m = _state_player(
|
|
modules,
|
|
awake=True,
|
|
foreground="com.netflix.ninja",
|
|
sessions=[
|
|
MediaSessionInfo(
|
|
package="com.netflix.ninja", buffering=True, title="Movie"
|
|
)
|
|
],
|
|
)
|
|
|
|
np = player._get_adb_now_playing()
|
|
assert np["state"] == "buffering"
|
|
assert np["title"] == "Movie"
|
|
|
|
async def test_buffering_survives_privacy_mode(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, _m = _state_player(
|
|
modules,
|
|
awake=True,
|
|
foreground="com.netflix.ninja",
|
|
sessions=[
|
|
MediaSessionInfo(
|
|
package="com.netflix.ninja", buffering=True, title="Secret"
|
|
)
|
|
],
|
|
)
|
|
player._entry.options["adb_store_app_info"] = False
|
|
|
|
np = player._get_adb_now_playing()
|
|
assert np["state"] == "buffering"
|
|
assert np["title"] != "Secret"
|
|
|
|
async def test_last_known_buffering_reaches_the_foreground_tier(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, manager = _state_player(modules, awake=True, foreground="ar.tvplayer.tv")
|
|
manager._clock = lambda: 0.0
|
|
info = MediaSessionInfo(
|
|
package="ar.tvplayer.tv", buffering=True, title="Sky News"
|
|
)
|
|
manager._last_known["ar.tvplayer.tv"] = (info, 0.0)
|
|
|
|
np = player._get_app_now_playing()
|
|
assert np == {"state": "buffering", "title": "Sky News", "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",
|
|
"host": "192.168.1.50", # t41: the one address field
|
|
"adb_host": "192.168.1.50", # legacy spellings stay redacted too
|
|
"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"]["host"] == "**REDACTED**"
|
|
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:
|
|
"""t61/t63: the HA media browser publishes NO media — and says so unambiguously."""
|
|
|
|
async def test_the_browser_is_a_childless_root_with_a_real_id(self, modules):
|
|
"""t63: the null id is what made the live browse dialog spin for ever."""
|
|
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.children == [] # not one app row
|
|
assert root.can_play is False
|
|
assert root.can_expand is False # there is nothing to open
|
|
assert root.title == "mdarcy" # the device, not a second app list
|
|
# t63: the id has to be a real STRING. HA core never serves a node without
|
|
# one, and the frontend keys the node it is loading on that id — a None id
|
|
# left the dialog re-requesting the same node for ever (the live report).
|
|
assert isinstance(root.media_content_id, str) and root.media_content_id
|
|
assert root.media_content_id == "root"
|
|
# ...and the type must not claim a collection this node does not have:
|
|
# core pairs MediaType.APPS with children=[...] — a folder OF apps.
|
|
assert root.media_content_type == "directory"
|
|
# t37 D1: still a standard media_player.BrowseMedia root — explicit
|
|
# class/type, NO media_source `domain` (that parameter belongs to
|
|
# media_source.models.BrowseMediaSource, which also builds the
|
|
# media_source:// ids and is not exportable from media_player at all).
|
|
assert root.media_class == "directory"
|
|
assert root.children_media_class is None
|
|
assert not hasattr(root, "domain")
|
|
# ...and it serialises into the payload the frontend consumes
|
|
payload = root.as_dict()
|
|
assert payload["children"] == []
|
|
assert payload["can_expand"] is False
|
|
assert payload["media_content_id"] == "root"
|
|
|
|
async def test_apps_stay_reachable_through_the_sources(self, modules):
|
|
"""What the browser no longer offers, source_list/select_source still do."""
|
|
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.children == []
|
|
assert player.source_list == ["TiviMate", "Netflix"]
|
|
|
|
protocol = player._manager._protocol
|
|
await player.async_select_source("Netflix")
|
|
|
|
assert protocol.launch_calls == ["com.netflix.ninja"] # still launchable
|
|
async def test_every_requested_id_answers_the_same_leaf(self, modules):
|
|
"""A repeat request must be idempotent — not a loop the frontend can ride.
|
|
|
|
The handler ignores both arguments, so the root id, ``None`` and a stale
|
|
child id all return the same node: the frontend cannot bounce between
|
|
"loading" and "loaded" for one id.
|
|
"""
|
|
player, _s, _i = make_player(modules)
|
|
|
|
first = (await player.async_browse_media()).as_dict()
|
|
|
|
for content_type, content_id in (
|
|
(None, None),
|
|
("apps", "root"),
|
|
("apps", "apps"),
|
|
("app", "ar.tvplayer.tv"),
|
|
("channel", "12"),
|
|
):
|
|
repeat = await player.async_browse_media(content_type, content_id)
|
|
|
|
assert repeat.as_dict() == first
|
|
|
|
|
|
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):
|
|
"""A Current App sensor over the given manager/entry (t40).
|
|
|
|
t41: the sensor reads the manager's store and nothing else, so the harness
|
|
hands it a hass whose state machine is empty — a value here can only come
|
|
from our own transports.
|
|
"""
|
|
from types import SimpleNamespace
|
|
|
|
hass = SimpleNamespace(states=SimpleNamespace(get=lambda entity_id: None))
|
|
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.cache_app_label("com.netflix.ninja", "Netflix")
|
|
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_recent_apps_fallback_resolves_through_the_label_tier(self, modules):
|
|
player, _served, _intents = make_player(modules)
|
|
manager = player._manager
|
|
manager.cache_app_label("ar.tvplayer.tv", "TiviMate")
|
|
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"}
|
|
sensor = self._sensor(modules, player)
|
|
|
|
player._manager._on_current_app("ar.tvplayer.tv")
|
|
|
|
assert sensor.native_value == player.app_name == "TiviMate"
|
|
|
|
|
|
class _LabelPoller(FakeAdbPoller):
|
|
"""FakeAdbPoller + the surface a real watchdog cycle needs.
|
|
|
|
``is_awake``/``poll_media_sessions``/``all_sessions``/``poll_running_apps``
|
|
make ``_refresh`` a harmless cycle that still reaches the
|
|
``_drain_app_label_requests`` call, so a test drives exactly the
|
|
one-fetch-per-cycle behaviour the label tier depends on.
|
|
"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.pm_dumps: dict[str, str] = {}
|
|
self.dumps: list[str] = []
|
|
#: t57: True = the transport never delivered output (a FAILED fetch)
|
|
self.shell_failure = False
|
|
|
|
@property
|
|
def connected(self):
|
|
# FakeAdbPoller's property is always True; restated so this class reads
|
|
# on its own and the watchdog cycle spends no retry on connect.
|
|
return True
|
|
|
|
async def is_awake(self) -> bool:
|
|
return True
|
|
|
|
async def poll_media_sessions(self, foreground_package=None, head_lines=200):
|
|
return None
|
|
|
|
@property
|
|
def all_sessions(self) -> list:
|
|
return []
|
|
|
|
@property
|
|
def last_dump(self) -> str:
|
|
return ""
|
|
|
|
async def poll_running_apps(self) -> list[dict]:
|
|
return []
|
|
|
|
async def shell(self, cmd: str) -> str | None:
|
|
"""Model BOTH real outcomes of a label dump (t57).
|
|
|
|
A package in ``pm_dumps`` answers with its dump. A package that is NOT
|
|
there answers with the exit marker and nothing else — the shape a real
|
|
transport produces for "the command ran and this package has no
|
|
label". ``shell_failure`` turns every dump into the OTHER real shape:
|
|
the transport never delivered output at all (``adb_client.shell``
|
|
returns None when it is not connected or the command raised).
|
|
"""
|
|
self.dumps.append(cmd)
|
|
import shlex
|
|
|
|
if self.shell_failure:
|
|
return None
|
|
for package, dump in self.pm_dumps.items():
|
|
if shlex.quote(package) in cmd or package in cmd:
|
|
return dump
|
|
if APP_LABEL_EXIT_MARKER in cmd:
|
|
return f"{APP_LABEL_EXIT_MARKER}0\n"
|
|
return None
|
|
|
|
|
|
class TestCurrentAppNameEndToEndT55:
|
|
"""t55: the label fetch is driven off the CURRENT app, and a miss retries.
|
|
|
|
The live symptom was the Current App sensor (and app_name) showing a
|
|
package id. These drive the real chain: sensor render → app_label cache
|
|
miss → the fetch the manager enqueues → the next watchdog cycle drains it
|
|
for the package that is actually on screen, NOT some earlier app.
|
|
"""
|
|
|
|
def _wire(self, modules, current_app, dumps):
|
|
player, _served, _intents = make_player(modules)
|
|
manager = player._manager
|
|
poller = _LabelPoller()
|
|
poller.pm_dumps.update(dumps)
|
|
manager._poller = poller
|
|
player._entry.runtime_data = manager
|
|
sensor = make_app_sensor(modules, manager, player._entry)
|
|
|
|
manager._on_current_app(current_app)
|
|
return player, manager, sensor, poller
|
|
|
|
async def test_known_app_resolves_by_fetching_its_label(self, modules):
|
|
player, manager, sensor, poller = self._wire(
|
|
modules, "com.netflix.ninja", {"com.netflix.ninja": "application-label: 'Netflix'"}
|
|
)
|
|
|
|
# first render: cache miss → the sensor falls through to the package
|
|
assert sensor.native_value == "com.netflix.ninja"
|
|
assert manager.app_label_requests == {"com.netflix.ninja"}
|
|
|
|
# the watchdog drains the fetch for the CURRENT package
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
assert poller.dumps and "com.netflix.ninja" in poller.dumps[0]
|
|
|
|
# next render: the resolved device name, in the sensor AND the player
|
|
assert sensor.native_value == "Netflix"
|
|
assert sensor.extra_state_attributes["app_name"] == "Netflix"
|
|
assert player.app_name == "Netflix"
|
|
|
|
async def test_mapped_app_never_needs_the_device_label(self, modules):
|
|
player, manager, sensor, poller = self._wire(
|
|
modules, "ar.tvplayer.tv", {"ar.tvplayer.tv": "application-label: 'TiviMate'"}
|
|
)
|
|
player._entry.options["app_name_map"] = {"ar.tvplayer.tv": "TiviMate TV"}
|
|
|
|
assert sensor.native_value == "TiviMate TV"
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
|
|
# the explicit map wins and no pm dump was issued for it
|
|
assert sensor.native_value == "TiviMate TV"
|
|
assert not [d for d in poller.dumps if "pm dump" in d]
|
|
|
|
async def test_unknown_app_falls_through_and_the_next_poll_retries(self, modules):
|
|
"""The requirement, end to end: no permanent miss for an unresolvable app.
|
|
|
|
A failed fetch must not be cached as an answer: the next poll tries
|
|
again. Only the fetch that RAN may stamp the negative.
|
|
"""
|
|
player, manager, sensor, poller = self._wire(
|
|
modules, "com.example.unknown", {}
|
|
)
|
|
real_shell = poller.shell
|
|
failures = {"n": 0}
|
|
|
|
async def _flaky(cmd):
|
|
if "pm dump" in cmd:
|
|
failures["n"] += 1
|
|
raise RuntimeError("adb shell blew up")
|
|
return await real_shell(cmd)
|
|
|
|
poller.shell = _flaky
|
|
|
|
# poll 1: fetch fails → still the package id, nothing cached
|
|
assert sensor.native_value == "com.example.unknown"
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
assert sensor.native_value == "com.example.unknown"
|
|
assert failures["n"] == 1
|
|
assert "com.example.unknown" not in manager._app_labels
|
|
|
|
# poll 2: the render after the failure RE-ENQUEUES it — the retry is
|
|
# driven by the next render, exactly like the live poll cycle does
|
|
assert manager.app_label("com.example.unknown") is None
|
|
assert manager.app_label_requests == {"com.example.unknown"}
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
assert failures["n"] == 2
|
|
assert "com.example.unknown" not in manager._app_labels
|
|
|
|
# poll 3: the shell answers, the dump carries no label → a definitive
|
|
# negative is cached and the sensor keeps falling through to the id
|
|
poller.shell = real_shell
|
|
assert manager.app_label("com.example.unknown") is None
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
assert sensor.native_value == "com.example.unknown"
|
|
assert "com.example.unknown" in manager._app_labels
|
|
|
|
@pytest.mark.parametrize("blank", [None, ""], ids=["none", "empty"])
|
|
async def test_a_dump_with_no_output_is_a_failure_not_an_answer(
|
|
self, modules, blank
|
|
):
|
|
"""t59 F1 (HIGH): the live symptom, driven through the SENSOR.
|
|
|
|
``adb_client.shell`` swallows its error and returns None, and a real
|
|
dump can also come back as an empty string. Neither is the device
|
|
saying "this package has no label": reading either as an answer stamped
|
|
a 24 h definitive negative after ONE transient failure, so the sensor
|
|
stayed on the package id, exactly one dump was ever attempted, and a
|
|
later GOOD dump could not recover the label. Both shapes are driven
|
|
here — the pre-fix tree fails this at the second cycle.
|
|
"""
|
|
player, manager, sensor, poller = self._wire(
|
|
modules, "ar.tvplayer.tv", {"ar.tvplayer.tv": "application-label: 'TiviMate'"}
|
|
)
|
|
|
|
async def _blank(_cmd):
|
|
poller.dumps.append(_cmd)
|
|
return blank
|
|
|
|
# cycle 1: the transport delivered nothing → a FAILED fetch
|
|
poller.shell = _blank
|
|
assert sensor.native_value == "ar.tvplayer.tv"
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
|
|
assert "ar.tvplayer.tv" not in manager._app_labels # nothing stamped
|
|
assert sensor.native_value == "ar.tvplayer.tv"
|
|
|
|
# the next render re-enqueues, and the retry is what recovers the label
|
|
assert manager.app_label("ar.tvplayer.tv") is None
|
|
assert manager.app_label_requests == {"ar.tvplayer.tv"}
|
|
dump_count = len([d for d in poller.dumps if "pm dump" in d])
|
|
|
|
real_shell = _LabelPoller.shell.__get__(poller, _LabelPoller)
|
|
poller.shell = real_shell
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
|
|
# a later GOOD dump recovers it — the pre-fix cache made this impossible
|
|
assert len([d for d in poller.dumps if "pm dump" in d]) == dump_count + 1
|
|
assert sensor.native_value == "TiviMate"
|
|
assert sensor.extra_state_attributes["app_name"] == "TiviMate"
|
|
assert player.app_name == "TiviMate"
|
|
|
|
@pytest.mark.parametrize("blank", [None, ""], ids=["none", "empty"])
|
|
async def test_the_player_recovers_the_label_after_a_failed_fetch(
|
|
self, modules, blank
|
|
):
|
|
"""Same defect, driven through ``media_player.app_name`` (t59 F1).
|
|
|
|
The sensor and the player read the same chain, but they are two
|
|
different surfaces and the reviewer's reproduction observed BOTH
|
|
stuck on the package id — so both are pinned.
|
|
"""
|
|
player, manager, sensor, poller = self._wire(
|
|
modules, "com.netflix.ninja", {"com.netflix.ninja": "application-label: 'Netflix'"}
|
|
)
|
|
|
|
async def _blank(_cmd):
|
|
poller.dumps.append(_cmd)
|
|
return blank
|
|
|
|
poller.shell = _blank
|
|
assert player.app_name == "com.netflix.ninja"
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
assert "com.netflix.ninja" not in manager._app_labels
|
|
assert player.app_name == "com.netflix.ninja"
|
|
|
|
assert manager.app_label("com.netflix.ninja") is None # re-enqueued
|
|
poller.shell = _LabelPoller.shell.__get__(poller, _LabelPoller)
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
|
|
assert player.app_name == "Netflix"
|
|
assert sensor.native_value == "Netflix"
|
|
|
|
async def test_a_genuinely_label_less_package_is_still_negatively_cached(
|
|
self, modules
|
|
):
|
|
"""The other half: a dump that RAN and found no label stays definitive.
|
|
|
|
This is what keeps the t24 F1 rule closed — a label-less package costs
|
|
zero further round trips. The marker in the output is what separates
|
|
this from the failed fetch above.
|
|
"""
|
|
player, manager, sensor, poller = self._wire(modules, "com.oem.blank", {})
|
|
|
|
assert sensor.native_value == "com.oem.blank"
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
|
|
assert "com.oem.blank" in manager._app_labels # a definitive negative
|
|
assert manager.app_label("com.oem.blank") is None
|
|
assert manager.app_label_requests == set() # no re-enqueue, ever
|
|
|
|
dumps = [d for d in poller.dumps if "pm dump" in d]
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
assert [d for d in poller.dumps if "pm dump" in d] == dumps # zero more
|
|
|
|
async def test_the_fetch_targets_the_app_on_screen_not_the_previous_one(self, modules):
|
|
player, manager, sensor, poller = self._wire(
|
|
modules,
|
|
"ar.tvplayer.tv",
|
|
{
|
|
"ar.tvplayer.tv": "application-label: 'TiviMate'",
|
|
"com.google.android.youtube.tv": "application-label: 'YouTube'",
|
|
},
|
|
)
|
|
|
|
assert sensor.native_value == "ar.tvplayer.tv" # miss for the 1st app
|
|
manager._on_current_app("com.google.android.youtube.tv")
|
|
assert sensor.native_value == "com.google.android.youtube.tv"
|
|
|
|
# both packages are queued; the watchdog drains ONE per cycle
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
await manager._watchdog_cycle(5.0)
|
|
await asyncio.sleep(0)
|
|
|
|
# every drained fetch names a package, and the app that is CURRENT is
|
|
# the one the sensor now shows the label of
|
|
assert "ar.tvplayer.tv" in poller.dumps[0]
|
|
assert "com.google.android.youtube.tv" in poller.dumps[1]
|
|
assert sensor.native_value == "YouTube"
|
|
|
|
|
|
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
|
|
|
|
|
|
class TestReportingFactsT58:
|
|
"""t58 UNIT 0 — the t59 F2 ruling: ON = "no power or playback fact reported".
|
|
|
|
The live report was "On overrides my playing states"; t55 fixed the loud half
|
|
(a powered-on Shield with an app open). This is the quiet half: a source that
|
|
ANSWERS "not playing" was still falling through to ON. Every case below is a
|
|
source that answered, so the ladder must stop at IDLE.
|
|
"""
|
|
|
|
# ── Jellyfin, in every state it can report ──
|
|
|
|
async def test_jellyfin_idle_is_idle_not_on(self, modules):
|
|
player, _m = _state_player(modules, jellyfin_state="idle")
|
|
|
|
assert player.state.value == "idle"
|
|
|
|
async def test_jellyfin_off_is_idle_not_on(self, modules):
|
|
"""`off` is a REPORT about the session, not the Shield's own power:
|
|
only the configured power entity and the protocol push can say OFF."""
|
|
player, _m = _state_player(modules, jellyfin_state="off")
|
|
|
|
assert player.state.value == "idle"
|
|
|
|
async def test_jellyfin_standby_is_idle_not_on(self, modules):
|
|
player, _m = _state_player(modules, jellyfin_state="standby")
|
|
|
|
assert player.state.value == "idle"
|
|
|
|
async def test_jellyfin_unavailable_is_idle_not_on(self, modules):
|
|
"""The one the ruling names explicitly: unavailable still answered."""
|
|
player, _m = _state_player(modules, jellyfin_state="unavailable")
|
|
|
|
assert player.state.value == "idle"
|
|
|
|
async def test_a_configured_but_missing_jellyfin_entity_is_idle(self, modules):
|
|
"""Configured IS the report: we are watching a source that exists.
|
|
|
|
`_read` returns None here (the entity is not in the state machine), and
|
|
the old ladder fell straight through to ON — for an entry whose owner
|
|
had explicitly configured a media source.
|
|
"""
|
|
player, _m = _state_player(modules, jellyfin_configured=True)
|
|
|
|
assert player.state.value == "idle"
|
|
|
|
# ── a live session with no playback flag, and a bare foreground app ──
|
|
|
|
async def test_a_session_with_no_playback_flag_is_idle_not_on(self, modules):
|
|
"""The app publishes a session; that it is NOT playing is the report."""
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, _m = _state_player(
|
|
modules, sessions=[MediaSessionInfo(package="com.netflix.ninja")]
|
|
)
|
|
|
|
assert player.state.value == "idle"
|
|
|
|
async def test_a_bare_foreground_app_is_idle_not_on(self, modules):
|
|
"""The protocol's current-app push answered — nothing else did."""
|
|
player, _m = _state_player(modules, foreground="ar.tvplayer.tv")
|
|
|
|
assert player.state.value == "idle"
|
|
|
|
# ── the negative control: ON is still reachable, just narrowly ──
|
|
|
|
async def test_on_survives_when_truly_nothing_reported(self, modules):
|
|
player, _m = _state_player(modules)
|
|
|
|
assert player.state.value == "on"
|
|
|
|
async def test_on_is_unreachable_with_any_source_configured(self, modules):
|
|
"""The rule as a shape: each single source takes ON off the table."""
|
|
for kwargs in (
|
|
{"jellyfin_state": "idle"},
|
|
{"jellyfin_state": "off"},
|
|
{"jellyfin_state": "standby"},
|
|
{"jellyfin_state": "unavailable"},
|
|
{"jellyfin_configured": True},
|
|
{"foreground": "ar.tvplayer.tv"},
|
|
{"protocol_on": True},
|
|
{"awake": True},
|
|
{"power_entity_state": "on"},
|
|
):
|
|
player, _m = _state_player(modules, **kwargs)
|
|
assert player.state.value != "on", kwargs
|
|
|
|
# ── a silent source must not MASK a live one (the ladder keeps going) ──
|
|
|
|
async def test_idle_jellyfin_does_not_mask_a_playing_session(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, _m = _state_player(
|
|
modules,
|
|
jellyfin_state="idle",
|
|
sessions=[MediaSessionInfo(package="com.netflix.ninja", playing=True)],
|
|
foreground="com.netflix.ninja",
|
|
)
|
|
|
|
assert player.state.value == "playing"
|
|
|
|
async def test_unavailable_jellyfin_does_not_mask_a_paused_session(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, _m = _state_player(
|
|
modules,
|
|
jellyfin_state="unavailable",
|
|
sessions=[MediaSessionInfo(package="com.netflix.ninja", paused=True)],
|
|
foreground="com.netflix.ninja",
|
|
)
|
|
|
|
assert player.state.value == "paused"
|
|
|
|
async def test_an_off_power_entity_still_outranks_a_playing_session(self, modules):
|
|
"""The t55 rule is untouched by the F2 correction."""
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, _m = _state_player(
|
|
modules,
|
|
power_entity_state="off",
|
|
sessions=[MediaSessionInfo(package="com.netflix.ninja", playing=True)],
|
|
)
|
|
|
|
assert player.state.value == "off"
|
|
|