James Mackay
a1ce28996e
The user reads attributes as YAML — "These could do with being separate attributes rather than an object in one?" — so the nested `media_dag_now_playing` object is gone and every field it carried is its own top-level attribute: now_playing_via, now_playing_state, now_playing_title, now_playing_artist, now_playing_album, now_playing_media_type, now_playing_series, now_playing_season, now_playing_episode, now_playing_duration, now_playing_position, now_playing_position_updated_at, now_playing_thumbnail, now_playing_media_info Rules the surface now holds to: - an unknown field is OMITTED, never published as None (`title: null` inside the object was exactly the noise this removes); 0 and False are real values - a device reading `off` publishes no now-playing row at all — a Jellyfin session that has gone stale must not put a title beside a device that is off — and `on` (nothing reported) publishes none either, because there is no reading - `now_playing_state` IS the entity's own state, so the ladder stays the single answer even while the foreground tier's last-known bridge serves a title - the tier indicator is `now_playing_via` (`jellyfin` / `media_session` / `app`), NOT `source`: on a media_player `source`/`source_list` mean the selected app input (`Netflix`, `TiviMate`, …) and they still do — a tier named `source` is what made the reading look like media metadata from somewhere else. The internal reading key is renamed `via` as well, and the Jellyfin tier now names itself too - every published name is declared unrecorded (G12): these change every poll COMPATIBILITY: `media_dag_now_playing` is removed, not deprecated, and no alias is kept — a nested copy would double every poll's payload. README documents the exact names, these rules, the sensor's own flat set and a one-line template migration. sensor.py: the Jellyfin now-playing sensor follows the same no-None rule (`title` and `media_type` used to be published as null there). Tests: 17 new cases pin the flattening, the tier rename, the five states, the off/on gate, the playing->idle transition, the last-known bridge, privacy mode and agreement with the Current App sensor; the tier-dict tests follow `via`.
2460 lines
93 KiB
Python
2460 lines
93 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
|
|
# t65: the Jellyfin tier reads this; ``__init__`` sets it, the harness has to
|
|
# mirror it or a Jellyfin-shaped player raises on the flattened surface
|
|
player._media_info = 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["via"] == "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",
|
|
"via": "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", "via": "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", "via": "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", "via": "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", "via": "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", "via": "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
|
|
# t65: the nested ``media_dag_now_playing`` key is GONE; every flattened
|
|
# name it was replaced by has to be declared, or the recorder stores a new
|
|
# row per poll for each of them (G12)
|
|
from nsh.media_player import now_playing_attribute_names
|
|
|
|
assert now_playing_attribute_names() <= declared
|
|
assert "now_playing_via" in declared
|
|
assert "now_playing_title" in declared
|
|
assert "media_dag_now_playing" not in declared # the nested object is gone
|
|
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"
|
|
|
|
|
|
# ──────────────────────── t65: the flattened now-playing surface ─────────────
|
|
#
|
|
# The user reads attributes as YAML: "These could do with being separate attributes
|
|
# rather than an object in one?" The nested ``media_dag_now_playing`` object is
|
|
# gone; every field it carried is a top-level ``now_playing_*`` attribute, the tier
|
|
# indicator is ``now_playing_via`` (never ``source``, which on a media_player means
|
|
# the selected app input), and no now-playing attribute is ever published as None.
|
|
|
|
|
|
def _now_playing_attrs(player) -> dict:
|
|
"""Just the now-playing surface of an entity payload (t65).
|
|
|
|
The nested pre-t65 key is included on purpose, so a regression that quietly
|
|
brings the object back shows up in every state assertion here, not only in
|
|
the one test that names it.
|
|
"""
|
|
return {
|
|
key: value
|
|
for key, value in player.extra_state_attributes.items()
|
|
if key.startswith("now_playing_") or key == "media_dag_now_playing"
|
|
}
|
|
|
|
|
|
def make_now_playing_sensor(modules, jellyfin_state):
|
|
"""The Jellyfin now-playing sensor over one canned Jellyfin entity state."""
|
|
entry = SimpleNamespace(
|
|
entry_id="e1",
|
|
title="Shield",
|
|
options={"jellyfin_entity": "media_player.jellyfin"},
|
|
)
|
|
hass = SimpleNamespace(
|
|
states=SimpleNamespace(get=lambda entity_id: jellyfin_state)
|
|
)
|
|
return modules["sensor"].NvidiaShieldNowPlayingSensor(hass, entry)
|
|
|
|
|
|
class TestNowPlayingFlattenedT65:
|
|
"""t65: one reading -> separate top-level attributes, and never a None row."""
|
|
|
|
def test_a_jellyfin_reading_lands_in_its_own_attributes(self, modules):
|
|
player, _m = _state_player(modules, jellyfin_state="playing")
|
|
|
|
attrs = player.extra_state_attributes
|
|
|
|
assert "media_dag_now_playing" not in attrs # the object is gone
|
|
assert attrs["now_playing_via"] == "jellyfin"
|
|
assert attrs["now_playing_state"] == "playing"
|
|
assert attrs["now_playing_title"] == "Movie"
|
|
# the DAG role is a different fact and keeps its own meaning
|
|
assert attrs["media_dag_role"] == "source"
|
|
|
|
def test_the_tier_is_now_playing_via_and_source_stays_the_app_input(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, _m = _state_player(
|
|
modules,
|
|
sessions=[
|
|
MediaSessionInfo(
|
|
package="com.netflix.ninja",
|
|
title="Sky News",
|
|
playing=True,
|
|
artist="Anchor",
|
|
)
|
|
],
|
|
foreground="com.netflix.ninja",
|
|
)
|
|
player._entry.options["app_name_map"] = {"com.netflix.ninja": "Netflix"}
|
|
|
|
attrs = player.extra_state_attributes
|
|
|
|
assert attrs["now_playing_via"] == "media_session"
|
|
assert attrs["now_playing_title"] == "Sky News"
|
|
assert attrs["now_playing_artist"] == "Anchor" # its own attribute now
|
|
# `source`/`source_list` still mean the SELECTED APP INPUT
|
|
assert player.source_list == ["Netflix"]
|
|
assert attrs["media_dag_role"] == "source"
|
|
|
|
def test_the_flattener_drops_unknowns_and_keeps_real_zeros(self):
|
|
from nsh.media_player import flatten_now_playing
|
|
|
|
assert flatten_now_playing(None) == {}
|
|
assert flatten_now_playing({}) == {}
|
|
assert flatten_now_playing(
|
|
{
|
|
"via": "app",
|
|
"state": "idle",
|
|
"title": None,
|
|
"artist": "",
|
|
"thumbnail": None,
|
|
"position": 0,
|
|
"season": 0,
|
|
}
|
|
) == {
|
|
"now_playing_via": "app",
|
|
"now_playing_state": "idle",
|
|
"now_playing_position": 0,
|
|
"now_playing_season": 0,
|
|
}
|
|
|
|
def test_an_empty_reading_publishes_no_rows_at_all(self, modules):
|
|
player, _m = _state_player(modules, jellyfin_state="playing")
|
|
player._read = lambda entity_id: (
|
|
_MpState("playing") if entity_id == "media_player.jellyfin" else None
|
|
)
|
|
|
|
attrs = _now_playing_attrs(player)
|
|
|
|
# Jellyfin answered "playing" but has no metadata at all: the reading's
|
|
# own facts survive and the unknowns are OMITTED, never nulled
|
|
assert attrs == {
|
|
"now_playing_via": "jellyfin",
|
|
"now_playing_state": "playing",
|
|
}
|
|
|
|
def test_no_now_playing_attribute_is_ever_null(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
players = [
|
|
_state_player(modules, jellyfin_state="playing")[0],
|
|
_state_player(
|
|
modules,
|
|
sessions=[MediaSessionInfo(package="com.netflix.ninja", playing=True)],
|
|
foreground="com.netflix.ninja",
|
|
)[0],
|
|
_ladder_player(modules, foreground=TIVIMATE_PACKAGE)[0],
|
|
]
|
|
|
|
for player in players:
|
|
assert [k for k, v in _now_playing_attrs(player).items() if v is None] == []
|
|
|
|
def test_every_state_keeps_the_reading_and_the_state_in_step(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
cases = [
|
|
({"jellyfin_state": "playing"}, "playing"),
|
|
({"jellyfin_state": "paused"}, "paused"),
|
|
(
|
|
{
|
|
"sessions": [
|
|
MediaSessionInfo(
|
|
package="com.netflix.ninja", buffering=True, title="Sky News"
|
|
)
|
|
],
|
|
"foreground": "com.netflix.ninja",
|
|
},
|
|
"buffering",
|
|
),
|
|
({"power_entity_state": "off"}, "off"),
|
|
]
|
|
|
|
for kwargs, expected in cases:
|
|
player, _m = _state_player(modules, **kwargs)
|
|
attrs = _now_playing_attrs(player)
|
|
|
|
assert player.state.value == expected, kwargs
|
|
if expected == "off":
|
|
# nothing is playing on a device that is off — and the power
|
|
# entity's verdict outranks a stale session either way
|
|
assert attrs == {}, kwargs
|
|
continue
|
|
assert attrs["now_playing_state"] == expected, kwargs
|
|
|
|
def test_off_publishes_no_rows_even_with_a_stale_jellyfin_session(self, modules):
|
|
"""The gate that keeps a stale title off a device that is off."""
|
|
player, _m = _state_player(
|
|
modules, power_entity_state="off", jellyfin_state="playing"
|
|
)
|
|
|
|
assert player._get_now_playing() is not None # the stale reading exists...
|
|
assert player.state.value == "off" # ...and the device is off, so it is dropped
|
|
assert _now_playing_attrs(player) == {}
|
|
|
|
def test_on_publishes_no_rows_because_nothing_reported(self, modules):
|
|
player, _m = _state_player(modules)
|
|
|
|
assert player.state.value == "on"
|
|
assert _now_playing_attrs(player) == {}
|
|
|
|
def test_an_idle_app_publishes_the_idle_row(self, modules):
|
|
player, _m = _ladder_player(modules, foreground=TIVIMATE_PACKAGE)
|
|
player._last_channel = "101"
|
|
|
|
assert player.state.value == "idle"
|
|
assert _now_playing_attrs(player) == {
|
|
"now_playing_via": "app",
|
|
"now_playing_state": "idle",
|
|
"now_playing_title": "101",
|
|
}
|
|
|
|
def test_playing_to_idle_does_not_leave_the_title_behind(self, modules):
|
|
"""The transition the user feels: the old title must not linger."""
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, manager = _state_player(
|
|
modules,
|
|
sessions=[
|
|
MediaSessionInfo(
|
|
package="com.netflix.ninja", playing=True, title="Sky News"
|
|
)
|
|
],
|
|
foreground="com.netflix.ninja",
|
|
)
|
|
assert _now_playing_attrs(player)["now_playing_title"] == "Sky News"
|
|
|
|
# the session stops (the app is still open) and nothing is remembered
|
|
manager.store.set("sessions", [])
|
|
manager._last_known.clear()
|
|
|
|
attrs = _now_playing_attrs(player)
|
|
|
|
assert player.state.value == "idle"
|
|
assert "Sky News" not in attrs.values()
|
|
assert attrs == {}
|
|
|
|
def test_a_remembered_title_never_contradicts_the_entity_state(self, modules):
|
|
"""t55's TTL bridge serves a last-known PLAYBACK state in the tier.
|
|
|
|
The published row must still agree with the entity: the ladder is the
|
|
authority, so a dashboard reading `now_playing_state` never sees
|
|
"playing" beside a state of idle.
|
|
"""
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, manager = _ladder_player(modules, foreground="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)
|
|
|
|
attrs = _now_playing_attrs(player)
|
|
|
|
assert player.state.value == "idle" # nothing LIVE is playing
|
|
assert attrs["now_playing_state"] == "idle" # the row agrees with it
|
|
assert attrs["now_playing_title"] == "Sky News" # the remembered identity
|
|
assert attrs["now_playing_via"] == "app"
|
|
|
|
def test_privacy_mode_publishes_the_app_identity_not_the_media(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, _m = _state_player(
|
|
modules,
|
|
sessions=[
|
|
MediaSessionInfo(
|
|
package="com.netflix.ninja", playing=True, title="Sky News"
|
|
)
|
|
],
|
|
foreground="com.netflix.ninja",
|
|
)
|
|
player._entry.options["app_name_map"] = {"com.netflix.ninja": "Netflix"}
|
|
player._entry.options["adb_store_app_info"] = False # privacy mode
|
|
|
|
attrs = _now_playing_attrs(player)
|
|
|
|
assert attrs["now_playing_title"] == "Netflix" # the app, never the media
|
|
assert attrs["now_playing_via"] == "media_session"
|
|
assert "Sky News" not in attrs.values()
|
|
assert player.state.value == "playing" # privacy does not hide the state
|
|
|
|
def test_the_identity_agrees_with_the_current_app_sensor(self, modules):
|
|
player, manager = _state_player(modules, awake=True)
|
|
player._entry.options["app_name_map"] = {"ar.tvplayer.tv": "TiviMate"}
|
|
player._entry.runtime_data = manager
|
|
manager.store.set("current_app", "ar.tvplayer.tv")
|
|
sensor = make_app_sensor(modules, manager, player._entry)
|
|
|
|
attrs = player.extra_state_attributes
|
|
|
|
assert attrs["app_id"] == sensor.extra_state_attributes["app_id"]
|
|
assert attrs["app_name"] == sensor.native_value == "TiviMate"
|
|
|
|
|
|
class TestNowPlayingSensorRowsT65:
|
|
"""t65: the Jellyfin now-playing sensor publishes no None rows either."""
|
|
|
|
def test_a_missing_media_type_is_omitted_not_nulled(self, modules):
|
|
sensor = make_now_playing_sensor(
|
|
modules, _MpState("playing", {"media_title": "Movie"})
|
|
)
|
|
|
|
attrs = sensor.extra_state_attributes
|
|
|
|
assert attrs == {"playback_state": "playing", "title": "Movie"}
|
|
assert "media_type" not in attrs
|
|
|
|
def test_a_zero_position_and_duration_are_real_readings(self, modules):
|
|
sensor = make_now_playing_sensor(
|
|
modules,
|
|
_MpState(
|
|
"paused",
|
|
{
|
|
"media_title": "Movie",
|
|
"media_position": 0,
|
|
"media_duration": 0,
|
|
"media_season": 0,
|
|
"media_episode": 0,
|
|
},
|
|
),
|
|
)
|
|
|
|
attrs = sensor.extra_state_attributes
|
|
|
|
assert attrs["position"] == 0
|
|
assert attrs["duration"] == 0
|
|
assert attrs["season"] == 0
|
|
assert attrs["episode"] == 0
|
|
|
|
def test_an_idle_entity_publishes_only_the_idle_row(self, modules):
|
|
sensor = make_now_playing_sensor(modules, _MpState("idle"))
|
|
|
|
assert sensor.extra_state_attributes == {"playback_state": "idle"}
|
|
|
|
def test_a_missing_entity_is_still_idle(self, modules):
|
|
sensor = make_now_playing_sensor(modules, None)
|
|
|
|
assert sensor.extra_state_attributes == {"playback_state": "idle"}
|
|
|