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`.
472 lines
19 KiB
Python
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
|