"""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:``; ``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_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")]