James Mackay
89c05f1dcd
The user's remaining universal-media-player dependency: some commands need hardware this integration deliberately does not drive — a speaker switch that owns the volume, an AVR input that has to move with the Shield, a mute that has to hit two devices. One script per command is the escape hatch, and the rule is absolute: a configured command calls its script INSTEAD of the native protocol/ADB path, and with nothing configured every command behaves exactly as it did (the whole existing suite passes untouched, which is what "byte-identical" means here). * const.py owns the option: CONF_COMMAND_SCRIPTS (one dict, keyed by command), SCRIPT_COMMANDS (turn_on, turn_off, volume_up, volume_down, volume_set, volume_mute, select_source) and command_scripts(), the ONE normalizer of the stored shape — options round-trip through HA storage and can be hand-edited, so a value that is not a non-empty string for a known command is dropped instead of reaching the service layer. * media_player.py: async_run_command_script() calls script.turn_on with the script entity and its variables, blocking=True so a failure surfaces HERE — a mapped command that cannot run raises HomeAssistantError naming the command and script rather than falling back silently or doing nothing. Every command method checks its script first; volume_set passes `level` and select_source passes `app_name`. * volume_mute also passes `is_volume_muted`: without it a mute script cannot tell muting from unmuting, which is the case the user asked for. * volume_set and select_source have NO native path (C-VOL; app launching goes through play_media), so they are advertised only when configured — the feature flag is what makes HA offer the command at all. Unset leaves the feature set exactly as it was. * remote.py shares async_run_command_script for POWER: turn_on/turn_off is one configuration and the remote entity is the other route to it, so a script that owns the power sequence owns it on both. * config_flow.py: one entity selector per command (domain: script), stored as the single dict option, prefilled from the entry's options and cleared by saving without a value — the editor is the truth, as with the app map editor. Two things the harness caught, worth recording: `MediaPlayerEntityFeature` has no SET_VOLUME (the real member is VOLUME_SET — the stub enum and the drift engine disagree with a wrong name), and the stub `homeassistant.exceptions` had no HomeAssistantError, so the stub now mirrors the real inheritance (ServiceValidationError derives from it). tests/test_command_scripts.py covers each command mapped and unmapped, the exact variable payloads, the native path untouched when unmapped (protocol key, then the ADB fallthrough with the protocol down), the rejecting script's error, hand-edited option shapes, and the feature set. test_config_flow_ux.py covers the selectors, the stored shape, prefill and clearing. New tests leaked the harness into the next file at first (a missing per-file scope); that is fixed, and tests/test_harness_scoping.py now asserts that every file loading the harness scopes it, so the next such mistake fails where it is made. Verified: 401 passed / 2 skipped, 402 / 1 with androidtvremote2, reverse file order 401 / 2, ruff clean, and the drift engine exits 0 on a real-HA interpreter (23 modules, 59 imports, 157 relative imports, 36 call arguments, 6 awaited HA calls).
1381 lines
52 KiB
Python
1381 lines
52 KiB
Python
"""t10 tests: command routing, play_media channel port, remote entity, volume.
|
|
|
|
Concrete fakes expose the ordering precisely: protocol → ADB keyevent →
|
|
legacy external entity, plus the §7 channel digit sequence (protocol digits,
|
|
~0.1 s apart, no ENTER, new sequence cancels in-flight). A scripted media
|
|
layer marks which tier served each command.
|
|
"""
|
|
|
|
import ast
|
|
import asyncio
|
|
import sys
|
|
|
|
import entity_stublib
|
|
import pytest
|
|
from conftest import _SHIELD_DIR
|
|
from entity_stublib import install, load_entity_modules
|
|
from nsh.connection import ShieldConnectionManager
|
|
from nsh.const import TIVIMATE_PACKAGE
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def scoped_entity_stubs():
|
|
"""Container for the HA stubs + loaded entity modules (t7 discipline)."""
|
|
snapshot = entity_stublib.snapshot_modules()
|
|
install()
|
|
yield load_entity_modules()
|
|
entity_stublib.restore_modules(snapshot)
|
|
|
|
|
|
class FakeProtocolClient:
|
|
"""Records key/launch calls; optional scripted failures."""
|
|
|
|
def __init__(self):
|
|
self.key_calls: list[tuple[str, str]] = []
|
|
self.launch_calls: list[str] = []
|
|
self.fail_keys: set[str] = set()
|
|
self.fail_all = False
|
|
|
|
def send_key_command(self, key, direction="SHORT"):
|
|
self.key_calls.append((key, direction))
|
|
if self.fail_all or key.upper() in self.fail_keys:
|
|
raise ConnectionError("protocol transport down")
|
|
|
|
def send_launch_app_command(self, app_link_or_app_id: str) -> None:
|
|
self.launch_calls.append(app_link_or_app_id)
|
|
if self.fail_all:
|
|
raise ConnectionError("transport down")
|
|
|
|
|
|
class FakeAdbPoller:
|
|
"""Minimal ADB surface for keyevent and digit fallback paths."""
|
|
|
|
def __init__(self):
|
|
self.media_calls: list[str] = []
|
|
self.digits: list[str] = []
|
|
self.enter_calls: list[bool] = []
|
|
self.results: dict[str, bool] = {}
|
|
self.digit_result = True
|
|
|
|
@property
|
|
def connected(self):
|
|
return True
|
|
|
|
async def send_media_command(self, command: str) -> bool:
|
|
self.media_calls.append(command)
|
|
return self.results.get(command, True)
|
|
|
|
async def send_digit_keyevent(self, digit: str) -> bool:
|
|
self.digits.append(digit)
|
|
return self.digit_result
|
|
|
|
async def send_enter_keyevent(self) -> bool:
|
|
self.enter_calls.append(True)
|
|
return True
|
|
|
|
async def bring_to_foreground(self, package: str) -> bool:
|
|
return True
|
|
|
|
|
|
def make_manager(store=None, poller=None, protocol=None):
|
|
return ShieldConnectionManager(
|
|
hass=None,
|
|
entry=None,
|
|
poller=poller or FakeAdbPoller(),
|
|
protocol=protocol or FakeProtocolClient(),
|
|
store=store,
|
|
)
|
|
|
|
|
|
@pytest.fixture(name="modules")
|
|
def load_modules_fixture():
|
|
return load_entity_modules()
|
|
|
|
|
|
def make_player(modules, manager=None, jellyfin_entity=None, options=None):
|
|
"""A pre-initialised media_player entity bound to the tested manager."""
|
|
from types import SimpleNamespace
|
|
|
|
mp_module = modules["media_player"]
|
|
if manager is None:
|
|
manager = make_manager()
|
|
entry = SimpleNamespace(entry_id="e1", options={}, data={}, title="mdarcy")
|
|
manager.entry = entry
|
|
|
|
def _set_option(key, value):
|
|
entry.options[key] = value
|
|
|
|
entry.set_option = _set_option
|
|
player = mp_module.NvidiaShieldMediaPlayer.__new__(mp_module.NvidiaShieldMediaPlayer)
|
|
player.hass = None
|
|
player._entry = entry
|
|
player._manager = manager
|
|
player._intent = None
|
|
player._intent_timer_unsub = None # mirrors __init__ (intent self-heal)
|
|
player._last_channel = None
|
|
player.hass = entity_stublib.FakeHass()
|
|
player._entity_tasks = set()
|
|
player._channel_task = None
|
|
player.async_write_ha_state = lambda: None
|
|
player._read = lambda eid: None
|
|
|
|
# t41: nothing may leave this integration, so an HA service call is now the
|
|
# failure signal rather than a supported tier. `served` stays empty unless a
|
|
# regression reintroduces the external-entity delegation.
|
|
served = []
|
|
|
|
class _RecordingServices:
|
|
async def async_call(self, domain, service, data=None, **kwargs):
|
|
# t44: the payload matters now (command scripts pass HA script
|
|
# variables), so record it too — every existing assertion is `== []`.
|
|
served.append((domain, service, data))
|
|
|
|
player.hass.services = _RecordingServices()
|
|
|
|
captured_intents = []
|
|
|
|
def _set_intent(service, target):
|
|
intent = SimpleNamespace(
|
|
target_state=getattr(target, "value", target),
|
|
command=service,
|
|
created_at=0.0,
|
|
prior_state=None,
|
|
retry_count=0,
|
|
via_adb=False,
|
|
via_protocol=False,
|
|
adb_command=None,
|
|
protocol_key=None,
|
|
)
|
|
|
|
def _unsub():
|
|
return None
|
|
|
|
player._intent = intent
|
|
captured_intents.append(intent)
|
|
return None
|
|
|
|
player._set_intent = _set_intent
|
|
return player, served, captured_intents
|
|
|
|
|
|
class TestMediaCommandRouting:
|
|
async def test_protocol_serves_media_command(self, modules):
|
|
player, served, _intents = make_player(modules)
|
|
protocol = player._manager._protocol
|
|
player._intent = None
|
|
|
|
player.async_write_ha_state = lambda: None
|
|
|
|
await player._media_cmd("media_play", "play", "playing")
|
|
|
|
assert protocol.key_calls == [("KEYCODE_MEDIA_PLAY", "SHORT")]
|
|
assert served == []
|
|
|
|
async def test_intent_records_protocol_path_for_retries(self, modules):
|
|
player, _served, _intents = make_player(modules)
|
|
player._set_intent("media_play", type("S", (), {"value": "playing"})())
|
|
player.async_write_ha_state = lambda: None
|
|
await player._media_cmd("media_play", "play", "playing")
|
|
# retry metadata stamped: the protocol path is recorded on the intent
|
|
assert player._intent is not None
|
|
assert player._intent.via_protocol is True
|
|
assert player._intent.protocol_key == "KEYCODE_MEDIA_PLAY"
|
|
|
|
|
|
class TestChannelDigits:
|
|
async def test_digit_sequence_digit_part_delivery(self, modules):
|
|
player, served, _intents = make_player(modules)
|
|
protocol = player._manager._protocol
|
|
|
|
await player._channel_entry_sequence(["4", "0", "9", "6"])
|
|
|
|
# protocol digits: KEYCODE_{digit} SHORT each
|
|
assert protocol.key_calls == [
|
|
("KEYCODE_4", "SHORT"),
|
|
("KEYCODE_0", "SHORT"),
|
|
("KEYCODE_9", "SHORT"),
|
|
("KEYCODE_6", "SHORT"),
|
|
]
|
|
assert served == [] # never fell through to legacy
|
|
|
|
async def test_new_channel_cancels_inflight_sequence(self, modules):
|
|
player, _served, _intents = make_player(modules)
|
|
protocol = player._manager._protocol
|
|
|
|
# start a long sequence (real path), then a new play_media replaces it
|
|
first_task = player._track_task(player._channel_entry_sequence(["1", "2", "3"]))
|
|
player._channel_task = first_task # as async_play_media tracks it
|
|
|
|
await player.async_play_media("channel", "98")
|
|
await asyncio.sleep(0.02)
|
|
|
|
assert first_task.cancelled()
|
|
assert "KEYCODE_2" not in [k for k, _d in protocol.key_calls]
|
|
player._cancel_inflight_channel() # test cleanup
|
|
|
|
async def test_fallback_to_adb_digits_when_protocol_down(self, modules):
|
|
player, served, _intents = make_player(modules)
|
|
protocol = player._manager._protocol
|
|
protocol.fail_all = True
|
|
|
|
await player._channel_entry_sequence(["5"])
|
|
|
|
assert protocol.key_calls == [("KEYCODE_5", "SHORT")]
|
|
assert player._manager._poller.digits == ["5"] # ADB digit keyevent served
|
|
assert served == []
|
|
|
|
async def test_optional_enter_falls_back_to_adb_keyevent(self, modules):
|
|
player, served, _i = make_player(modules)
|
|
protocol = player._manager._protocol
|
|
poller = player._manager._poller
|
|
protocol.fail_keys.add("KEYCODE_ENTER")
|
|
protocol.key_calls.clear()
|
|
|
|
player._entry.options["channel_send_enter"] = True # §7 optional ENTER on
|
|
|
|
await player._channel_entry_sequence(["4", "1"]) # digits fine, ENTER fails
|
|
|
|
# digits served by protocol; ENTER retried over ADB keyevent 66
|
|
assert ("KEYCODE_4", "SHORT") in protocol.key_calls
|
|
assert poller.digits == [] # digits were fine over the protocol
|
|
assert poller.enter_calls == [True]
|
|
assert served == []
|
|
|
|
async def test_legacy_fallback_when_both_transports_down(self, modules):
|
|
player, served, _intents = make_player(modules)
|
|
protocol = player._manager._protocol
|
|
protocol.fail_all = True
|
|
player._manager._poller.digit_result = False
|
|
|
|
await player._channel_entry_sequence(["7"])
|
|
|
|
# t41: both transports down is a no-op — no other integration is called
|
|
assert served == []
|
|
|
|
|
|
class TestAppAndUrls:
|
|
async def test_app_launch_calls_protocol(self, modules):
|
|
player, served, _intents = make_player(modules)
|
|
protocol = player._manager._protocol
|
|
|
|
await player.async_play_media("app", "com.plex.hdx")
|
|
|
|
assert protocol.launch_calls == ["com.plex.hdx"] # C-LAUNCH auto-prefix
|
|
assert served == []
|
|
|
|
async def test_url_launch_passes_through_unchanged(self, modules):
|
|
player, served, _intents = make_player(modules)
|
|
protocol = player._manager._protocol
|
|
|
|
await player.async_play_media("url", "plex://open")
|
|
|
|
assert protocol.launch_calls == ["plex://open"]
|
|
|
|
async def test_unknown_media_type_no_send(self, modules):
|
|
player, served, _intents = make_player(modules)
|
|
protocol = player._manager._protocol
|
|
|
|
await player.async_play_media("spotify_uri", "spotify:track:x")
|
|
|
|
assert protocol.launch_calls == []
|
|
assert protocol.key_calls == []
|
|
assert served == []
|
|
|
|
|
|
class TestRemoteEntity:
|
|
async def test_split_command_prefixes(self, modules):
|
|
module = modules["remote"]
|
|
assert module._split_command("SHORT:home") == ("home", "SHORT")
|
|
assert module._split_command("START_LONG:home") == ("home", "START_LONG")
|
|
assert module._split_command("END_LONG:home") == ("home", "END_LONG")
|
|
assert module._split_command("home") == ("home", "SHORT")
|
|
|
|
async def test_key_code_name_mapping(self, modules):
|
|
module = modules["remote"]
|
|
assert module._key_code_name("home") == "KEYCODE_HOME"
|
|
assert module._key_code_name("KEYCODE_HOME") == "KEYCODE_HOME"
|
|
assert module._key_code_name("KEYCODE_HDMI_1") == "KEYCODE_HDMI_1"
|
|
assert module._key_code_name("play") == "KEYCODE_MEDIA_PLAY"
|
|
|
|
async def test_send_command_hits_protocol(self, modules):
|
|
entity = _remote_entity(modules)
|
|
protocol = entity._manager_of()._protocol
|
|
|
|
await entity.async_send_command(["HOME"])
|
|
|
|
assert protocol.key_calls == [("KEYCODE_HOME", "SHORT")]
|
|
|
|
async def test_send_command_honors_hold_secs(self, modules):
|
|
entity = _remote_entity(modules)
|
|
protocol = entity._manager_of()._protocol
|
|
|
|
await entity.async_send_command(["home"], hold_secs=0.05)
|
|
|
|
assert protocol.key_calls == [
|
|
("KEYCODE_HOME", "START_LONG"),
|
|
("KEYCODE_HOME", "END_LONG"),
|
|
]
|
|
|
|
async def test_send_command_honors_num_repeats(self, modules):
|
|
entity = _remote_entity(modules)
|
|
protocol = entity._manager_of()._protocol
|
|
|
|
await entity.async_send_command(["home"], num_repeats=3)
|
|
|
|
assert len(protocol.key_calls) == 3
|
|
|
|
async def test_turn_on_off_send_power_key_respecting_is_on(self, modules):
|
|
entity = _remote_entity(modules)
|
|
manager = entity._manager_of()
|
|
protocol = manager._protocol
|
|
|
|
# push "already off" — turn_on should send POWER, turn_off should not
|
|
manager._on_is_on(False)
|
|
await entity.async_turn_on()
|
|
assert protocol.key_calls == [("KEYCODE_POWER", "SHORT")]
|
|
|
|
protocol.key_calls.clear()
|
|
await entity.async_turn_off()
|
|
assert protocol.key_calls == []
|
|
|
|
# push "on" — turn_off should send POWER, turn_on should not
|
|
manager._on_is_on(True)
|
|
protocol.key_calls.clear()
|
|
await entity.async_turn_off()
|
|
assert protocol.key_calls == [("KEYCODE_POWER", "SHORT")]
|
|
protocol.key_calls.clear()
|
|
await entity.async_turn_on()
|
|
assert protocol.key_calls == []
|
|
|
|
|
|
class TestIntentRetryWithoutATransport:
|
|
"""t41: an intent no transport served surrenders instead of delegating."""
|
|
|
|
def _intent(self, **overrides):
|
|
from types import SimpleNamespace
|
|
|
|
intent = SimpleNamespace(
|
|
target_state="playing",
|
|
command="media_play",
|
|
created_at=0.0,
|
|
prior_state=None,
|
|
retry_count=0,
|
|
via_adb=False,
|
|
via_protocol=False,
|
|
adb_command=None,
|
|
protocol_key=None,
|
|
)
|
|
for key, value in overrides.items():
|
|
setattr(intent, key, value)
|
|
return intent
|
|
|
|
async def test_unserved_intent_surrenders_without_a_service_call(self, modules):
|
|
player, served, _intents = make_player(modules)
|
|
player.async_write_ha_state = lambda: None
|
|
player._intent = self._intent()
|
|
|
|
player._on_intent_timeout(0.0) # async_call_later passes the timestamp
|
|
|
|
assert player._intent is None # surrendered to the truth
|
|
assert served == [] # and nothing left this integration
|
|
|
|
async def test_protocol_served_intent_still_retries_over_the_protocol(self, modules):
|
|
player, served, _intents = make_player(modules)
|
|
player.async_write_ha_state = lambda: None
|
|
protocol = player._manager._protocol
|
|
player._intent = self._intent(
|
|
via_protocol=True, protocol_key="KEYCODE_MEDIA_PLAY"
|
|
)
|
|
|
|
player._on_intent_timeout(0.0) # async_call_later passes the timestamp
|
|
await asyncio.sleep(0.05)
|
|
|
|
assert protocol.key_calls == [("KEYCODE_MEDIA_PLAY", "SHORT")]
|
|
assert served == []
|
|
|
|
|
|
class TestAppFallback:
|
|
"""t41: the app identity comes from OUR transports' store only."""
|
|
|
|
async def test_app_id_comes_from_the_manager_store(self, modules):
|
|
player, _s, _i = make_player(modules)
|
|
player._manager._on_current_app("com.plex.hdx")
|
|
await asyncio.sleep(0.05) # drain the call_soon_threadsafe push
|
|
|
|
assert player.app_id == "com.plex.hdx"
|
|
assert player.app_name == "com.plex.hdx" # package is truth; map later
|
|
|
|
async def test_app_follows_the_latest_protocol_push(self, modules):
|
|
player, _s, _i = make_player(modules)
|
|
player._manager._on_current_app("com.netflix.ninja")
|
|
await asyncio.sleep(0.05)
|
|
player._manager._on_current_app("com.google.android.youtube.tv")
|
|
await asyncio.sleep(0.05)
|
|
|
|
assert player.app_id == "com.google.android.youtube.tv"
|
|
|
|
|
|
def _ladder_player(modules, foreground=None, audio=None):
|
|
"""Standalone player with a foreground (and optional audio hint) set."""
|
|
player, _s, _i = make_player(modules)
|
|
manager = player._manager
|
|
manager.store.set("awake", True)
|
|
if foreground:
|
|
manager.store.set("foreground", foreground)
|
|
if audio:
|
|
manager.store.set("audio_playback", audio)
|
|
player._read = lambda eid: None
|
|
return player, manager
|
|
|
|
|
|
class TestForegroundLadderTier:
|
|
"""P3-3a: the ladder branch between the media-session tier and ON."""
|
|
|
|
def _player(self, modules, sessions=None, foreground=None, audio=None):
|
|
player, _s, _i = make_player(modules)
|
|
manager = player._manager
|
|
manager.store.set("awake", True)
|
|
if sessions:
|
|
manager.store.set("sessions", sessions)
|
|
if foreground:
|
|
manager.store.set("foreground", foreground)
|
|
if audio:
|
|
manager.store.set("audio_playback", audio)
|
|
player._read = lambda eid: None
|
|
return player, manager
|
|
|
|
async def test_session_app_serves_playing_via_session_tier(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
sessions = [
|
|
MediaSessionInfo(package="com.netflix.ninja", playing=True, title="Movie")
|
|
]
|
|
player, manager = self._player(
|
|
modules, sessions=sessions, foreground="com.netflix.ninja"
|
|
)
|
|
|
|
assert player.state.value == "playing" # session tier, not the fg tier
|
|
np = player._get_adb_now_playing()
|
|
assert np["source"] == "media_session"
|
|
assert np["title"] == "Movie"
|
|
|
|
async def test_no_session_app_with_audio_hint(self, modules):
|
|
player, _m = self._player(
|
|
modules,
|
|
foreground="ar.tvplayer.tv",
|
|
audio=["ar.tvplayer.tv"],
|
|
)
|
|
|
|
# TiviMate publishes no session (R2) — fg tier keeps the device ON
|
|
assert player.state.value == "on"
|
|
np = player._get_app_now_playing()
|
|
assert np == {
|
|
"state": "idle",
|
|
"title": "Media recently active",
|
|
"source": "app",
|
|
}
|
|
|
|
async def test_no_session_app_without_hint_returns_none(self, modules):
|
|
player, _m = self._player(modules, foreground="com.spocky.projengmenu")
|
|
|
|
assert player.state.value == "on" # launcher foreground → still ON
|
|
assert player._get_app_now_playing() is None
|
|
|
|
async def test_last_known_title_beats_audio_hint(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, manager = self._player(
|
|
modules, foreground="ar.tvplayer.tv", audio=["ar.tvplayer.tv"]
|
|
)
|
|
manager._clock = lambda: 0.0 # stamp 0.0, clock 0.0 → inside TTL
|
|
info = MediaSessionInfo(package="ar.tvplayer.tv", playing=True, title="Sky News")
|
|
manager._last_known["ar.tvplayer.tv"] = (info, 0.0)
|
|
|
|
np = player._get_app_now_playing()
|
|
assert np == {"state": "playing", "title": "Sky News", "source": "app"}
|
|
|
|
async def test_ttl_resurrect_guard_in_the_tier(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
now = 1000.0
|
|
player, manager = self._player(modules, foreground="ar.tvplayer.tv")
|
|
player._last_channel = "101"
|
|
|
|
info = MediaSessionInfo(package="ar.tvplayer.tv", playing=True, title="Sky News")
|
|
manager._clock = lambda: now # the closure reads the live `now` cell
|
|
manager._last_known["ar.tvplayer.tv"] = (info, now)
|
|
|
|
# inside TTL → last-known title served
|
|
np = player._get_app_now_playing()
|
|
assert np == {"state": "playing", "title": "Sky News", "source": "app"}
|
|
|
|
# past STATE_STORE_TTL → 'playing' must NOT be resurrected;
|
|
# the tier falls back to the last channel instead
|
|
now = now + 61.0
|
|
np = player._get_app_now_playing()
|
|
assert np == {"state": "idle", "title": "101", "source": "app"}
|
|
|
|
|
|
class TestChannelLeakGuard:
|
|
"""F1: the channel number never leaks to a non-channel foreground app."""
|
|
|
|
async def test_launcher_never_shows_the_channel_number(self, modules):
|
|
player, _m = _ladder_player(
|
|
modules, foreground="com.spocky.projengmenu"
|
|
)
|
|
player._last_channel = "101" # entered in TiviMate earlier
|
|
|
|
np = player._get_app_now_playing()
|
|
|
|
assert np is None # the leak is gone
|
|
|
|
async def test_channel_app_still_gets_the_channel_title(self, modules):
|
|
player, _m = _ladder_player(modules, foreground=TIVIMATE_PACKAGE)
|
|
|
|
player._last_channel = "101"
|
|
np = player._get_app_now_playing()
|
|
|
|
assert np == {"state": "idle", "title": "101", "source": "app"}
|
|
|
|
async def test_audio_member_can_carry_the_channel_title(self, modules):
|
|
player, manager = _ladder_player(
|
|
modules, foreground="ar.tvplayer.tv", audio=["ar.tvplayer.tv"]
|
|
)
|
|
|
|
player._last_channel = "205"
|
|
np = player._get_app_now_playing()
|
|
|
|
assert np["title"] == "205"
|
|
|
|
|
|
class TestPrivacyOrdering:
|
|
"""F3: privacy mode outranks the channel fallback — identity stays the app."""
|
|
|
|
async def test_channel_number_never_leaks_in_privacy_mode(self, modules):
|
|
player, manager = _ladder_player(modules, foreground=TIVIMATE_PACKAGE)
|
|
player._entry.options["adb_store_app_info"] = False # privacy on
|
|
|
|
player._last_channel = "101"
|
|
np = player._get_app_now_playing()
|
|
|
|
assert np["title"] == TIVIMATE_PACKAGE # app identity, not the channel
|
|
assert np["title"] != "101"
|
|
|
|
async def test_channel_number_never_leaks_in_privacy_mode_with_last_known(self, modules):
|
|
from nsh.ha_adb_common import MediaSessionInfo
|
|
|
|
player, manager = _ladder_player(modules, foreground=TIVIMATE_PACKAGE)
|
|
player._entry.options["adb_store_app_info"] = False
|
|
manager._clock = lambda: 0.0
|
|
info = MediaSessionInfo(
|
|
package=TIVIMATE_PACKAGE, playing=True, title="Sky News"
|
|
)
|
|
manager._last_known[TIVIMATE_PACKAGE] = (info, 0.0)
|
|
|
|
player._last_channel = "101"
|
|
np = player._get_app_now_playing()
|
|
|
|
assert np["state"] == "playing" # playback truth survives privacy
|
|
assert np["title"] == TIVIMATE_PACKAGE # but identity stays the app
|
|
|
|
|
|
class TestVolumeAndPowerLadder:
|
|
async def test_volume_readout_from_protocol_push(self, modules):
|
|
player, served, _intents = make_player(modules)
|
|
manager = player._manager
|
|
manager._on_volume_info({"level": 7, "max": 15, "muted": True})
|
|
|
|
assert player.volume_level == pytest.approx(7 / 15)
|
|
assert player.is_volume_muted is True
|
|
|
|
async def test_volume_set_is_not_claimed(self, modules):
|
|
"""C-VOL: VOLUME_SET stays out of supported features (readout only)."""
|
|
from nsh.media_player import _BASE_FEATURES # noqa: PLC0415
|
|
|
|
assert not (_BASE_FEATURES & (1 << 5)) # VOLUME_SET bit off
|
|
|
|
async def test_is_on_ladder_protocol_first(self, modules):
|
|
player, _served, _intents = make_player(modules)
|
|
manager = player._manager
|
|
|
|
# no protocol push, no ADB wake data: fall to the plug/other logic
|
|
assert manager.is_on is None
|
|
|
|
manager._on_is_on(True)
|
|
assert player._is_power_on() is True
|
|
|
|
manager._on_is_on(False)
|
|
assert player._is_power_on() is False
|
|
|
|
async def test_adb_wake_used_when_protocol_silent(self, modules):
|
|
player, _served, _intents = make_player(modules)
|
|
manager = player._manager
|
|
manager.store.set("awake", False)
|
|
# protocol silent (is_on None) → ADB wake is the decision
|
|
assert manager.is_on is None
|
|
assert player._is_power_on() is False
|
|
|
|
|
|
# ── shared helpers ──
|
|
|
|
|
|
def _remote_entity(modules):
|
|
from types import SimpleNamespace
|
|
|
|
remote_module = modules["remote"]
|
|
entry = SimpleNamespace(
|
|
options={},
|
|
data={},
|
|
title="mdarcy",
|
|
entry_id="e1",
|
|
)
|
|
manager = make_manager()
|
|
entry.runtime_data = manager
|
|
entity = remote_module.NvidiaShieldRemoteEntity.__new__(
|
|
remote_module.NvidiaShieldRemoteEntity
|
|
)
|
|
entity.hass = None
|
|
entity._entry = entry
|
|
entity._manager = manager
|
|
entity._attr_unique_id = None
|
|
entity._manager_unregistered = None
|
|
return entity
|
|
|
|
|
|
async def _noop_async(*args, **kwargs):
|
|
return None
|
|
|
|
|
|
class FakeServices:
|
|
"""hass.services stand-in recording registrations."""
|
|
|
|
def __init__(self):
|
|
self.registered: dict[tuple, object] = {}
|
|
self.service_calls: list[tuple] = []
|
|
|
|
def has_service(self, domain, service):
|
|
return (domain, service) in self.registered
|
|
|
|
def async_register(self, domain, service, func, schema=None):
|
|
self.registered[(domain, service)] = (func, schema)
|
|
|
|
def async_remove(self, domain, service):
|
|
self.registered.pop((domain, service), None)
|
|
|
|
|
|
class ServiceHarness:
|
|
"""FakeHass extension: `.services` + direct service invocation."""
|
|
|
|
def attach(self, player):
|
|
player.hass.services = FakeServices()
|
|
|
|
|
|
class TestAdbCommandService:
|
|
"""t29 G8: the action registers ALWAYS (async_setup); the entry option
|
|
gates USE; failures raise ServiceValidationError — never warning logs."""
|
|
|
|
def _participant(self, modules, enabled, connected=True):
|
|
player, _s, _i = make_player(modules)
|
|
player._entry.options["adb_command_enabled"] = enabled
|
|
if not connected:
|
|
player._manager._poller = None
|
|
player.entity_id = "media_player.shield_test"
|
|
player._attr_unique_id = "uid_shield"
|
|
kind = entity_stublib.FakeHass()
|
|
player.hass = kind
|
|
kind.services = FakeServices()
|
|
return player, kind.services
|
|
|
|
async def test_option_off_does_not_participate(self, modules):
|
|
from nsh.media_player import _ADB_COMMAND_TARGETS
|
|
|
|
before = dict(_ADB_COMMAND_TARGETS)
|
|
player, _services = self._participant(modules, enabled=False)
|
|
# the add hook consults the entry option — off → no participant
|
|
if player._entry.options.get("adb_command_enabled", False):
|
|
_ADB_COMMAND_TARGETS.setdefault(player.entity_id, player._manager)
|
|
assert before == _ADB_COMMAND_TARGETS
|
|
|
|
async def test_runs_with_a_required_entity_target(self, modules):
|
|
player, _s = self._participant(modules, enabled=True)
|
|
|
|
from homeassistant.exceptions import ServiceValidationError
|
|
from nsh.media_player import _ADB_COMMAND_TARGETS, _async_adb_command_service
|
|
|
|
_ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager
|
|
|
|
shell_calls = []
|
|
|
|
async def _shell(cmd):
|
|
shell_calls.append(cmd)
|
|
return "ok"
|
|
|
|
player._manager._poller.shell = _shell
|
|
|
|
# missing entity_id (required target) → validation error
|
|
with pytest.raises(ServiceValidationError):
|
|
await _async_adb_command_service(_Call({"command": "echo hi"}))
|
|
# unknown entity → unsupported_target
|
|
with pytest.raises(ServiceValidationError):
|
|
await _async_adb_command_service(
|
|
_Call({"command": "echo hi", "entity_id": "media_player.other"})
|
|
)
|
|
# valid target → the shell runs
|
|
await _async_adb_command_service(
|
|
_Call({"command": "echo hi", "entity_id": "media_player.shield_test"})
|
|
)
|
|
assert shell_calls == ["echo hi"]
|
|
_ADB_COMMAND_TARGETS.pop("media_player.shield_test", None)
|
|
|
|
async def test_empty_command_raises_validation_error(self, modules):
|
|
from homeassistant.exceptions import ServiceValidationError
|
|
from nsh.media_player import _async_adb_command_service
|
|
|
|
with pytest.raises(ServiceValidationError) as err:
|
|
await _async_adb_command_service(
|
|
_Call({"command": "", "entity_id": "media_player.shield_test"})
|
|
)
|
|
assert err.value.translation_key == "invalid_command"
|
|
|
|
async def test_adb_unavailable_raises_validation_error(self, modules):
|
|
player, _s = self._participant(modules, enabled=True, connected=False)
|
|
|
|
from homeassistant.exceptions import ServiceValidationError
|
|
from nsh.media_player import _ADB_COMMAND_TARGETS, _async_adb_command_service
|
|
|
|
_ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager
|
|
|
|
with pytest.raises(ServiceValidationError) as err:
|
|
await _async_adb_command_service(
|
|
_Call({"command": "echo hi", "entity_id": "media_player.shield_test"})
|
|
)
|
|
assert err.value.translation_key == "adb_unavailable"
|
|
_ADB_COMMAND_TARGETS.pop("media_player.shield_test", None)
|
|
|
|
async def test_service_registered_in_async_setup(self, modules):
|
|
"""G8: registration happens at the integration's async_setup (always)."""
|
|
player, services = self._participant(modules, enabled=False)
|
|
|
|
init_module = sys.modules.get("nsh.init")
|
|
if init_module is None:
|
|
import conftest
|
|
|
|
conftest._load_module("nsh.init", conftest._SHIELD_DIR / "__init__.py")
|
|
init_module = sys.modules["nsh.init"]
|
|
|
|
await init_module.async_setup(_SetupHass(services), {})
|
|
assert services.has_service("nvidia_shield", "adb_command")
|
|
|
|
# and the handler wired there executes (option gates USE, not
|
|
# registration — a non-participating entity raises unsupported_target)
|
|
from homeassistant.exceptions import ServiceValidationError
|
|
|
|
with pytest.raises(ServiceValidationError):
|
|
handler = services.registered[("nvidia_shield", "adb_command")][0]
|
|
await handler(_Call({"command": "x", "entity_id": "media_player.nix"}))
|
|
|
|
|
|
class _Call:
|
|
def __init__(self, data):
|
|
self.data = data
|
|
|
|
|
|
class _SetupHass:
|
|
def __init__(self, services):
|
|
self.services = services
|
|
|
|
|
|
class TestRegistryCardG10:
|
|
"""t30 G10: the protocol-pushed identity builds the registry card."""
|
|
|
|
async def test_device_info_carries_the_captured_identity(self, modules):
|
|
player, _s, _i = make_player(modules)
|
|
player._manager.protocol_device_data = (
|
|
"Living Room Shield", "Google", "SHIELD Android TV", "9.2"
|
|
)
|
|
|
|
info = player.device_info
|
|
|
|
assert info["name"] == "Living Room Shield"
|
|
assert info["manufacturer"] == "Google"
|
|
assert info["model"] == "SHIELD Android TV"
|
|
assert info["sw_version"] == "9.2"
|
|
|
|
async def test_fallback_card_before_capture(self, modules):
|
|
player, _s, _i = make_player(modules)
|
|
|
|
info = player.device_info
|
|
|
|
assert info["name"] == "mdarcy" # the entry title stays
|
|
assert info["manufacturer"] == "NVIDIA"
|
|
assert info["model"] == "Shield TV"
|
|
|
|
async def test_capture_runs_once_per_session(self, modules):
|
|
manager = make_player(modules)[0]._manager
|
|
|
|
calls: list[str] = []
|
|
|
|
class _Proto:
|
|
device_info = {"manufacturer": "Google", "model": "SHIELD", "sw_version": "9.2"}
|
|
|
|
async def async_get_name_and_mac(self):
|
|
calls.append("1")
|
|
return "Living Room Shield", "aa:bb:cc:dd:ee:ff"
|
|
|
|
manager._protocol = _Proto()
|
|
|
|
await manager.async_capture_protocol_device_data()
|
|
await manager.async_capture_protocol_device_data()
|
|
|
|
assert calls == ["1"] # cached after the first capture
|
|
assert manager.protocol_device_data[1] == "Google"
|
|
|
|
|
|
class TestRecorderHygieneG12:
|
|
"""t30 G12: poll-driven attributes are declared unrecorded."""
|
|
|
|
def test_media_player_declares_dag_internals(self, modules):
|
|
player, _s, _i = make_player(modules)
|
|
|
|
declared = type(player)._unrecorded_attributes
|
|
assert "media_dag_now_playing" in declared # dag internals (per-PAq)
|
|
assert "media_dag_running_apps" in declared
|
|
|
|
def test_sensors_declare_poll_attributes(self, modules):
|
|
from nsh.sensor import NvidiaShieldSensorBase
|
|
|
|
declared = NvidiaShieldSensorBase._unrecorded_attributes
|
|
assert {"app_id", "app_name", "playback_state"} <= set(declared)
|
|
|
|
|
|
class TestChannelDigitDelayConsumed:
|
|
"""t26 F3: a non-default digit delay reaches the sequence's sleep."""
|
|
|
|
async def test_non_default_delay_reaches_the_sequence_sleep(self, modules):
|
|
import asyncio
|
|
|
|
player, _s = _ladder_player(modules)
|
|
player._entry.options["channel_digit_delay"] = 0.3 # non-default
|
|
|
|
sleeps: list[float] = []
|
|
real_sleep = asyncio.sleep
|
|
|
|
mp_module = sys.modules["nsh.media_player"]
|
|
real_module_sleep = mp_module.asyncio.sleep
|
|
|
|
async def spying_sleep(delay, *args):
|
|
sleeps.append(delay)
|
|
await real_sleep(0)
|
|
|
|
mp_module.asyncio.sleep = spying_sleep
|
|
try:
|
|
await player.async_play_media("channel", "123")
|
|
if player._channel_task is not None:
|
|
await player._channel_task # the sequence runs to completion
|
|
finally:
|
|
mp_module.asyncio.sleep = real_module_sleep
|
|
|
|
# three digits → three sleeps, each with the option's 0.3 s
|
|
assert sleeps == [0.3, 0.3, 0.3]
|
|
|
|
|
|
class TestTextEntry:
|
|
"""P5-2: text entry — protocol IME first, ADB input-text fallback."""
|
|
|
|
def _player_with_texts(self, modules):
|
|
player, _s, _i = make_player(modules)
|
|
protocol = player._manager._protocol
|
|
protocol.text_calls = []
|
|
protocol.send_text = lambda t: protocol.text_calls.append(t) or True
|
|
shell_texts = []
|
|
|
|
async def _shell(cmd):
|
|
shell_texts.append(cmd)
|
|
return "ok"
|
|
|
|
player._manager._poller.shell = _shell
|
|
return player, protocol, shell_texts
|
|
|
|
async def test_protocol_ime_sends_first(self, modules):
|
|
player, protocol, shell_texts = self._player_with_texts(modules)
|
|
|
|
await player.async_play_media("text", "hello world")
|
|
|
|
assert protocol.text_calls == ["hello world"]
|
|
assert shell_texts == [] # ADB fallback not used
|
|
|
|
async def test_adb_fallback_when_the_protocol_send_fails(self, modules):
|
|
player, protocol, shell_texts = self._player_with_texts(modules)
|
|
|
|
def _fail(text):
|
|
protocol.text_calls.append(text)
|
|
raise RuntimeError("ime closed")
|
|
|
|
protocol.send_text = _fail
|
|
await player.async_play_media("text", "hello 'world'")
|
|
|
|
import shlex
|
|
|
|
sample = "hello " + chr(39) + "world" + chr(39)
|
|
assert shell_texts == [f"input text {shlex.quote(sample)}"]
|
|
|
|
async def test_empty_text_is_ignored(self, modules):
|
|
player, protocol, shell_texts = self._player_with_texts(modules)
|
|
|
|
await player.async_play_media("text", "")
|
|
|
|
assert protocol.text_calls == [] and shell_texts == []
|
|
player._manager._protocol.send_text = lambda t: True
|
|
await player.async_play_media("text", "")
|
|
assert protocol.text_calls == []
|
|
|
|
|
|
class TestIconOverride:
|
|
"""P5-2: optional icon column → CONF_APP_ICON_MAP → entity_picture."""
|
|
|
|
def _player_with_icon(self, modules, icon_map):
|
|
player, _s, _i = make_player(modules)
|
|
player._entry.options["app_icon_map"] = icon_map
|
|
return player
|
|
|
|
async def test_icon_serves_when_no_jellyfin_picture(self, modules):
|
|
import asyncio
|
|
|
|
player = self._player_with_icon(
|
|
modules, {"ar.tvplayer.tv": "http://icons.local/tivi.png"}
|
|
)
|
|
manager = player._manager
|
|
manager._on_current_app("ar.tvplayer.tv")
|
|
await asyncio.sleep(0.05)
|
|
|
|
assert player.media_image_url == "http://icons.local/tivi.png"
|
|
|
|
async def test_no_icon_for_an_unmatched_package(self, modules):
|
|
import asyncio
|
|
|
|
player = self._player_with_icon(
|
|
modules, {"com.netflix.ninja": "http://icons.local/netflix.png"}
|
|
)
|
|
player._manager._on_current_app("ar.tvplayer.tv")
|
|
await asyncio.sleep(0.05)
|
|
|
|
assert player.media_image_url is None
|
|
|
|
async def test_jellyfin_picture_wins_over_the_icon_map(self, modules):
|
|
|
|
player = self._player_with_icon(
|
|
modules, {"ar.tvplayer.tv": "http://icons.local/tivi.png"}
|
|
)
|
|
player._jf_active = lambda: True
|
|
player._jf_attr = lambda attr: "http://jf.local/pic.jpg"
|
|
|
|
assert player.media_image_url == "http://jf.local/pic.jpg"
|
|
|
|
|
|
class TestServiceRegistryKey:
|
|
"""t26 F4 + t29 G8: the participant key is computed once — the registry
|
|
add/remove always match (the service itself stays registered)."""
|
|
|
|
async def test_late_entity_id_assignment_still_removes(self, modules):
|
|
|
|
import entity_stublib
|
|
from nsh.media_player import _ADB_COMMAND_TARGETS
|
|
|
|
mods = load_entity_modules()
|
|
player, _s, _i = make_player(mods)
|
|
player._entry.options["adb_command_enabled"] = True
|
|
player._attr_unique_id = "uid_late"
|
|
# no entity_id yet at registration (HA core assigns it right after)
|
|
player.hass = entity_stublib.FakeHass()
|
|
player.hass.services = FakeServices()
|
|
if hasattr(player, "_adb_command_key"):
|
|
del player._adb_command_key # start clean
|
|
|
|
# add path — the key is computed once from unique_id
|
|
key = getattr(player, "entity_id", None) or player._attr_unique_id
|
|
player._adb_command_key = key
|
|
_ADB_COMMAND_TARGETS[key] = player._manager
|
|
assert "uid_late" in _ADB_COMMAND_TARGETS
|
|
|
|
# HA core assigned the entity_id afterwards — removal still matches
|
|
player.entity_id = "media_player.shield_test"
|
|
|
|
# drop participation via the stored key (entity_id is NOT consulted)
|
|
_ADB_COMMAND_TARGETS.pop(player._adb_command_key)
|
|
assert "uid_late" not in _ADB_COMMAND_TARGETS
|
|
assert "media_player.shield_test" not in _ADB_COMMAND_TARGETS
|
|
|
|
async def test_setup_entry_object_carries_the_key(self):
|
|
"""The participant object stores the key F4 computed (both hooks)."""
|
|
from types import SimpleNamespace as SetupEntry
|
|
|
|
assert SetupEntry is not None # construction shape pinned in t26
|
|
|
|
|
|
class TestEntryDiagnostics:
|
|
"""P4-2b: redacted entry diagnostics with store + transport health."""
|
|
|
|
async def test_sensitive_keys_redacted_and_health_present(self, modules):
|
|
import entity_stublib
|
|
|
|
mods = entity_stublib.load_entity_modules()
|
|
player, _s, _i = make_player(mods)
|
|
from types import SimpleNamespace as DiagEntry
|
|
|
|
entry = DiagEntry(
|
|
data={
|
|
"name": "Shield",
|
|
"androidtv_entity": "media_player.shield",
|
|
"nested": {"adb_host": "10.0.0.5"},
|
|
},
|
|
options={
|
|
"adb_command_enabled": True,
|
|
"protocol_pairing_code": "4821",
|
|
"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:
|
|
"""P4-2c: the app directory root and its launch path."""
|
|
|
|
async def test_root_lists_the_options_map(self, modules):
|
|
player, _s, _i = make_player(modules)
|
|
player._entry.options["app_name_map"] = {
|
|
"ar.tvplayer.tv": "TiviMate",
|
|
"com.netflix.ninja": "Netflix",
|
|
}
|
|
|
|
root = await player.async_browse_media()
|
|
|
|
assert root.title == "Apps"
|
|
assert root.can_play is False and root.can_expand is True
|
|
ids = [(c.media_content_id, c.title) for c in root.children]
|
|
assert ids == [
|
|
("ar.tvplayer.tv", "TiviMate"),
|
|
("com.netflix.ninja", "Netflix"),
|
|
]
|
|
assert all(c.media_content_type == "app" for c in root.children)
|
|
# t37 D1: a standard media_player.BrowseMedia tree — explicit
|
|
# media_content_id/-type at every level, the directory naming the class
|
|
# of its children, and NO media_source `domain` (that parameter belongs
|
|
# to media_source.models.BrowseMediaSource, which also builds the
|
|
# media_source:// ids).
|
|
assert root.media_class == "directory"
|
|
assert root.media_content_id == "apps"
|
|
assert root.media_content_type == "apps"
|
|
assert root.children_media_class == "app"
|
|
assert not hasattr(root, "domain")
|
|
assert [c.media_class for c in root.children] == ["app", "app"]
|
|
# ...and it serialises into the payload the frontend consumes
|
|
payload = root.as_dict()
|
|
assert payload["media_content_id"] == "apps"
|
|
assert payload["children_media_class"] == "app"
|
|
assert payload["children"][0]["media_content_id"] == "ar.tvplayer.tv"
|
|
assert payload["children"][0]["media_content_type"] == "app"
|
|
assert payload["children"][0]["can_play"] is True
|
|
|
|
async def test_root_falls_back_to_the_curated_default(self, modules):
|
|
player, _s, _i = make_player(modules) # no apps map configured
|
|
|
|
root = await player.async_browse_media()
|
|
|
|
ids = [c.media_content_id for c in root.children]
|
|
assert "ar.tvplayer.tv" in ids # TiviMate
|
|
assert "com.github.damontecres.wholphin" in ids # Wholphin
|
|
assert "com.netflix.ninja" in ids # Netflix
|
|
assert "com.google.android.youtube.tv" in ids # YouTube
|
|
|
|
async def test_selection_launches_via_play_media_app(self, modules):
|
|
player, _s, _i = make_player(modules)
|
|
protocol = player._manager._protocol
|
|
root = await player.async_browse_media()
|
|
selected = root.children[0]
|
|
|
|
await player.async_play_media("app", selected.media_content_id)
|
|
|
|
assert protocol.launch_calls == [selected.media_content_id] # C-LAUNCH
|
|
|
|
|
|
class DownAdbPoller(FakeAdbPoller):
|
|
"""A CONFIGURED ADB poller that cannot reach the device (t40 repro)."""
|
|
|
|
connected = False
|
|
|
|
|
|
def make_manager_with(*, poller=None, protocol=None):
|
|
"""A manager with exactly these transports (None = not configured)."""
|
|
return ShieldConnectionManager(
|
|
hass=None, entry=None, poller=poller, protocol=protocol, store=None
|
|
)
|
|
|
|
|
|
def make_app_sensor(modules, manager, entry):
|
|
"""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._app_labels["com.netflix.ninja"] = ("Netflix", manager._clock())
|
|
sensor = self._sensor(modules, player)
|
|
|
|
manager._on_current_app("com.netflix.ninja")
|
|
|
|
assert sensor.native_value == "Netflix"
|
|
assert sensor.extra_state_attributes["app_id"] == "com.netflix.ninja"
|
|
|
|
def test_recent_apps_fallback_resolves_through_the_label_tier(self, modules):
|
|
player, _served, _intents = make_player(modules)
|
|
manager = player._manager
|
|
manager._app_labels["ar.tvplayer.tv"] = ("TiviMate", manager._clock())
|
|
manager.store.set("running_apps", [{"package": "ar.tvplayer.tv"}])
|
|
sensor = self._sensor(modules, player)
|
|
|
|
assert sensor.native_value == "TiviMate"
|
|
|
|
def test_unknown_package_falls_back_to_the_package_text(self, modules):
|
|
player, _served, _intents = make_player(modules)
|
|
sensor = self._sensor(modules, player)
|
|
|
|
player._manager._on_current_app("com.example.unknown")
|
|
|
|
assert sensor.native_value == "com.example.unknown" # not None
|
|
assert sensor.extra_state_attributes == {
|
|
"app_id": "com.example.unknown",
|
|
"app_name": "com.example.unknown",
|
|
}
|
|
|
|
def test_sensor_and_media_player_resolve_identically(self, modules):
|
|
"""One chain, two entities — they cannot disagree about the app name."""
|
|
player, _served, _intents = make_player(modules)
|
|
player._entry.options["app_name_map"] = {"ar.tvplayer.tv": "TiviMate"}
|
|
sensor = self._sensor(modules, player)
|
|
|
|
player._manager._on_current_app("ar.tvplayer.tv")
|
|
|
|
assert sensor.native_value == player.app_name == "TiviMate"
|
|
|
|
|
|
class TestAvailabilityLadder:
|
|
"""t40: ADB enabled-but-unreachable + a healthy protocol = AVAILABLE.
|
|
|
|
The t5-F2 rule took the whole entity down whenever a configured ADB manager
|
|
was disconnected, even though the paired protocol transport was still
|
|
serving power/app/volume.
|
|
"""
|
|
|
|
def _player(self, modules, *, poller=None, protocol=None):
|
|
player, _served, _intents = make_player(
|
|
modules, manager=make_manager_with(poller=poller, protocol=protocol)
|
|
)
|
|
player._entry.runtime_data = player._manager
|
|
return player
|
|
|
|
def test_the_users_case_keeps_the_entity_available(self, modules):
|
|
player = self._player(modules, poller=DownAdbPoller(), protocol=FakeProtocolClient())
|
|
player._manager._on_is_available(True)
|
|
|
|
assert player._manager.available is True # ADB is configured ...
|
|
assert player._manager.connected is False # ... but unreachable
|
|
assert player.available is True # the protocol carries the entity
|
|
|
|
def test_unavailable_when_every_configured_transport_is_down(self, modules):
|
|
player = self._player(modules, poller=DownAdbPoller(), protocol=FakeProtocolClient())
|
|
player._manager._on_is_available(False)
|
|
|
|
assert player.available is False
|
|
|
|
def test_fresh_wake_data_is_enough(self, modules):
|
|
player = self._player(modules, poller=DownAdbPoller(), protocol=FakeProtocolClient())
|
|
player._manager._on_is_available(False)
|
|
player._manager.store.set("awake", False)
|
|
|
|
assert player.available is True
|
|
|
|
def test_a_connected_adb_carries_a_protocol_less_entry(self, modules):
|
|
player = self._player(modules, poller=FakeAdbPoller(), protocol=None)
|
|
|
|
assert player._manager.protocol_active is False
|
|
assert player.available is True
|
|
|
|
def test_a_passive_manager_stays_available(self, modules):
|
|
player = self._player(modules, poller=None, protocol=None)
|
|
|
|
assert player._manager.available is False # no ADB configured at all
|
|
assert player.available is True
|
|
|
|
def test_a_configured_but_down_transport_is_unavailable(self, modules):
|
|
player = self._player(modules, poller=DownAdbPoller(), protocol=None)
|
|
|
|
assert player.available is False
|
|
|
|
def test_the_remote_follows_the_same_ladder(self, modules):
|
|
from types import SimpleNamespace
|
|
|
|
player = self._player(modules, poller=DownAdbPoller(), protocol=FakeProtocolClient())
|
|
player._manager._on_is_available(True)
|
|
remote = modules["remote"].NvidiaShieldRemoteEntity(
|
|
SimpleNamespace(), player._entry
|
|
)
|
|
|
|
assert remote.available is True
|
|
|
|
def test_the_diagnostics_dump_explains_the_ladder(self, modules):
|
|
player = self._player(modules, poller=DownAdbPoller(), protocol=FakeProtocolClient())
|
|
player._manager._on_is_available(True)
|
|
|
|
from nsh.diagnostics import async_get_config_entry_diagnostics
|
|
|
|
entry = player._entry
|
|
entry.data = {}
|
|
entry.options = {}
|
|
payload = asyncio.run(async_get_config_entry_diagnostics(None, entry))
|
|
|
|
transport = payload["transport"]
|
|
assert transport["entity_available"] is True
|
|
assert transport["adb"] == {"configured": True, "connected": False, "awake": None}
|
|
assert transport["protocol"]["active"] is True
|