Files
ha-nvidia-shield/tests/test_media_sources.py
T
James Mackay 0c7e405108 fix(connection): a failed app-label fetch is retried, not cached as a miss (t55)
The Current App sensor and media_player.app_name showed a package id in
the live run. Root cause was in the manifest-label tier: `_fetch_app_label`
returned early (no poller, ADB down) or logged and returned on a shell
error WITHOUT touching the cache, but `app_label` had already put the
package in `_app_label_requests`; the drain then threw the request away,
so a render between the failure and the next cycle re-enqueued — while
nothing distinguished "the device answered, this package has no label"
from "we never got an answer". Nothing was visibly retried, and the label
tier could stay silent indefinitely.

The cache is now (label, stamp, resolved) and records the FETCH, not the
wishful result:

  * a fetch that RAN caches its outcome for APP_LABEL_TTL — the label, or
    the empty string for a genuinely label-less package. Definitive: no
    re-enqueue, so the t24 F1 per-heartbeat round-trip tax stays closed.
  * a fetch that could NOT run stamps nothing, logs at debug (so a
    package id in the UI is diagnosable), and the next render retries.
  * a queued or in-flight fetch is not enqueued twice, so a burst of
    renders still costs one `pm dump` per watchdog cycle.

`cache_app_label()` is the one public writer of the cache (tests and any
future out-of-band reader no longer poke the private tuple), and
`app_label_requests`/`app_label_inflight` expose the queue for tests.

The old 2-tuple seeds in the label tests move to `cache_app_label`.
2026-10-12 06:53:09 +01:00

472 lines
19 KiB
Python

