James Mackay
364d4b7da3
The live complaint in one assertion: every app the user has, plus the curated four, is offered — TiviMate, Wholphin, Netflix, YouTube, Kodi, Plex, Spotify and an app only this device knows, with the launcher added and no system app invented that `pm list packages -3` never reported.
524 lines
20 KiB
Python
524 lines
20 KiB
Python
"""t57: installed-app discovery — parsers, the bounded sweep, and the surfaces.
|
|
|
|
No real Shield was available, so the command OUTPUT here is the part that is
|
|
fixture-based, not observed: the shapes are the documented Android ones
|
|
(``pm list packages`` prints ``package:<id>``; ``resolve-activity --brief``
|
|
prints the verbose intent then ``package/activity``; ``pm dump`` prints
|
|
``application-label: '...'``). The parsers are deliberately tolerant because
|
|
that is the property that matters when a real device's output differs in
|
|
whitespace or ordering, and every tolerance is pinned below.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import entity_stublib # noqa: E402
|
|
import pytest # noqa: E402
|
|
from entity_stublib import load_entity_modules # noqa: E402
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def scoped_entity_stubs():
|
|
"""Container for the HA stubs + loaded entity modules (t7 discipline).
|
|
|
|
Without the teardown the stubs this file installs leak into whichever test
|
|
file runs next — tests/test_harness_scoping.py exists because that mistake
|
|
is invisible until an unrelated file fails.
|
|
"""
|
|
snapshot = entity_stublib.snapshot_modules()
|
|
entity_stublib.install()
|
|
yield load_entity_modules()
|
|
entity_stublib.restore_modules(snapshot)
|
|
|
|
from nsh.connection import ShieldConnectionManager # noqa: E402
|
|
from nsh.const import ( # noqa: E402
|
|
APP_ICON_FALLBACKS,
|
|
APP_LIST_TTL,
|
|
CONF_APP_NAME_MAP,
|
|
)
|
|
from nsh.ha_adb_common import ( # noqa: E402
|
|
icon_for_package,
|
|
installed_packages,
|
|
is_launchable,
|
|
parse_active_launcher,
|
|
parse_package_list,
|
|
)
|
|
|
|
# ── realistic device output ────────────────────────────────────────────────
|
|
|
|
#: `pm list packages -3` on a Shield with a normal app set. Wrapped the way a
|
|
#: real adb shell answers (a bare list, no header).
|
|
USER_PACKAGES = (
|
|
"package:ar.tvplayer.tv\n"
|
|
"package:com.github.damontecres.wholphin\n"
|
|
"package:com.netflix.ninja\n"
|
|
"package:com.google.android.youtube.tv\n"
|
|
"package:org.xbmc.kodi\n"
|
|
"package:com.plexapp.android\n"
|
|
"package:com.spotify.tv.android\n"
|
|
"package:com.example.handmade\n"
|
|
)
|
|
|
|
#: `cmd package resolve-activity --brief -c android.intent.category.HOME` —
|
|
#: the verbose header then the component, exactly as Android prints it.
|
|
LAUNCHER_RESOLVED = (
|
|
"priority=0 preferredOrder=0 match=0x108000 specificIndex=-1 isDefault=false\n"
|
|
"com.google.android.tvlauncher/com.google.android.apps.tv.launcherx.MainActivity\n"
|
|
)
|
|
|
|
#: `pm dump` for an app whose manifest and locale variants both exist. The
|
|
#: parser takes the first default-locale label and ignores the localized ones.
|
|
DUMP_TIVIMATE = (
|
|
"Package [ar.tvplayer.tv] (7f1a2b3c):\n"
|
|
" application-label:'TiviMate'\n"
|
|
" application-label-de:'TiviMate DE'\n"
|
|
" application-label-fr:'TiviMate FR'\n"
|
|
" versionName=4.7.0\n"
|
|
)
|
|
|
|
#: A dump that RAN and carries no label at all — several system/OEM packages
|
|
#: look like this. It is a definitive negative, not a failed fetch.
|
|
DUMP_LABEL_LESS = (
|
|
"Package [com.example.handmade] (aa11bb22):\n"
|
|
" versionName=1.0\n"
|
|
" codePath=/data/app/com.example.handmade\n"
|
|
)
|
|
|
|
|
|
class _DiscoveryPoller:
|
|
"""A poller whose shell answers the discovery commands from fixtures."""
|
|
|
|
def __init__(self, *, packages=USER_PACKAGES, launcher=LAUNCHER_RESOLVED,
|
|
dumps=None, failure=False):
|
|
self.connected = True
|
|
self.connected_override = True
|
|
self.packages = packages
|
|
self.launcher = launcher
|
|
self.dumps = dict(dumps or {})
|
|
self.failure = failure
|
|
self.shell_calls: list[str] = []
|
|
|
|
@property
|
|
def is_connected(self):
|
|
return self.connected
|
|
|
|
async def shell(self, cmd: str):
|
|
self.shell_calls.append(cmd)
|
|
if self.failure:
|
|
return None
|
|
if "pm list packages" in cmd:
|
|
return self.packages
|
|
if "resolve-activity" in cmd:
|
|
return self.launcher
|
|
if "pm dump" in cmd:
|
|
import shlex
|
|
|
|
for package, dump in self.dumps.items():
|
|
if shlex.quote(package) in cmd or package in cmd:
|
|
return dump
|
|
return "__NSH_EXIT:0\n" # ran, no label
|
|
return None
|
|
|
|
|
|
def _manager(poller):
|
|
return ShieldConnectionManager(hass=None, entry=None, poller=poller)
|
|
|
|
|
|
class TestPackageListParsing:
|
|
"""`pm list packages -3`."""
|
|
|
|
def test_real_output_parses_one_package_per_line(self):
|
|
assert parse_package_list(USER_PACKAGES) == [
|
|
"ar.tvplayer.tv",
|
|
"com.github.damontecres.wholphin",
|
|
"com.netflix.ninja",
|
|
"com.google.android.youtube.tv",
|
|
"org.xbmc.kodi",
|
|
"com.plexapp.android",
|
|
"com.spotify.tv.android",
|
|
"com.example.handmade",
|
|
]
|
|
|
|
def test_noise_and_duplicates_are_tolerated(self):
|
|
noisy = (
|
|
"adb: connecting to 10.0.0.9:5555\n"
|
|
"package:com.a\n"
|
|
"\n"
|
|
"package:com.a\n"
|
|
"package:com.b \n"
|
|
"some warning line\n"
|
|
)
|
|
assert parse_package_list(noisy) == ["com.a", "com.b"]
|
|
|
|
def test_empty_and_none_are_an_empty_list(self):
|
|
assert parse_package_list(None) == []
|
|
assert parse_package_list("") == []
|
|
|
|
def test_a_package_id_is_sanity_checked(self):
|
|
assert is_launchable("com.example.app")
|
|
assert is_launchable("ar.tvplayer.tv")
|
|
assert not is_launchable("")
|
|
assert not is_launchable("notapackage")
|
|
assert not is_launchable("com..b")
|
|
|
|
|
|
class TestLauncherParsing:
|
|
"""`cmd package resolve-activity --brief -c ...HOME`."""
|
|
|
|
def test_the_component_after_the_verbose_header_is_the_launcher(self):
|
|
assert parse_active_launcher(LAUNCHER_RESOLVED) == "com.google.android.tvlauncher"
|
|
|
|
def test_activity_only_output_still_parses(self):
|
|
assert parse_active_launcher("com.google.android.tvlauncher/.MainActivity") == (
|
|
"com.google.android.tvlauncher"
|
|
)
|
|
|
|
def test_nothing_resolved_is_none(self):
|
|
assert parse_active_launcher("No activity found") is None
|
|
assert parse_active_launcher(None) is None
|
|
|
|
def test_the_tv_settings_entry_point_is_not_a_launcher(self):
|
|
# a HOME resolution can land on the settings app's entry point; that is
|
|
# not the app a user opens
|
|
output = "com.android.tv.settings/.MainActivity"
|
|
assert parse_active_launcher(output) is None
|
|
|
|
|
|
class TestInstalledPackagesMerge:
|
|
"""User apps plus the active launcher, de-duplicated."""
|
|
|
|
def test_launcher_is_added_last(self):
|
|
packages = installed_packages(USER_PACKAGES, LAUNCHER_RESOLVED)
|
|
assert packages[-1] == "com.google.android.tvlauncher"
|
|
assert packages[0] == "ar.tvplayer.tv"
|
|
|
|
def test_a_third_party_launcher_is_not_duplicated(self):
|
|
user = "package:com.example.launcher\npackage:com.a\n"
|
|
launcher = "com.example.launcher/.MainActivity"
|
|
assert installed_packages(user, launcher) == ["com.example.launcher", "com.a"]
|
|
|
|
def test_no_launcher_still_yields_the_user_apps(self):
|
|
assert installed_packages(USER_PACKAGES, None) == parse_package_list(USER_PACKAGES)
|
|
|
|
|
|
class TestIconHeuristic:
|
|
"""Icons are a curated map plus a generic fallback — and always a string."""
|
|
|
|
def test_a_known_package_gets_its_curated_icon(self):
|
|
assert icon_for_package("com.netflix.ninja", APP_ICON_FALLBACKS) == "mdi:netflix"
|
|
|
|
def test_an_unknown_package_gets_the_generic_icon(self):
|
|
assert icon_for_package("com.example.handmade", APP_ICON_FALLBACKS) == "mdi:television"
|
|
|
|
def test_the_result_is_always_a_string(self):
|
|
for package in ("", None, "com.example.handmade", "ar.tvplayer.tv"):
|
|
icon = icon_for_package(package, APP_ICON_FALLBACKS)
|
|
assert isinstance(icon, str) and icon
|
|
|
|
|
|
class TestDiscoverySweep:
|
|
"""The bounded sweep on the connection manager."""
|
|
|
|
async def test_a_sweep_lists_labels_and_icons_every_app(self):
|
|
poller = _DiscoveryPoller(dumps={
|
|
"ar.tvplayer.tv": DUMP_TIVIMATE,
|
|
"com.example.handmade": DUMP_LABEL_LESS,
|
|
})
|
|
manager = _manager(poller)
|
|
|
|
rows = await manager.async_refresh_installed_apps()
|
|
|
|
packages = [row["package"] for row in rows]
|
|
assert "ar.tvplayer.tv" in packages
|
|
assert "com.netflix.ninja" in packages
|
|
assert "com.google.android.tvlauncher" in packages # the launcher is there
|
|
by_pkg = {row["package"]: row for row in rows}
|
|
assert by_pkg["ar.tvplayer.tv"]["name"] == "TiviMate" # from the dump
|
|
assert by_pkg["ar.tvplayer.tv"]["icon"] == "mdi:television-classic"
|
|
# a label-less app keeps its discovered name (the package id) and is
|
|
# still a row — it must not vanish from the table
|
|
assert by_pkg["com.example.handmade"]["name"] == "com.example.handmade"
|
|
# and every row's name/icon is a string
|
|
for row in rows:
|
|
assert isinstance(row["name"], str) and row["name"]
|
|
assert isinstance(row["icon"], str) and row["icon"]
|
|
|
|
async def test_the_live_bug_the_discovered_list_is_a_superset(self):
|
|
"""The complaint in one assertion: every app the user has, plus the
|
|
curated four, is offered — not just the curated four."""
|
|
from nsh.const import DEFAULT_APP_DIRECTORY
|
|
|
|
poller = _DiscoveryPoller(dumps={"ar.tvplayer.tv": DUMP_TIVIMATE})
|
|
manager = _manager(poller)
|
|
|
|
rows = await manager.async_refresh_installed_apps()
|
|
packages = {row["package"] for row in rows}
|
|
|
|
# the curated list the integration used to fall back to
|
|
assert set(DEFAULT_APP_DIRECTORY) <= packages
|
|
# ...and the apps it missed, which is what the user reported
|
|
for package in (
|
|
"ar.tvplayer.tv", # TiviMate
|
|
"com.github.damontecres.wholphin", # Wholphin
|
|
"com.netflix.ninja", # Netflix
|
|
"com.google.android.youtube.tv", # YouTube
|
|
"org.xbmc.kodi",
|
|
"com.plexapp.android",
|
|
"com.spotify.tv.android",
|
|
"com.example.handmade",
|
|
):
|
|
assert package in packages, package
|
|
assert "com.google.android.tvlauncher" in packages # + the launcher
|
|
# a system app the device never lists as -3 is NOT invented here
|
|
assert "com.android.systemui" not in packages
|
|
|
|
async def test_the_sweep_issues_the_documented_commands(self):
|
|
poller = _DiscoveryPoller(dumps={"ar.tvplayer.tv": DUMP_TIVIMATE})
|
|
manager = _manager(poller)
|
|
|
|
await manager.async_refresh_installed_apps()
|
|
|
|
assert "pm list packages -3" in poller.shell_calls
|
|
assert any("resolve-activity" in cmd for cmd in poller.shell_calls)
|
|
assert "android.intent.category.HOME" in " ".join(poller.shell_calls)
|
|
|
|
async def test_labels_are_fetched_in_a_bounded_burst(self):
|
|
"""A full sweep must not serialise one shell at a time, and must not flood."""
|
|
poller = _DiscoveryPoller()
|
|
manager = _manager(poller)
|
|
concurrent = {"now": 0, "peak": 0, "total": 0}
|
|
|
|
real_shell = poller.shell
|
|
|
|
async def _counted(cmd):
|
|
if "pm dump" in cmd:
|
|
concurrent["now"] += 1
|
|
concurrent["total"] += 1
|
|
concurrent["peak"] = max(concurrent["peak"], concurrent["now"])
|
|
try:
|
|
await asyncio.sleep(0) # yield so overlap is observable
|
|
return await real_shell(cmd)
|
|
finally:
|
|
concurrent["now"] -= 1
|
|
return await real_shell(cmd)
|
|
|
|
poller.shell = _counted
|
|
rows = await manager.async_refresh_installed_apps()
|
|
|
|
assert len(rows) >= 5
|
|
assert concurrent["total"] >= 5
|
|
# bounded: never more than the configured workers in flight at once
|
|
from nsh.const import APP_SWEEP_CONCURRENCY
|
|
|
|
assert concurrent["peak"] <= APP_SWEEP_CONCURRENCY
|
|
assert concurrent["peak"] > 1 # ...and genuinely concurrent, not serial
|
|
|
|
async def test_a_cached_table_is_reused_and_costs_no_shell(self):
|
|
poller = _DiscoveryPoller(dumps={"ar.tvplayer.tv": DUMP_TIVIMATE})
|
|
manager = _manager(poller)
|
|
|
|
first = await manager.async_refresh_installed_apps()
|
|
calls = len(poller.shell_calls)
|
|
second = await manager.async_refresh_installed_apps()
|
|
|
|
assert second == first
|
|
assert len(poller.shell_calls) == calls # cached: zero new shells
|
|
assert manager.installed_apps_stale is False
|
|
|
|
async def test_force_refreshes_even_a_fresh_table(self):
|
|
poller = _DiscoveryPoller()
|
|
manager = _manager(poller)
|
|
|
|
await manager.async_refresh_installed_apps()
|
|
calls = len(poller.shell_calls)
|
|
await manager.async_refresh_installed_apps(force=True)
|
|
|
|
assert len(poller.shell_calls) > calls
|
|
|
|
async def test_a_stale_table_is_not_served(self):
|
|
now = [1000.0]
|
|
poller = _DiscoveryPoller()
|
|
manager = ShieldConnectionManager(
|
|
hass=None, entry=None, poller=poller, clock=lambda: now[0]
|
|
)
|
|
await manager.async_refresh_installed_apps()
|
|
assert manager.installed_apps()
|
|
|
|
now[0] += APP_LIST_TTL + 1
|
|
assert manager.installed_apps() == []
|
|
assert manager.installed_apps_stale is True
|
|
|
|
async def test_concurrent_callers_share_one_sweep(self):
|
|
poller = _DiscoveryPoller()
|
|
manager = _manager(poller)
|
|
list_calls = {"n": 0}
|
|
real_shell = poller.shell
|
|
|
|
async def _counted(cmd):
|
|
if "pm list packages" in cmd:
|
|
list_calls["n"] += 1
|
|
await asyncio.sleep(0.01)
|
|
return await real_shell(cmd)
|
|
|
|
poller.shell = _counted
|
|
results = await asyncio.gather(
|
|
manager.async_refresh_installed_apps(force=True),
|
|
manager.async_refresh_installed_apps(force=True),
|
|
manager.async_refresh_installed_apps(force=True),
|
|
)
|
|
|
|
assert list_calls["n"] == 1 # one sweep, shared
|
|
assert results[0] == results[1] == results[2]
|
|
|
|
|
|
class TestDiscoveryDegradesGracefully:
|
|
"""ADB off, down or mute: an empty list and a note, never a crash or a loop."""
|
|
|
|
async def test_no_poller_is_an_empty_list(self):
|
|
manager = ShieldConnectionManager(hass=None, entry=None, poller=None)
|
|
|
|
rows = await manager.async_refresh_installed_apps()
|
|
|
|
assert rows == []
|
|
assert manager.installed_apps() == []
|
|
|
|
async def test_a_disconnected_poller_is_an_empty_list(self):
|
|
poller = _DiscoveryPoller()
|
|
manager = _manager(poller)
|
|
manager._poller = poller
|
|
poller.connected = False
|
|
# the manager's own connected flag is what the sweep reads
|
|
assert manager.connected is False
|
|
|
|
rows = await manager.async_refresh_installed_apps()
|
|
|
|
assert rows == []
|
|
assert poller.shell_calls == [] # nothing was even attempted
|
|
|
|
async def test_a_mute_transport_reports_empty_without_raising(self):
|
|
poller = _DiscoveryPoller(failure=True)
|
|
manager = _manager(poller)
|
|
|
|
rows = await manager.async_refresh_installed_apps()
|
|
|
|
assert rows == []
|
|
assert poller.shell_calls # it tried
|
|
# and a second call does NOT re-shell in a loop: the empty answer is
|
|
# stamped, and only an explicit force/TTL asks again
|
|
calls = len(poller.shell_calls)
|
|
assert await manager.async_refresh_installed_apps() == []
|
|
assert len(poller.shell_calls) == calls
|
|
|
|
async def test_the_accessor_never_shells(self):
|
|
poller = _DiscoveryPoller()
|
|
manager = _manager(poller)
|
|
|
|
assert manager.installed_apps() == []
|
|
assert poller.shell_calls == []
|
|
|
|
|
|
class TestInstalledAppsSensor:
|
|
"""The dashboard surface: state = count, attribute = the table."""
|
|
|
|
def _sensor(self, manager, options=None):
|
|
sensor_module = load_entity_modules()["sensor"]
|
|
entry = type("E", (), {})()
|
|
entry.entry_id = "e1"
|
|
entry.options = dict(options or {})
|
|
entry.runtime_data = manager
|
|
entry.title = "mdarcy"
|
|
sensor = sensor_module.NvidiaShieldAppsSensor.__new__(
|
|
sensor_module.NvidiaShieldAppsSensor
|
|
)
|
|
sensor.hass = entity_stublib.FakeHass()
|
|
sensor._entry = entry
|
|
return sensor
|
|
|
|
async def test_state_is_the_count_and_the_attribute_is_the_table(self):
|
|
poller = _DiscoveryPoller(dumps={"ar.tvplayer.tv": DUMP_TIVIMATE})
|
|
manager = _manager(poller)
|
|
await manager.async_refresh_installed_apps()
|
|
sensor = self._sensor(manager)
|
|
|
|
assert sensor.native_value == len(manager.installed_apps())
|
|
attrs = sensor.extra_state_attributes
|
|
assert attrs["discovery"] == "ok"
|
|
assert any(
|
|
app["package"] == "ar.tvplayer.tv" and app["name"] == "TiviMate"
|
|
for app in attrs["installed_apps"]
|
|
)
|
|
|
|
def test_no_discovery_yet_is_an_empty_table_with_a_reason(self):
|
|
poller = _DiscoveryPoller()
|
|
manager = _manager(poller)
|
|
sensor = self._sensor(manager)
|
|
|
|
assert sensor.native_value == 0
|
|
attrs = sensor.extra_state_attributes
|
|
assert attrs["installed_apps"] == []
|
|
assert attrs["discovery"] in ("pending", "not_connected", "adb_disabled")
|
|
|
|
def test_adb_disabled_is_reported_as_such(self):
|
|
poller = _DiscoveryPoller()
|
|
manager = _manager(poller)
|
|
poller.connected = False
|
|
sensor = self._sensor(manager, {"adb_enabled": False})
|
|
|
|
attrs = sensor.extra_state_attributes
|
|
|
|
assert attrs == {"installed_apps": [], "discovery": "adb_disabled"}
|
|
|
|
async def test_the_sensor_refresh_is_a_no_op_without_a_manager(self):
|
|
sensor = self._sensor(None)
|
|
sensor._entry.runtime_data = None
|
|
|
|
assert sensor.native_value == 0
|
|
assert sensor.extra_state_attributes["installed_apps"] == []
|
|
await sensor.async_update() # must not raise
|
|
|
|
|
|
class TestDiscoveredAppsStillRespectTheUsersMap:
|
|
"""The user's own map is an OVERRIDE, and discovery fills the rest (t57)."""
|
|
|
|
async def test_a_mapped_package_keeps_the_users_name_not_the_device_label(self):
|
|
poller = _DiscoveryPoller(dumps={"ar.tvplayer.tv": DUMP_TIVIMATE})
|
|
manager = _manager(poller)
|
|
await manager.async_refresh_installed_apps()
|
|
|
|
# the shared resolver is what the sensor and the player use; discovery
|
|
# only supplies the device label underneath the user's map
|
|
from nsh.media_player import resolve_app_name
|
|
|
|
options = {CONF_APP_NAME_MAP: {"ar.tvplayer.tv": "Telly"}}
|
|
assert (
|
|
resolve_app_name(
|
|
"ar.tvplayer.tv", options[CONF_APP_NAME_MAP], manager.app_label
|
|
)
|
|
== "Telly"
|
|
)
|
|
# ...and the device label is what discovery put UNDER it, so dropping
|
|
# the map entry still resolves to a real name rather than a package id
|
|
assert (
|
|
resolve_app_name("ar.tvplayer.tv", {}, manager.app_label) == "TiviMate"
|
|
)
|
|
|
|
|
|
def test_the_label_exit_marker_never_becomes_a_name():
|
|
"""The marker is a transport detail; it must not leak into a label."""
|
|
from nsh.ha_adb_common import parse_application_label
|
|
|
|
assert parse_application_label("__NSH_EXIT:0\n") is None
|
|
assert parse_application_label(DUMP_TIVIMATE + "__NSH_EXIT:0\n") == "TiviMate"
|
|
|
|
|
|
def test_importing_the_discovery_module_needs_no_home_assistant():
|
|
"""The parsers live in ha_adb_common and stay importable without HA."""
|
|
import nsh.ha_adb_common.installed_apps as module
|
|
|
|
assert module.__doc__ and "pm list packages -3" in module.__doc__
|
|
assert not [n for n in dir(module) if n.startswith("hass")]
|