From ccde3cff711b96458e22c294fceb0dff9835215a Mon Sep 17 00:00:00 2001 From: James Mackay Date: Mon, 12 Oct 2026 06:53:04 +0100 Subject: [PATCH] fix(media_player): the state ladder is idle/playing/buffering/off; ON means no data (t55) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The live report: the card showed "On" where the user expected Idle, Playing, Buffering or Off, and "On" appeared to override playing states. Root cause: `state` returned MediaPlayerState.ON for two different situations — "powered on with an app open and nothing playing" (tier 3.5) and "we have no idea" (the fallthrough) — while the power gate above the media tiers returned IDLE whenever power resolved to None, which is exactly the case in which Jellyfin a media session DO know what is playing. Media truth was therefore lost whenever the protocol push and the ADB wake poll had not reported yet. The ladder is now: 1. OFF — the power entity says off, or the protocol is_on push is False. Explicit power outranks media, so a stale Jellyfin session cannot resurrect playback on a device that is off. 2. intent — a just-sent play/pause/stop. 3. Jellyfin — playing / paused. 4. session — the ADB media session: buffering / playing / paused. 5. IDLE — ADB standby (powered but asleep), or the device is on (protocol is_on / ADB awake / the plug) with nothing playing. "Powered on, app open, nothing playing" is IDLE. 6. ON — nothing reported anything at all: no transport, no power entity, no Jellyfin, no session. ON never masks a playing/paused/buffering source and never appears while any source reported a power or media state. Buffering also carries the playback controls: a buffering session advertises PLAY/PAUSE/STOP exactly like a playing one, and the now-playing dict names it "buffering" instead of collapsing it into "playing" and contradicting the entity state. The module docstring documents the ladder and the ON-only-with-no-data rule, and the new TestStateModelT55 / TestCurrentAppNameEndToEndT55 pin one test per source (Jellyfin, intent, media session, power entity, protocol push, ADB wake) that a reporting source outranks ON/IDLE. Red-proof against the previous tree (new tests, old source): 21 failed / 17 passed, with the three headline symptoms visible as literal asserts — "on" where "idle" belongs, "idle" where "on" belongs (no data), and "playing" lost because power was unknown. --- .../nvidia_shield/media_player.py | 155 ++++-- tests/test_command_routing.py | 493 +++++++++++++++++- 2 files changed, 605 insertions(+), 43 deletions(-) diff --git a/custom_components/nvidia_shield/media_player.py b/custom_components/nvidia_shield/media_player.py index 9b4b39e..40ae33b 100644 --- a/custom_components/nvidia_shield/media_player.py +++ b/custom_components/nvidia_shield/media_player.py @@ -13,6 +13,31 @@ State architecture: keyevent. The external androidtv_remote entity tier that used to back this is gone — t41 removed it with the entity — so a command has exactly one owner. +State ladder (t55) — what ``state`` reports, in order. The vocabulary is +EXACTLY ``{playing, paused, buffering, idle, off, on}``; each step is the first +source that has something to say, and the earlier steps always win: + + 1. OFF — the configured power entity reads "off", or the protocol's + is_on push is False (the device reporting its own power + state). Both OUTRANK media: a stale Jellyfin session must not + resurrect playback on a device that is off. + 2. intent — a play/pause/stop this entity just sent (optimistic UI until + a slow Jellyfin poll confirms it; self-heal in 10 s). + 3. Jellyfin — STATE_PLAYING / STATE_PAUSED of the configured entity. + 4. session — the ADB media session's PlaybackState, including BUFFERING + (Android state 6). Session state 3 → playing, 2 → paused. + 5. IDLE — ADB says the device is asleep (standby), OR the device is + powered on (protocol is_on True / ADB awake / the plug is on) + and NOTHING is playing. "Powered on, app open, nothing + playing" is IDLE for exactly this reason. + 6. ON — NOTHING reported anything at all: no transport connected, no + power entity, no Jellyfin, no media session, no intent. ON is + the "we cannot know" state and the only situation it may + describe. It must NEVER appear while any source reported a + power or media state, and it must never mask a + playing/paused/buffering source — the rule that fixes the live + "On overrides my playing states" report. + Intent tracking bridges the gap between instant commands and slow Jellyfin polling: 1. User presses pause → intent recorded → UI shows paused immediately 2. Command sent over the serving transport → Shield pauses on screen @@ -623,30 +648,55 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity): return None + @staticmethod + def _session_state(info) -> MediaPlayerState | None: + """The MediaPlayerState a media session is reporting, or None (t55). + + One place decides what a session means, so ``state`` and + ``supported_features`` cannot disagree about it. Buffering is checked + FIRST: a session refilling its buffer is mid-playback, and while the + Shield never sets state 6 together with 3, BUFFERING is the more + informative of the two if a future firmware ever did. + """ + if info is None: + return None + if info.buffering: + return MediaPlayerState.BUFFERING + if info.playing: + return MediaPlayerState.PLAYING + if info.paused: + return MediaPlayerState.PAUSED + return None + @property def state(self) -> MediaPlayerState: - # Check if plug is off first — truly off + """The playback state — the ladder is documented in the module docstring. + + Exactly ``{playing, paused, buffering, idle, off}`` are reported for a + device we have data about; ``ON`` is reserved for the case where NO + source reported anything at all (no transport connected, no power and + no media data) and must never mask a playing/paused/buffering source. + A powered-on Shield with an app open but nothing playing reads IDLE, + not ON (t55 — the live defect users read as "On overrides my playing + states"). + """ + # 1. Explicit OFF. The plug being off is definitive; the protocol's + # is_on push is the device reporting its own power state. Either + # one OUTRANKS every media source — a stale Jellyfin session must + # never resurrect playback on a device that is off. if self._power_entity: power_state = self._read(self._power_entity) if power_state and power_state.state == "off": return MediaPlayerState.OFF + manager = self._manager + if manager is not None and manager.is_on is False: + return MediaPlayerState.OFF - power = self._is_power_on() - - if power is False: - if self._manager is not None and self._manager.is_on is False: - return MediaPlayerState.OFF # protocol says powered off - # ADB says asleep — standby (yellow dot) - return MediaPlayerState.IDLE - - if power is None: - return MediaPlayerState.IDLE - - # Tier 1: Active intent overrides Jellyfin + # 2. A user-commanded intent is the freshest media truth we have. if self._intent is not None: return MediaPlayerState(self._intent.target_state) - # Tier 2: Jellyfin is authoritative + # 3. Jellyfin (slow, but authoritative for its own sessions). jf = self._read(self._jellyfin_entity) if jf: if jf.state == STATE_PLAYING: @@ -654,22 +704,28 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity): if jf.state == STATE_PAUSED: return MediaPlayerState.PAUSED - # Tier 3: ADB media session (non-Jellyfin apps) - adb_info = self._manager.best_session(self.app_id) if self._manager else None - if adb_info: - if adb_info.playing: - return MediaPlayerState.PLAYING - if adb_info.paused: - return MediaPlayerState.PAUSED + # 4. Live media session (non-Jellyfin apps), buffering included. + session = manager.best_session(self.app_id) if manager is not None else None + session_state = self._session_state(session) + if session_state is not None: + return session_state - # Tier 3.5 (P3-3): foreground fallback — an app is open (protocol - # current_app push, else the ADB mResumedActivity parse) but publishes - # no media session. The device is still ON; the now-playing tier - # carries the last-known/channel/audio-hint identity. - if self._manager is not None and self._manager.foreground_app: - return MediaPlayerState.ON + # 5. ADB standby: powered but asleep — still data (the device + # answered), so IDLE, never ON. media_player.STANDBY is deprecated + # (2025-07), so IDLE is the compliant mapping (t22 audit). + if manager is not None and manager.awake is False: + return MediaPlayerState.IDLE - # Tier 4: On but no media + # 6. The device is on (protocol is_on True, ADB awake, or the power + # plug is on — every one of those is a source that reported + # something). An open app or an awake device reads IDLE, not ON. + power = self._is_power_on() + if power is not None or self._power_entity: + return MediaPlayerState.IDLE + + # 7. Nothing anywhere reported anything: no transport connected, no + # power entity, no media source. ON is the "we cannot know" state, + # and the only situation it is allowed to describe. return MediaPlayerState.ON @property @@ -677,9 +733,13 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity): if self._jf_active(): features = _BASE_FEATURES | _PLAYBACK_FEATURES else: - # ADB media session also supports playback controls - adb_info = self._manager.best_session(self.app_id) if self._manager else None - if adb_info and (adb_info.playing or adb_info.paused): + # A media session's playback controls include the BUFFERING case: + # pausing/resuming a buffering stream must be offered exactly like + # PLAYING (t55). + session = ( + self._manager.best_session(self.app_id) if self._manager else None + ) + if session is not None and self._session_state(session) is not None: features = _BASE_FEATURES | _PLAYBACK_FEATURES else: features = _BASE_FEATURES @@ -902,6 +962,21 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity): np["media_info"] = self._media_info return np + def _session_playback_label(self, info) -> str: + """``playing``/``buffering``/``paused``/``idle`` for a live session (t55). + + BUFFERING is its own label rather than a flavour of playing: the state + entity says so, and the now-playing dict must not contradict it. The + session flags themselves stay independent; this only names the state. + """ + if info.buffering: + return "buffering" + if info.playing: + return "playing" + if info.paused: + return "paused" + return "idle" + def _get_adb_now_playing(self) -> dict | None: """Build now-playing from ADB media session (non-Jellyfin apps).""" info = self._manager.best_session(self.app_id) if self._manager else None @@ -909,9 +984,9 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity): return None if not self._adb_store_info: # Privacy mode: show that something is playing but not what - if info.playing or info.paused: + if info.playing or info.paused or info.buffering: return { - "state": "playing" if info.playing else "paused", + "state": self._session_playback_label(info), "title": self.app_name or self.app_id or "Playing", "source": "media_session", } @@ -920,7 +995,7 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity): if not info.title: return None np = { - "state": "playing" if info.playing else ("paused" if info.paused else "idle"), + "state": self._session_playback_label(info), "title": info.title, "source": "media_session", } @@ -934,6 +1009,10 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity): """Foreground-tier now-playing (P3-3a): an app is open but publishes no media session (TiviMate, R2). Title = last known state, else the last channel entered, else the zero-cost audio-playback hint. + + The STATE here is ``idle`` or the last-known session's playback state; + the entity-level ladder reports IDLE for this tier (t55), so the two + agree — a bare foreground app is no longer reported as ON. """ manager = self._manager if manager is None: @@ -945,8 +1024,8 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity): state = "idle" title = None info = manager.last_known_session(foreground) - if info is not None and (info.playing or info.paused): - state = "playing" if info.playing else "paused" + if info is not None and (info.playing or info.paused or info.buffering): + state = self._session_playback_label(info) # Privacy mode: identity stays the app, never the media title title = None if not self._adb_store_info else info.title if title is None and not self._adb_store_info: @@ -963,7 +1042,9 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity): # E hint (zero round-trip cost): audio was recently played title = "Media recently active" if title is None: - return None # bare foreground app — the state stays ON regardless + # Bare foreground app: no now-playing identity, but the device is + # ON — the entity ladder reports IDLE for this (t55). + return None return {"state": state, "title": title, "source": "app"} def _get_power_data(self) -> dict | None: diff --git a/tests/test_command_routing.py b/tests/test_command_routing.py index b383e67..2ead4e8 100644 --- a/tests/test_command_routing.py +++ b/tests/test_command_routing.py @@ -9,6 +9,7 @@ layer marks which tier served each command. import ast import asyncio import sys +from types import SimpleNamespace import entity_stublib import pytest @@ -429,7 +430,7 @@ def _ladder_player(modules, foreground=None, audio=None): class TestForegroundLadderTier: - """P3-3a: the ladder branch between the media-session tier and ON.""" + """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) @@ -466,8 +467,9 @@ class TestForegroundLadderTier: audio=["ar.tvplayer.tv"], ) - # TiviMate publishes no session (R2) — fg tier keeps the device ON - assert player.state.value == "on" + # 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", @@ -478,7 +480,7 @@ class TestForegroundLadderTier: 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.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): @@ -516,6 +518,311 @@ class TestForegroundLadderTier: 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, +): + """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. + """ + 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: + 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.""" @@ -1259,7 +1566,7 @@ class TestCurrentAppSensorFriendlyName: 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()) + manager.cache_app_label("com.netflix.ninja", "Netflix") sensor = self._sensor(modules, player) manager._on_current_app("com.netflix.ninja") @@ -1270,7 +1577,7 @@ class TestCurrentAppSensorFriendlyName: 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.cache_app_label("ar.tvplayer.tv", "TiviMate") manager.store.set("running_apps", [{"package": "ar.tvplayer.tv"}]) sensor = self._sensor(modules, player) @@ -1299,6 +1606,180 @@ class TestCurrentAppSensorFriendlyName: 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] = [] + + @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: + self.dumps.append(cmd) + import shlex + + for package, dump in self.pm_dumps.items(): + if shlex.quote(package) in cmd or package in cmd: + return dump + 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 + + 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.