"""tests/test_media_sources.py — source_list, source and native select_source (t47).
The last piece a universal-media-player replacement needs from this integration:
a dropdown of the apps (``source_list``), an answer to "what is on screen"
(``source``), and a way to pick one (``media_player.select_source``). All three read
the SAME app directory — the user's Apps map, else the curated default — so the
dropdown cannot offer a name the launch path cannot resolve.
The t44 rule still holds: a configured ``select_source`` script takes precedence,
because selecting a source can be more than launching an app. Without one, the name
resolves back to its package and goes out over the shared ``play_media("app", …)``
path; an unknown name is a translated error, never a silent no-op.
"""
import entity_stublib
import pytest
from nsh.const import CONF_APP_NAME_MAP, CONF_COMMAND_SCRIPTS, SCRIPT_VAR_APP_NAME
from test_command_routing import make_player
TIVIMATE = "ar.tvplayer.tv"
SCRIPT = "script.shield_source"
@pytest.fixture(name="modules", autouse=True)
def scoped_modules():
"""Stub scope + loaded entity modules for one test (t7 discipline)."""
snapshot = entity_stublib.snapshot_modules()
entity_stublib.install()
try:
yield entity_stublib.load_entity_modules()
finally:
entity_stublib.restore_modules(snapshot)
def _player(modules, *, app_map=None, scripts=None, foreground=None):
player, served, _intents = make_player(modules)
if app_map is not None:
player._entry.options[CONF_APP_NAME_MAP] = app_map
if scripts is not None:
player._entry.options[CONF_COMMAND_SCRIPTS] = scripts
if foreground is not None:
player._manager._on_current_app(foreground)
return player, served
class TestSourceList:
"""The dropdown: friendly names, in a stable order, from the one directory."""
def test_it_comes_from_the_apps_map(self, modules):
player, _served = _player(
modules,
app_map={"com.custom.one": "One", "com.custom.two": "Two"},
)
assert player.source_list == ["One", "Two"] # map order, not shuffled
def test_it_falls_back_to_the_curated_directory(self, modules):
player, _served = _player(modules) # no map configured
assert player.source_list == ["TiviMate", "Wholphin", "Netflix", "YouTube"]
def test_a_blank_name_never_reaches_the_dropdown(self, modules):
"""t49: the map is hand-editable, so 'com.x=' must not offer an empty entry.
An empty entry is worst than useless: it is a dropdown item that
select_source has to reject as an unknown source.
"""
player, _served = _player(
modules,
app_map={
"com.blank": "",
TIVIMATE: "TiviMate",
"com.spaces": " ",
"com.netflix.ninja": "Netflix",
},
)
assert player.source_list == ["TiviMate", "Netflix"]
assert "" not in player.source_list
async def test_the_list_matches_the_browse_titles_exactly(self, modules):
"""Same accessor, same order, same bytes — a blank entry excepted (t49)."""
for app_map in (
None, # the curated default
{"ar.tvplayer.tv": "TiviMate", "com.netflix.ninja": "Netflix"},
):
player, _served = _player(modules, app_map=app_map)
root = await player.async_browse_media()
assert player.source_list == [child.title for child in root.children]
async def test_the_survivors_still_resolve_and_launch(self, modules):
"""The filter must not cost a resolvable name (t49)."""
player, _served = _player(
modules,
app_map={"com.blank": "", TIVIMATE: "TiviMate", "com.blank2": " "},
)
for name in player.source_list:
assert player._source_package(name) is not None
await player.async_select_source("TiviMate")
assert player._manager._protocol.launch_calls == [TIVIMATE]
# the blank entries are not selectable, and never launch
assert player._source_package("") is None
assert player._source_package(" ") is None
def test_every_offered_name_resolves_to_a_package(self, modules):
"""The invariant the shared directory buys: no name without a destination.
t51 extends it to the hand-edited shapes: blanks, padded names and two
entries normalizing to one name all still leave only resolvable offers.
"""
for app_map in (
None,
{"com.custom.one": "One"},
{
"com.blank": "",
"com.blank.package": " ",
TIVIMATE: " TiviMate ",
"com.a": "Netflix",
"com.b": " Netflix ",
"": "Ghost",
},
):
player, _served = _player(modules, app_map=app_map)
for name in player.source_list:
assert player._source_package(name) is not None, name
class TestSource:
"""`source` is the foreground app's resolved name — the app_name chain."""
def test_it_resolves_through_the_apps_map(self, modules):
player, _served = _player(
modules, app_map={TIVIMATE: "TiviMate"}, foreground=TIVIMATE
)
assert player.source == "TiviMate"
assert player.source == player.app_name # one truth, not two
def test_it_uses_the_cached_device_label_when_unmapped(self, modules):
player, _served = _player(modules, foreground="com.netflix.ninja")
player._manager.cache_app_label("com.netflix.ninja", "Netflix")
assert player.source == "Netflix"
def test_it_falls_back_to_the_raw_package(self, modules):
player, _served = _player(modules, foreground="com.unknown.app")
assert player.source == "com.unknown.app"
def test_it_is_unknown_when_nothing_has_reported(self, modules):
player, _served = _player(modules)
assert player.source is None
class TestSelectSource:
"""Picking a source: script first, else the native launch, else an error."""
async def test_a_name_launches_its_package(self, modules):
player, served = _player(
modules, app_map={TIVIMATE: "TiviMate", "com.netflix.ninja": "Netflix"}
)
await player.async_select_source("TiviMate")
assert player._manager._protocol.launch_calls == [TIVIMATE]
assert served == [] # no script was configured, and nothing else left
async def test_the_curated_names_launch_too(self, modules):
player, _served = _player(modules)
await player.async_select_source("Netflix")
assert player._manager._protocol.launch_calls == ["com.netflix.ninja"]
async def test_a_raw_package_and_any_case_are_accepted(self, modules):
player, _served = _player(
modules, app_map={TIVIMATE: "TiviMate", "com.netflix.ninja": "Netflix"}
)
await player.async_select_source(TIVIMATE)
await player.async_select_source("tivimate")
await player.async_select_source(" NETFLIX ")
assert player._manager._protocol.launch_calls == [
TIVIMATE,
TIVIMATE,
"com.netflix.ninja",
]
async def test_a_configured_script_still_wins(self, modules):
player, served = _player(
modules,
app_map={TIVIMATE: "TiviMate"},
scripts={"select_source": SCRIPT},
)
await player.async_select_source("TiviMate")
assert served == [
(
"script",
"turn_on",
{"entity_id": SCRIPT, "variables": {SCRIPT_VAR_APP_NAME: "TiviMate"}},
)
]
assert player._manager._protocol.launch_calls == [] # native path untouched
async def test_an_unknown_name_raises_a_translated_error(self, modules):
from homeassistant.exceptions import ServiceValidationError
player, _served = _player(modules, app_map={TIVIMATE: "TiviMate"})
with pytest.raises(ServiceValidationError) as raised:
await player.async_select_source("Not An App")
assert raised.value.translation_domain == "nvidia_shield"
assert raised.value.translation_key == "unknown_source"
assert raised.value.translation_placeholders == {"source": "Not An App"}
assert player._manager._protocol.launch_calls == [] # nothing was sent
async def test_an_empty_name_is_an_error_too(self, modules):
from homeassistant.exceptions import ServiceValidationError
player, _served = _player(modules)
with pytest.raises(ServiceValidationError):
await player.async_select_source(" ")
class TestAdvertisedFeatures:
"""HA routes a service by its feature flag, so the flag has to be honest."""
def test_a_source_list_advertises_select_source(self, modules):
from homeassistant.components.media_player import MediaPlayerEntityFeature
player, _served = _player(modules)
assert player.supported_features & MediaPlayerEntityFeature.SELECT_SOURCE
def test_a_script_advertises_it_even_with_no_directory(self, modules, monkeypatch):
from homeassistant.components.media_player import MediaPlayerEntityFeature
player, _served = _player(modules, scripts={"select_source": SCRIPT})
monkeypatch.setattr(type(player), "_app_directory", property(lambda self: {}))
assert player.supported_features & MediaPlayerEntityFeature.SELECT_SOURCE
def test_neither_means_it_is_not_offered(self, modules, monkeypatch):
from homeassistant.components.media_player import MediaPlayerEntityFeature
player, _served = _player(modules)
monkeypatch.setattr(type(player), "_app_directory", property(lambda self: {}))
assert not player.supported_features & MediaPlayerEntityFeature.SELECT_SOURCE
def test_select_source_is_not_gated_on_the_volume_set_script(self, modules):
"""The two script-only commands stay independent."""
from homeassistant.components.media_player import MediaPlayerEntityFeature
player, _served = _player(modules, scripts={"volume_set": "script.volume"})
assert player.supported_features & MediaPlayerEntityFeature.VOLUME_SET
assert player.supported_features & MediaPlayerEntityFeature.SELECT_SOURCE
class TestNormalizedDirectories:
"""t51: a hand-edited map is normalized ONCE, in the shared accessor.
The two residuals t49 left share one cause — `_app_directory` handed the raw map
out, so the Browse tree showed a child with a blank title and the reverse lookup
compared the user's padded name against itself. The policy now lives in that one
accessor (strip names and keys, drop blanks, first-in-map-order wins a shared
name) and is pinned here, together with the promise that a well-formed map is
untouched.
"""
BLANKED = {
"com.blank": "",
TIVIMATE: "TiviMate",
"com.spaces": " ",
"": "Ghost",
}
async def test_a_blank_entry_never_reaches_the_browse_tree(self, modules):
"""The Browse tree and the dropdown are the same, resolved offer."""
player, _served = _player(modules, app_map=dict(self.BLANKED))
root = await player.async_browse_media()
assert [child.title for child in root.children] == ["TiviMate"]
assert [child.media_content_id for child in root.children] == [TIVIMATE]
assert player.source_list == ["TiviMate"]
async def test_a_blank_package_is_dropped_too(self, modules):
"""An entry with no package is a child that cannot be launched."""
player, _served = _player(
modules, app_map={"": "Ghost", " ": "Ghost 2", TIVIMATE: "TiviMate"}
)
root = await player.async_browse_media()
assert [child.media_content_id for child in root.children] == [TIVIMATE]
assert player.source_list == ["TiviMate"]
async def test_a_padded_name_is_stripped_once_and_then_resolves(self, modules):
player, _served = _player(
modules,
app_map={TIVIMATE: " TiviMate ", "com.netflix.ninja": "Netflix"},
)
root = await player.async_browse_media()
# display and match key are one string, everywhere
assert player.source_list == ["TiviMate", "Netflix"]
assert [child.title for child in root.children] == ["TiviMate", "Netflix"]
# ...so every spelling the user may type finds the package
assert player._source_package("TiviMate") == TIVIMATE
assert player._source_package(" TiviMate ") == TIVIMATE
assert player._source_package("tivimate") == TIVIMATE
await player.async_select_source("TiviMate")
assert player._manager._protocol.launch_calls == [TIVIMATE]
async def test_two_packages_with_one_name_keep_a_deterministic_winner(self, modules):
"""Both apps stay reachable; the NAME goes to the first in map order."""
player, _served = _player(
modules, app_map={"com.a": "Netflix", "com.b": " Netflix "}
)
root = await player.async_browse_media()
# both are still offered in the tree, each by its own package id
assert [(c.media_content_id, c.title) for c in root.children] == [
("com.a", "Netflix"),
("com.b", "Netflix"),
]
# the dropdown offers the name once, and resolves to the first entry
assert player.source_list == ["Netflix"]
assert player._source_package("Netflix") == "com.a"
await player.async_select_source("Netflix")
assert player._manager._protocol.launch_calls == ["com.a"]
# map order decides — the user's order, not chance
swapped, _served_swapped = _player(
modules, app_map={"com.b": "Netflix", "com.a": "Netflix"}
)
assert swapped._source_package("Netflix") == "com.b"
async def test_a_wellformed_map_passes_through_unchanged(self, modules):
"""Stripping touches the EDGES: an inner space is part of the name."""
player, _served = _player(
modules, app_map={TIVIMATE: "TiviMate Pro", "com.netflix.ninja": "Netflix"}
)
root = await player.async_browse_media()
assert player.source_list == ["TiviMate Pro", "Netflix"]
assert [child.title for child in root.children] == ["TiviMate Pro", "Netflix"]
assert player._source_package("TiviMate Pro") == TIVIMATE
class TestCaseInsensitiveNames:
"""t53: one row per NAME, compared case-insensitively — as the lookup does.
The dropdown and the reverse lookup disagreed about case: two entries called
`Dup` and `dup` survived an exact-match dedupe as two rows, while
`_source_package` — case-insensitive by design (t47) — resolved BOTH to the
first entry. So the second row launched the first entry's app: a row that lies
about what it does, and the one shape here the editor can produce by itself.
"""
async def test_names_differing_only_by_case_are_offered_once(self, modules):
player, _served = _player(modules, app_map={"com.a": "Dup", "com.b": "dup"})
# one row, keeping the first spelling the user typed
assert player.source_list == ["Dup"]
# ...and the Browse tree still holds both apps, each under its own id
root = await player.async_browse_media()
assert [(c.media_content_id, c.title) for c in root.children] == [
("com.a", "Dup"),
("com.b", "dup"),
]
async def test_the_offered_row_launches_the_app_it_shows(self, modules):
player, _served = _player(modules, app_map={"com.a": "Dup", "com.b": "dup"})
await player.async_select_source(player.source_list[0])
assert player._manager._protocol.launch_calls == ["com.a"] # the row it shows
async def test_the_shadowed_app_stays_reachable_by_id(self, modules):
player, _served = _player(modules, app_map={"com.a": "Dup", "com.b": "dup"})
assert player._source_package("com.b") == "com.b"
await player.async_select_source("com.b")
assert player._manager._protocol.launch_calls == ["com.b"]
# the shadowed NAME still resolves to the one the list offers (first wins)
assert player._source_package("dup") == "com.a"
async def test_the_first_spelling_wins_in_map_order(self, modules):
player, _served = _player(modules, app_map={"com.b": "dup", "com.a": "Dup"})
assert player.source_list == ["dup"]
await player.async_select_source("dup")
assert player._manager._protocol.launch_calls == ["com.b"]
async def test_a_mixed_map_offers_each_name_once(self, modules):
player, _served = _player(
modules,
app_map={
"com.a": "Dup",
"com.b": "dup",
"com.netflix.ninja": "Netflix",
"com.google.android.youtube.tv": "netflix",
TIVIMATE: "TiviMate Pro",
},
)
assert player.source_list == ["Dup", "Netflix", "TiviMate Pro"]
assert player._source_package("Netflix") == "com.netflix.ninja"
def test_the_offered_list_never_repeats_a_name(self, modules):
"""The invariant behind the fix, over every map shape in this file."""
for app_map in (
None,
{"com.a": "Dup", "com.b": "dup"},
{"com.a": "Dup", "com.b": "DUP", "com.c": "dup"},
{"com.blank": "", TIVIMATE: " TiviMate ", "com.a": "Netflix", "com.b": "netflix"},
):
player, _served = _player(modules, app_map=app_map)
folded = [name.casefold() for name in player.source_list]
assert len(folded) == len(set(folded)), app_map
assert len(folded) == len(player.source_list)
def test_a_padded_map_value_is_displayed_verbatim_by_app_name(self, modules):
"""The README's boundary, pinned: the list cleans it, app_name does not.
The name map is the user's own override, so it is used as typed — which is
why `source` can be a string that is not in source_list.
"""
player, _served = _player(
modules, app_map={TIVIMATE: " TiviMate "}, foreground=TIVIMATE
)
assert player.source_list == ["TiviMate"] # the dropdown cleans it
assert player.source == " TiviMate " # this surface shows it as typed
assert player.source not in player.source_list
def test_a_blank_map_value_falls_through_to_the_raw_package(self, modules):
"""Blanks are not displayed as nothing — the chain continues past them."""
player, _served = _player(modules, app_map={TIVIMATE: ""}, foreground=TIVIMATE)
assert player.source == TIVIMATE