Not-Expected-Yet/ha-nvidia-shield
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity
Files
14c168c25e6f103e482a7fe8cf38024cf20267e9
ha-nvidia-shield/tests/test_command_routing.py
T
James Mackay 14c168c25e feat(remote): the app/activity picker, and a protocol transport that really connects (t15) 
The user's remote displayed but offered no app picker, and the protocol
transport was never actually switched on. Both halves are fixed here.

REMOTE PARITY. The entity declares RemoteEntityFeature.ACTIVITY — exactly
what core's androidtv_remote declares — and publishes activity_list, which
is what makes the frontend render an activity picker at all. The list is
the app directory (the user's apps map, else the curated table as a floor)
in map order, then the apps ADB discovery found and something can name, in
the one name chain (user map > device label > curated > package id).
Nothing nameless is listed as a bare package id, and current_activity is
None until something is known: no activity is invented. remote.turn_on
honours activity= by waking the device when it reports itself off and then
launching through the existing launch path; a listed name resolves
case-insensitively, a package id or a detail link passes through, and
anything else raises unknown_activity naming the value instead of silently
doing nothing. An empty/absent activity keeps the previous behaviour
exactly, async_turn_off is untouched, and a configured command script still
owns the sequence.

WHY THE TRANSPORT WAS DEAD, MEASURED. androidtvremote2's keep_reconnecting
only starts _async_reconnect, whose whole body is
`while self._remote_message_protocol:` — false on the first check with
nothing connected, so the task returned immediately: no connect attempt, no
retry, and the auth callback was never reachable. HA core connects FIRST and
only then keeps it alive. Our _start_protocol called keep_reconnecting after
generating a certificate and never called async_connect anywhere, so every
send_key_command/send_launch_app_command raised ConnectionClosed internally,
the helpers swallowed it into `return False` plus a debug line, and no push
(is_available/is_on/current_app/volume_info) ever arrived.

The fix is a background connect loop: async_connect under a bounded timeout,
network failures retried with the ADB watchdog's jitter and backoff,
InvalidAuth routed into the EXISTING reauth callback (entry.async_start_reauth),
and keep_reconnecting only after a session is up. Setup never blocks on a
transport. A send that cannot run is no longer silent: it is counted, its
error recorded verbatim, and logged at WARNING with the transport state, and
the remote publishes protocol_state / protocol_connected /
protocol_ever_authenticated / protocol_last_error / protocol_failures /
protocol_certificate plus sticky per-push *_seen flags — so "sent" and
"never switched on" can never look alike again.

CASE (b) IS FIXED, NOT DOCUMENTED: switching the transport on in the options
flow for an entry that never paired now routes into the real
async_step_protocol_pair, and "continue without pairing" persists the
transport OFF. One marker carries the pairing (CONF_PROTOCOL_PAIRED),
written by the setup pairing step, the reauth finisher and the reconfigure
finisher, so no entry can sit enabled-but-unpaired. The certificate from the
first pairing is reused by design, so a paired entry is asked for no new PIN
— which is also why the marker, not certificate material, is the source of
truth.

ADB STAYS THE PRIMARY STATE SOURCE. Arming the protocol introduces pushes
that were never live before, so the reading now names its tier:
manager.foreground_source drives the media player's app_source and the
remote's activity_source, and the new tests prove a full protocol push set
leaves the media state, the now-playing reading and every ADB store field
untouched, while an ADB-only entry still serves media player, sensors, apps,
scripts and receiver with only the protocol surface degraded.

Copy: unknown_activity, activity_launch_failed and protocol_unavailable in
BOTH strings.json and translations/en.json, pinned by a test; the README
documents the picker, activity resolution, how to read the transport-state
attributes, the script bypass (a script-routed press is never evidence of
pairing), the connect-then-keep-alive lifecycle and cert reuse versus the
other integration's own pairing.

Evidence: 6 new real-HA FlowManager tests in tests/test_config_flow_real_ha.py
(unpaired -> pairs, PIN -> marker, skip -> off, paired -> not asked again, and
the setup flow's marker on success and on skip), a real-HA entity contract
test in tests/test_ha_api_drift.py (exact supported_features equality with
core, activity_list/current_activity reaching the state machine, live _attr_
updates) and new suites in tests/test_protocol_transport.py (connect-first,
keep_reconnecting alone is a no-op, retry, auth rejection, honest send
failures, ADB-primary precedence) and tests/test_command_routing.py (the
activity contract, turn_on(activity=), script ownership, the transport
attribute states). No device and no browser exist here: device behaviour is
inferred from the library and core source, and every UI-facing claim is
proven at the entity/registry/metadata level under real Home Assistant.

Verify: suites 719/4 and 720/3; real-HA glob (tests/test_adb_command_real_ha.py
+ tests/test_config_flow_real_ha.py) 56 passed, 0 skipped; ruff clean; drift
exit 0 (25 modules, 76 imports, 206 relative imports, 1322 statement blocks,
57 call arguments) with the floors raised 40->70, 100->195, 500->1300 and
52->55.
2026-10-12 12:15:58 +01:00

3012 lines
118 KiB
Python
Raw Blame History

"""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,
CONF_APP_NAME_MAP,
CONF_COMMAND_SCRIPTS,
DEFAULT_APP_DIRECTORY,
TIVIMATE_PACKAGE,
)
def _service_validation_error():
"""HA's ServiceValidationError, as the stub harness installs it (t15).
Read at TEST time: the autouse fixture installs the faces per test, so the class
cannot be bound at import time.
"""
return sys.modules["homeassistant.exceptions"].ServiceValidationError
@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 TestRemoteActivityContract:
"""t15: the app/activity picker HA renders off ``RemoteEntityFeature.ACTIVITY``.
The user compared our remote to the default Android TV app: "it doesn't display
the full set of options" and "the UI ... doesn't actually show any buttons". HA
only offers the picker when the feature flag is set, and the list has to come
from OUR directory — ``AndroidTVRemote`` exposes no app-list API (measured), the
same reason core keeps a user-maintained map.
"""
def test_the_activity_feature_is_declared(self, modules):
entity = _remote_entity(modules)
feature = modules["remote"].RemoteEntityFeature
assert entity.supported_features & feature.ACTIVITY
def test_activity_list_follows_the_directory_order(self, modules):
entity = _remote_entity(modules)
entity._entry.options[CONF_APP_NAME_MAP] = {
"com.a.one": "One",
"com.b.two": "Two",
"com.c.three": "Three",
}
assert modules["remote"].activity_names(
{CONF_APP_NAME_MAP: entity._entry.options[CONF_APP_NAME_MAP]}
) == ["One", "Two", "Three"]
def test_the_curated_table_is_the_floor_and_names_nothing_else(self, modules):
"""Nothing is fabricated: with no user map the curated table IS the list."""
names = modules["remote"].activity_names({})
assert names == list(DEFAULT_APP_DIRECTORY.values())
assert modules["remote"].activity_names({CONF_APP_NAME_MAP: {}}) == names
def test_nameless_discovered_apps_are_skipped_not_doubled(self, modules):
"""Never "pkg=pkg": a row nothing can name is not offered as a name."""
names = modules["remote"].activity_names(
{CONF_APP_NAME_MAP: {"com.a.one": "One"}},
[
{"package": "com.a.one", "name": "One"}, # already listed
{"package": "com.d.four", "name": "Four"}, # a real label
{"package": "com.e.five", "name": ""}, # nameless
{"package": "com.f.six", "name": "com.f.six"}, # id as its name
],
)
assert names == ["One", "Four"]
def test_a_duplicate_name_is_offered_once(self, modules):
names = modules["remote"].activity_names(
{CONF_APP_NAME_MAP: {"com.a.one": "One", "com.b.one": "one", "com.c.two": "Two"}}
)
assert names == ["One", "Two"]
def test_a_blank_row_cannot_be_offered_as_a_name(self, modules):
names = modules["remote"].activity_names(
{CONF_APP_NAME_MAP: {"com.a.one": "One"}},
[{"package": "com.g.seven", "name": " "}],
)
assert names == ["One"]
def test_current_activity_resolves_the_foreground_app(self, modules):
entity = _remote_entity(modules)
entity._entry.options[CONF_APP_NAME_MAP] = {"com.netflix.ninja": "Netflix"}
assert entity._resolve_activity_name() is None, "nothing known yet"
entity._manager.store.set("foreground", "com.netflix.ninja")
assert entity._resolve_activity_name() == "Netflix"
def test_current_activity_uses_the_protocol_push_first(self, modules):
entity = _remote_entity(modules)
entity._entry.options[CONF_APP_NAME_MAP] = {
"com.netflix.ninja": "Netflix",
"com.google.android.youtube.tv": "YouTube",
}
entity._manager.store.set("foreground", "com.google.android.youtube.tv")
entity._manager._on_current_app("com.netflix.ninja")
assert entity._resolve_activity_name() == "Netflix"
assert entity.activity == "com.netflix.ninja"
async def test_turn_on_with_an_activity_launches_it(self, modules):
entity = _remote_entity(modules)
entity._entry.options[CONF_APP_NAME_MAP] = {"com.netflix.ninja": "Netflix"}
entity._manager._on_is_on(False)
protocol = entity._manager._protocol
await entity.async_turn_on(activity="Netflix")
assert protocol.key_calls == [("KEYCODE_POWER", "SHORT")]
assert protocol.launch_calls == ["com.netflix.ninja"]
async def test_turn_on_launches_an_unknown_package_verbatim(self, modules):
"""A package id (or deep link) the directory lacks is still launchable."""
entity = _remote_entity(modules)
entity._entry.options[CONF_APP_NAME_MAP] = {"com.netflix.ninja": "Netflix"}
entity._manager._on_is_on(True) # already on: no POWER
protocol = entity._manager._protocol
await entity.async_turn_on(activity="com.example.other")
assert protocol.key_calls == []
assert protocol.launch_calls == ["com.example.other"]
async def test_an_unknown_activity_name_is_a_clear_error(self, modules):
entity = _remote_entity(modules)
entity._entry.options[CONF_APP_NAME_MAP] = {"com.netflix.ninja": "Netflix"}
protocol = entity._manager._protocol
with pytest.raises(_service_validation_error()) as raised:
await entity.async_turn_on(activity="Netflx")
assert raised.value.translation_key == "unknown_activity"
assert protocol.launch_calls == [], "nothing was silently launched"
def test_an_empty_activity_keeps_the_old_behaviour(self, modules):
entity = _remote_entity(modules)
entity._entry.options[CONF_APP_NAME_MAP] = {"com.netflix.ninja": "Netflix"}
entity._manager._on_is_on(True)
protocol = entity._manager._protocol
asyncio.run(entity.async_turn_on(activity=""))
assert protocol.key_calls == []
assert protocol.launch_calls == []
async def test_a_command_script_still_owns_the_wake_sequence(self, modules):
"""The script masks the protocol: a press through it proves nothing (t15)."""
entity = _remote_entity(modules)
served = []
entity.hass = SimpleNamespace(services=_script_services(served))
entity._entry.options[CONF_COMMAND_SCRIPTS] = {"turn_on": "script.wake"}
entity._manager._on_is_on(False) # the native path would fire POWER
await entity.async_turn_on(activity="com.netflix.ninja")
assert [call[2]["entity_id"] for call in served] == ["script.wake"]
assert entity._manager._protocol.key_calls == [], "the script owns POWER"
assert entity.extra_state_attributes["power_on_route"] == "script:script.wake"
def test_the_transport_attributes_distinguish_the_states(self, modules):
entity = _remote_entity(modules)
attrs = entity.extra_state_attributes
assert attrs["protocol_active"] is True
assert attrs["protocol_state"] == "never_connected", (
"the pre-t15 transport state: wired, never connected. It must never "
"look like a healthy one"
)
assert attrs["protocol_connected"] is False
assert attrs["protocol_ever_authenticated"] is False
assert attrs["protocol_current_app_seen"] is False
assert attrs["protocol_last_error"] is None
assert attrs["protocol_failures"] == 0
assert attrs["power_on_route"] == "protocol"
def test_no_protocol_reads_as_off_not_as_unpaired(self, modules):
entity = _remote_entity(modules)
entity._manager._protocol = None
attrs = entity.extra_state_attributes
assert attrs["protocol_active"] is False
assert attrs["protocol_state"] == "off"
assert attrs["protocol_connected"] is False
def test_a_push_is_remembered_even_after_the_store_expires(self, modules):
entity = _remote_entity(modules)
entity._manager._on_current_app("com.netflix.ninja")
entity._refresh_state()
assert entity.extra_state_attributes["protocol_current_app_seen"] is True
assert entity.extra_state_attributes["protocol_ever_authenticated"] is True
entity._manager.store.clear() # TTL expiry / a cleared store
entity._refresh_state()
assert entity.extra_state_attributes["protocol_current_app_seen"] is True, (
"\"has the current app push EVER arrived?\" cannot be answered from a "
"TTL-capped store"
)
def test_a_failed_launch_is_never_silent(self, modules):
entity = _remote_entity(modules)
protocol = entity._manager._protocol
protocol.fail_all = True
with pytest.raises(_service_validation_error()) as raised:
asyncio.run(entity._async_launch_activity("com.netflix.ninja"))
assert raised.value.translation_key == "activity_launch_failed"
assert entity._manager.protocol_failures == 1
assert "com.netflix.ninja" in (entity._manager.protocol_last_error or "")
async def test_discovery_grows_the_activity_list(self, modules):
entity = _remote_entity(modules)
entity._entry.options[CONF_APP_NAME_MAP] = {"com.a.one": "One"}
entity._attr_activity_list = modules["remote"].activity_names(
entity._entry.options
)
written = []
entity.async_write_ha_state = lambda: written.append(True)
async def _rows():
return [{"package": "com.d.four", "name": "Four"}]
entity._manager.async_refresh_installed_apps = _rows
await entity._async_refresh_activity_list(entity._manager)
assert entity.activity_list == ["One", "Four"]
assert written == [True]
async def test_a_failed_sweep_changes_nothing(self, modules):
"""A failed sweep must not be cached as an answer (t62) — here, not a list."""
entity = _remote_entity(modules)
entity._entry.options[CONF_APP_NAME_MAP] = {"com.a.one": "One"}
entity._attr_activity_list = ["One"]
written = []
entity.async_write_ha_state = lambda: written.append(True)
async def _boom():
raise RuntimeError("sweep failed")
entity._manager.async_refresh_installed_apps = _boom
await entity._async_refresh_activity_list(entity._manager)
assert entity.activity_list == ["One"]
assert written == []
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 TestProtocolArmedKeepsTheAdbLadder:
"""t15: the pushes this unit turns ON must not change what ADB already answers.
The protocol had NEVER been connected here, so every media-session reading the user
relies on arrived while is_on / current_app / volume stayed permanently empty.
Arming the transport adds that traffic; these tests pin the boundary the user's
constraint names — ADB media-session tracking stays the state source, no ADB store
field is touched, and an entry with no protocol at all behaves as before with only
the protocol surface degraded.
"""
@staticmethod
def _adb_player(modules):
"""A playing Netflix session, an ADB foreground, and no media entity."""
from nsh.ha_adb_common import MediaSessionInfo
player, _served, _intents = make_player(modules)
manager = player._manager
manager.store.set("awake", True)
manager.store.set(
"sessions",
[MediaSessionInfo(package="com.netflix.ninja", playing=True, title="Movie")],
)
manager.store.set("foreground", "com.netflix.ninja")
player._read = lambda eid: None
return player, manager
def test_a_full_protocol_push_set_leaves_state_and_now_playing_alone(self, modules):
player, manager = self._adb_player(modules)
before_state = player.state.value
before_now = dict(player._get_adb_now_playing())
# the transport is armed: every push the library offers arrives at once
manager._on_is_available(True)
manager._on_is_on(True)
manager._on_current_app("com.google.android.youtube.tv")
manager._on_volume_info({"level": 5, "max": 15, "muted": False})
assert player.state.value == before_state == "playing"
assert dict(player._get_adb_now_playing()) == before_now
# the ONE field whose documented priority IS the push (P3-2) — and the ADB
# reading it shadows is still in the store, untouched (see the manager suite)
assert manager.store.get("foreground") == "com.netflix.ninja"
assert player.app_id == "com.google.android.youtube.tv"
# ...and the entity NAMES the tier that won, so the push is never a silent
# override of the ADB reading the user's setup is built on (t15)
assert player.extra_state_attributes["app_source"] == "protocol"
def test_a_protocol_reported_off_is_off_while_adb_keeps_its_truth(self, modules):
"""A device that says it is off outranks a stale session — newly reachable.
Before t15 this signal could not arrive at all, so an off Shield kept showing
its last session as playing. The ADB fields are still there (the push writes
only its own), which is why a later poll answers immediately again.
"""
player, manager = self._adb_player(modules)
manager._on_is_on(False)
assert player.state.value == "off"
assert manager.store.get("sessions"), "the ADB session reading was not cleared"
def test_an_adb_only_entry_still_answers_everything_it_did(self, modules):
"""Protocol disabled: only the protocol surface degrades (the t40 ladder)."""
player, manager = self._adb_player(modules)
manager._protocol = None # the entry never enabled the transport
remote = _remote_entity(modules)
remote._manager = manager
remote._entry.runtime_data = manager
remote._entry.options = dict(player._entry.options)
remote._refresh_state() # the manager ping that re-resolves the activity
assert player.state.value == "playing"
assert player.app_name == "Netflix" # curated table, no protocol involved
assert player.source_list # the app dropdown is independent of the transport
assert remote.available is True # a passive transport is not a broken one
assert remote.activity == "com.netflix.ninja" # the ADB parse names it
assert remote.current_activity == "Netflix"
assert remote.activity_list == player.source_list
assert remote.extra_state_attributes["protocol_state"] == "off"
assert remote.extra_state_attributes["protocol_connected"] is False
# an ADB-named activity says so: the tier is disclosed, not implied
assert remote.extra_state_attributes["activity_source"] == "adb"
def test_the_new_exception_keys_have_copy_in_both_files():
"""A raised translation_key with no copy shows the user a bare key (t15).
The three keys the remote raises: an activity name nothing can launch, a launch the
protocol could not run, and an entry with no protocol transport at all.
"""
import json
from pathlib import Path
root = (
Path(__file__).resolve().parent.parent
/ "custom_components"
/ "nvidia_shield"
)
for name in ("strings.json", "translations/en.json"):
exceptions = json.loads((root / name).read_text(encoding="utf-8"))["exceptions"]
for key in ("unknown_activity", "activity_launch_failed", "protocol_unavailable"):
assert exceptions[key]["title"], (name, key)
assert exceptions[key]["message"], (name, key)
# the placeholder the entity fills in is the one the copy names
assert "{activity}" in exceptions["unknown_activity"]["message"]
assert "{activity}" in exceptions["activity_launch_failed"]["message"]
# ...and the message tells the user where the answer is readable
assert "protocol_state" in exceptions["activity_launch_failed"]["message"]
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()
# t66: the identity tier now resolves through the shared chain, so the app
# reads by its name — still the APP, never the channel or the media title.
assert np["title"] == "TiviMate"
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" # but identity stays the app
assert "Sky News" not in np.values() # the media title never leaks
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
# t15: mirrors __init__ — the activity contract and the transport evidence the
# attribute payload publishes (a __new__-built entity skips __init__).
entity._attr_supported_features = remote_module.RemoteEntityFeature.ACTIVITY
entity._attr_activity_list = remote_module.activity_names(entry.options)
entity._attr_current_activity = None
entity._protocol_pushes_seen = set()
entity._protocol_certificate = "unknown"
entity._activity_refresh_task = None
return entity
async def _noop_async(*args, **kwargs):
return None
def _script_services(served):
"""A hass.services stand-in that records (domain, service, data) (t15)."""
class _RecordingServices:
async def async_call(self, domain, service, data=None, **kwargs):
served.append((domain, service, data))
return _RecordingServices()
class FakeServices:
"""hass.services stand-in recording registrations."""
def __init__(self):
self.registered: dict[tuple, object] = {}
self.service_calls: list[tuple] = []
#: t14: what each registration declared for the response.
self.response_support: dict[tuple, object] = {}
def has_service(self, domain, service):
return (domain, service) in self.registered
def async_register(
self, domain, service, func, schema=None, supports_response=None
):
"""t14: keyword-faithful to HA — ``supports_response`` is passed here."""
self.registered[(domain, service)] = (func, schema)
#: Kept beside the 2-tuple (readers assert on that shape) so a test can see
#: what the registration declared for the response (t14: OPTIONAL).
self.response_support[(domain, service)] = supports_response
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
# nothing selected → the target-less error, naming what to pass
with pytest.raises(ServiceValidationError) as err:
await _async_adb_command_service(_Call({"command": "echo hi"}))
assert err.value.translation_key == "no_target"
# unknown entity → unsupported_target
with pytest.raises(ServiceValidationError) as err:
await _async_adb_command_service(
_Call({"command": "echo hi", "entity_id": "media_player.other"})
)
assert err.value.translation_key == "unsupported_target"
# valid target → the shell runs, and the output comes back in the RESPONSE
result = await _async_adb_command_service(
_Call({"command": "echo hi", "entity_id": "media_player.shield_test"})
)
assert shell_calls == ["echo hi"]
assert result == {"adb_response": "ok"}
_ADB_COMMAND_TARGETS.pop("media_player.shield_test", None)
async def test_the_ui_target_shapes_resolve_to_exactly_one_shield(self, modules):
"""t14: the picker's value (text OR list) selects; two is an error.
HA merges the UI's target into the call as ``entity_id``; a picked target
arrives as a LIST. Both shapes must reach the same single Shield, and a
selection that covers two Shields must send nothing.
"""
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
_ADB_COMMAND_TARGETS["media_player.other"] = make_player(modules)[0]._manager
shell_calls = []
async def _shell(cmd):
shell_calls.append(cmd)
return "out"
player._manager._poller.shell = _shell
# the string form a YAML caller passes
assert await _async_adb_command_service(
_Call({"command": "echo a", "entity_id": "media_player.shield_test"})
) == {"adb_response": "out"}
# the list form the UI's picker produces for one picked entity
assert await _async_adb_command_service(
_Call({"command": "echo b", "entity_id": ["media_player.shield_test"]})
) == {"adb_response": "out"}
assert shell_calls == ["echo a", "echo b"]
# two selected → ambiguous_target, and NOTHING was sent to either
with pytest.raises(ServiceValidationError) as err:
await _async_adb_command_service(
_Call({
"command": "echo c",
"entity_id": ["media_player.shield_test", "media_player.other"],
})
)
assert err.value.translation_key == "ambiguous_target"
assert shell_calls == ["echo a", "echo b"]
_ADB_COMMAND_TARGETS.pop("media_player.shield_test", None)
_ADB_COMMAND_TARGETS.pop("media_player.other", None)
async def test_a_device_target_resolves_through_the_entity_registry(self, modules):
"""t14: `device_id` (what an automation targeting a DEVICE sends) works.
It expands only through this integration's own registry entries and only to
media players, so a device that carries no participant is an error, never a
guess.
"""
player, _s = self._participant(modules, enabled=True)
from homeassistant.exceptions import ServiceValidationError
from nsh.media_player import _ADB_COMMAND_TARGETS, _async_adb_command_service
registry = sys.modules["homeassistant.helpers.entity_registry"]
_ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager
registry.bind_device("dev_shield", "media_player.shield_test")
# the same device shape the real registry produces: sensors next to the player,
# and a player whose entry has the option OFF (not a participant)
registry.bind_device("dev_with_sensors", "media_player.shield_test", "sensor.shield_temp")
registry.bind_device("dev_off", "media_player.shield_off")
registry.bind_device("dev_sensor", "sensor.shield_temperature")
shell_calls = []
async def _shell(cmd):
shell_calls.append(cmd)
return "via device"
player._manager._poller.shell = _shell
try:
assert await _async_adb_command_service(
_Call({"command": "echo d", "device_id": "dev_shield"}, player.hass)
) == {"adb_response": "via device"}
assert shell_calls == ["echo d"]
# a device whose media player does not participate → error naming it
with pytest.raises(ServiceValidationError) as err:
await _async_adb_command_service(
_Call({"command": "echo e", "device_id": "dev_off"}, player.hass)
)
assert err.value.translation_key == "unsupported_target"
assert "media_player.shield_off" in err.value.translation_placeholders[
"entity_id"
]
assert shell_calls == ["echo d"]
# ...while the sensors of a participating device do not become targets
assert await _async_adb_command_service(
_Call(
{"command": "echo f", "device_id": "dev_with_sensors"},
player.hass,
)
) == {"adb_response": "via device"}
assert shell_calls == ["echo d", "echo f"]
# a device with no media player at all cannot be a target
with pytest.raises(ServiceValidationError) as err:
await _async_adb_command_service(
_Call({"command": "echo g", "device_id": "dev_sensor"}, player.hass)
)
assert err.value.translation_key == "no_target"
assert shell_calls == ["echo d", "echo f"]
finally:
registry.reset_bindings()
_ADB_COMMAND_TARGETS.pop("media_player.shield_test", None)
async def test_the_registration_publishes_the_target_and_response(self, modules):
"""t14: what async_setup tells HA — the picker and the response flag."""
_player, services = self._participant(modules, enabled=False)
from nsh.const import (
ADB_COMMAND_SERVICE_DESCRIPTION,
ADB_COMMAND_SERVICE_TARGET,
)
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), {})
# the registration declares that a RESPONSE is available (OPTIONAL: callers
# that do not ask are unaffected)
assert (
services.response_support[("nvidia_shield", "adb_command")] == "optional"
)
# and the runtime DESCRIPTION — the only place a target can be declared —
# carries the picker: the Shield is selected as its media_player entity
calls = [
call
for call in entity_stublib.SERVICE_SCHEMA_CALLS
if call[0] == "nvidia_shield" and call[1] == "adb_command"
]
assert calls, "async_setup must publish the adb_command description"
schema = calls[-1][2]
assert schema is ADB_COMMAND_SERVICE_DESCRIPTION
assert schema["target"] is ADB_COMMAND_SERVICE_TARGET
assert schema["target"] == {"entity": [{"domain": "media_player"}]}
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:
"""A service call as the handler sees it: ``data``, plus the ``hass`` a
device_id needs to resolve through the entity registry (t14)."""
def __init__(self, data, hass=None):
self.data = data
self.hass = hass
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):
# t66: a package NOT in the curated table, so the fall-through under test
# is really the raw id and not a curated name.
player, manager, sensor, poller = self._wire(
modules,
"com.example.labelled",
{"com.example.labelled": "application-label: 'Labelled App'"},
)
# first render: cache miss → the sensor falls through to the package
assert sensor.native_value == "com.example.labelled"
assert manager.app_label_requests == {"com.example.labelled"}
# the watchdog drains the fetch for the CURRENT package
await manager._watchdog_cycle(5.0)
await asyncio.sleep(0)
assert poller.dumps and "com.example.labelled" in poller.dumps[0]
# next render: the resolved device name, in the sensor AND the player
assert sensor.native_value == "Labelled App"
assert player.app_name == "Labelled App"
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,
"com.example.labelled",
{"com.example.labelled": "application-label: 'Labelled App'"},
)
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 == "com.example.labelled"
await manager._watchdog_cycle(5.0)
await asyncio.sleep(0)
assert "com.example.labelled" not in manager._app_labels # nothing stamped
assert sensor.native_value == "com.example.labelled"
# the next render re-enqueues, and the retry is what recovers the label
assert manager.app_label("com.example.labelled") is None
assert manager.app_label_requests == {"com.example.labelled"}
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 == "Labelled App"
assert sensor.extra_state_attributes["app_name"] == "Labelled App"
assert player.app_name == "Labelled App"
@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.example.labelled",
{"com.example.labelled": "application-label: 'Labelled App'"},
)
async def _blank(_cmd):
poller.dumps.append(_cmd)
return blank
poller.shell = _blank
assert player.app_name == "com.example.labelled"
await manager._watchdog_cycle(5.0)
await asyncio.sleep(0)
assert "com.example.labelled" not in manager._app_labels
assert player.app_name == "com.example.labelled"
assert manager.app_label("com.example.labelled") 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 == "Labelled App"
assert sensor.native_value == "Labelled App"
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,
"com.example.one",
{
"com.example.one": "application-label: 'One TV'",
"com.example.two": "application-label: 'Two TV'",
},
)
assert sensor.native_value == "com.example.one" # miss for the 1st app
manager._on_current_app("com.example.two")
assert sensor.native_value == "com.example.two"
# 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 "com.example.one" in poller.dumps[0]
assert "com.example.two" in poller.dumps[1]
assert sensor.native_value == "Two TV"
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"}
Reference in new issue View Git Blame Copy Permalink