From 91d9fb388332a1ae0c47efec9b510da453ce18ec Mon Sep 17 00:00:00 2001 From: James Mackay Date: Mon, 12 Oct 2026 12:26:18 +0100 Subject: [PATCH] feat(intents): discover an app's intents from its activity resolver table (t16) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First half of the approved intents feature (docs/INTENTS_PLAN.md §2): the scan that answers "what can this app actually be asked to do". No UI is wired here — this is the discovery module, the manager path it rides on, and the tests. THE DUMP IS FILTERED ON THE DEVICE. On the user's Shield `pm dump com.netflix.ninja` is ~2241 lines of whole dumpsys for ONE app, so shipping it over ADB to read a few dozen resolver lines repeats the mistake `| head -300` made for app labels (t66 — where a truncation artefact was read as "this app has no label"). The fetch now sends: pm dump 2>/dev/null | sed -n "/Activity Resolver Table/,/Service Resolver Table/p" | head -n 400 with a bounded `grep -A 400 -m 1 -F 'Activity Resolver Table:'` fallback for a device whose toolbox cannot do sed ranges, tried only when the primary answer did not contain the section. The package id is the only caller-supplied text in either command and is always shlex.quote'd, and the sed script is built from two constants — tests drive eight hostile package ids (semicolons, &&, backticks, $(), pipes, both quote kinds, a newline) and assert that the shell sees the package as exactly one verbatim token and that no extra command token appears. WHAT A CANDIDATE CARRIES: component, action, categories and the data (scheme/authority/path, plus the matcher kind — `/title.*` is a GLOB, not a path, and a caller that treats it as literal builds the wrong URI). A filter carrying several schemes/authorities/paths expands to one candidate per combination, because each is a concrete match, and dedup collapses the coincidences. MIME keys are deliberately NOT modelled (they are not one of §3's fields) — such a filter is still reported through its component, action and categories rather than being invented into a URI. RANK, NEVER HIDE. The four families the plan names as internal-looking noise (DEBUG_*, androidx.*, com.google.firebase.*, ConstraintProxy*) are FLAGGED, not deleted: they sort last so the useful rows come first, and the user still decides. Ordering is total (flag, action, scheme, authority, path, component, categories) so a re-scan is byte-identical, and the 200-candidate cap is applied after dedup and DISCLOSED through truncated/found — a capped answer never looks complete. THE CACHE KEEPS AN ANSWER AND A FAILURE APART, structurally: put() is its only writer, only a scan that RAN calls it, and it derives the entry's state — an empty candidate list is INTENT_ENTRY_EMPTY, a real ANSWER the device gave, while a scan whose transport failed caches nothing and is retried. A stale entry is dropped on read rather than served. The scan itself has one status per outcome (fresh / fresh_empty / cached / adb_disabled / adb_unavailable / unreadable / invalid_package) so "ADB is off", "the device did not answer" and "this app advertises nothing" can never be confused, and the not-answered statuses carry no candidates at all — an ADB-down scan cannot look like an app with no intents. ADB disabled and ADB unreachable are told apart from the entry's own option. Discovery is EXPLICIT and single: one package per request, never a sweep over the installed apps, and a test drives a watchdog cycle and asserts it starts no scan. Concurrent requests for the same package share one dump. THE FIXTURE IS A TRANSCRIPTION, NOT IMAGINATION. tests/fixtures/ pm_dump_netflix_resolver.txt reproduces the user's 2026-10-12 capture structure (whole-dumpsys envelope, the activity table with its Schemes/Non-Data Actions groups, the receiver/service/provider tables that bound it) and the test file's docstring states line by line what is VERBATIM from the ticket/plan and what is reconstructed — no real Shield was available, and the raw capture is not in this repository (the same limit t13 recorded). The capture's own entries are each pinned by an assertion: the cast entry, http/https with the four categories and the five www.netflix.com path globs, nflx:, netflix:, MAIN with LEANBACK_LAUNCHER + LAUNCHER, NETFLIX_KEY_START, DIAL_START and the two DEBUG_* noise actions — 18 candidates, deduplicated, with the receiver/provider tables proven unread. Evidence: 59 new tests in tests/test_intents.py (parser against the transcribed capture, degradation on shapes never seen, the command shape and quoting safety, the cache states and TTL, and the manager: one shell, cached, retried on failure, fallback recovery, ADB-off versus ADB-down, the cap disclosed, the watchdog silent, concurrent sharing). No device was available: the parser is proven against the transcription, and every "runs/does not run" claim is an assertion on the poller's recorded commands rather than on intentions. Verify: suites 778/4 and 779/3; ruff clean; drift exit 0 (26 modules, 76 imports, 242 relative imports, 1417 statement blocks, 57 call arguments) with the relative-import floor raised 195 -> 235 and the statement-block floor 1300 -> 1400 by the new module rather than dodged; the real-HA glob still runs green (56 passed, 0 skipped). --- custom_components/nvidia_shield/connection.py | 167 ++++ .../nvidia_shield/ha_adb_common/__init__.py | 46 + .../nvidia_shield/ha_adb_common/intents.py | 692 ++++++++++++++ tests/fixtures/pm_dump_netflix_resolver.txt | 101 +++ tests/ha_api_drift.py | 18 +- tests/test_intents.py | 850 ++++++++++++++++++ 6 files changed, 1869 insertions(+), 5 deletions(-) create mode 100644 custom_components/nvidia_shield/ha_adb_common/intents.py create mode 100644 tests/fixtures/pm_dump_netflix_resolver.txt create mode 100644 tests/test_intents.py diff --git a/custom_components/nvidia_shield/connection.py b/custom_components/nvidia_shield/connection.py index 3d8bdc0..924b8c2 100644 --- a/custom_components/nvidia_shield/connection.py +++ b/custom_components/nvidia_shield/connection.py @@ -45,6 +45,7 @@ from .const import ( APP_LIST_TTL, APP_SWEEP_CONCURRENCY, APP_SWEEP_SPACING, + CONF_ADB_ENABLED, CONF_APP_NAME_MAP, CONF_LOGCAT_EVENTS, DEFAULT_APP_DIRECTORY, @@ -59,13 +60,26 @@ from .const import ( SYSTEM_MEDIA_PACKAGES, ) from .ha_adb_common import ( + INTENT_SCAN_ADB_DISABLED, + INTENT_SCAN_ADB_UNAVAILABLE, + INTENT_SCAN_CACHED, + INTENT_SCAN_FRESH, + INTENT_SCAN_FRESH_EMPTY, + INTENT_SCAN_INVALID_PACKAGE, + INTENT_SCAN_UNREADABLE, + IntentScanCache, + IntentScanResult, MediaSessionInfo, foreground_from_logcat, + has_activity_resolver_table, icon_for_package, installed_packages, + intent_scan_command, + intent_scan_fallback_command, is_launchable, parse_app_label, parse_audio_playback, + parse_intent_report, resolve_package_name, ) @@ -292,6 +306,16 @@ class ShieldConnectionManager: self._logcat_ladder_sleep = asyncio.sleep # Injectable clock keeps the TTL expiry tests deterministic. self._clock = clock + # t16: intent discovery. ONE package per explicit request — never a sweep + # over the installed apps (that is ~2241 dump lines each) — cached with + # INTENT_SCAN_TTL. The cache is the module's, so the "a scan that RAN and + # found nothing is an answer, a scan whose transport failed is not" rule + # lives in one place (IntentScanCache.put is its only writer). + self._intents = IntentScanCache(clock=self._clock) + #: t16: scans in flight, keyed by package, so a second request for the + #: same app awaits the dump already being read instead of paying for a + #: second one. + self._intent_scans: dict[str, asyncio.Task] = {} self._tasks: list[asyncio.Task] = [] self._stopping = False self._failures = 0 @@ -1503,6 +1527,149 @@ class ShieldConnectionManager: self._create_task(self._fetch_app_label(package), f"app_label_{package}") ) + # ── Intent discovery (t16) ── + + @property + def intent_cache(self) -> IntentScanCache: + """The per-package intent-scan cache (t16) — tests, and the options page.""" + return self._intents + + def intent_scan_state(self, package: str | None) -> str | None: + """Which cached ANSWER holds for a package: "candidates", "empty", None (t16). + + None means no answer is cached: the package was never scanned, or the + last scan could not deliver one (a failed scan is never cached, so the + next request retries instead of serving "this app advertises nothing"). + """ + return self._intents.state(package) + + def _adb_disabled(self) -> bool: + """Is ADB switched OFF for this entry (t16)? + + The config option is the honest signal: no poller can mean "off" OR + "enabled but not connected yet", and those are different statuses with + different fixes (turn ADB on / pair it, versus wait or reconnect). + """ + options = getattr(self.entry, "options", None) or {} + return not options.get(CONF_ADB_ENABLED, False) + + async def async_scan_intents( + self, package: str, force: bool = False + ) -> IntentScanResult: + """Scan ONE app's advertised intents over ADB, on demand (t16). + + EXPLICIT AND SINGLE (the plan's ruling): one package per call, triggered + by the user scanning that app — never a sweep over the installed apps, + which on the user's Shield would be ~2241 dump lines each. The caller + gets an :class:`IntentScanResult` whose STATUS says what happened, so + "ADB is off", "the device did not answer" and "this app advertises no + intents" can never be confused for one another. + + Cached per package with ``INTENT_SCAN_TTL`` unless ``force``: a cached + answer is returned as ``cached`` with its age and its entry state, and + only a scan that RAN is ever cached (see :class:`IntentScanCache`). + """ + name = (package or "").strip() + if not is_launchable(name): + # A malformed id never reaches a shell, and never looks like an + # answer: the same rule the sweep applies to the rows it keeps. + return IntentScanResult(package=name, status=INTENT_SCAN_INVALID_PACKAGE) + if not force: + cached = self._intents.get(name) + if cached is not None: + return IntentScanResult( + package=name, + status=INTENT_SCAN_CACHED, + candidates=cached.candidates, + entry_state=cached.state, + age_seconds=self._intents.age(name), + truncated=cached.truncated, + ) + if self._poller is None: + status = ( + INTENT_SCAN_ADB_DISABLED + if self._adb_disabled() + else INTENT_SCAN_ADB_UNAVAILABLE + ) + _LOGGER.debug("intent scan for %s not attempted (%s)", name, status) + return IntentScanResult(package=name, status=status) + if not self.connected: + _LOGGER.debug( + "intent scan for %s not attempted (ADB connected: no) — the " + "device cannot be asked right now", + name, + ) + return IntentScanResult(package=name, status=INTENT_SCAN_ADB_UNAVAILABLE) + running = self._intent_scans.get(name) + if running is not None and not running.done(): + # One dump per package at a time: a second request awaits the scan + # already in flight rather than paying for a second one. + return await asyncio.shield(running) + task = self._create_task(self._scan_intents(name), f"intents_{name}") + self._intent_scans[name] = task + try: + return await task + finally: + if self._intent_scans.get(name) is task: + self._intent_scans.pop(name, None) + + async def _scan_intents(self, package: str) -> IntentScanResult: + """The scan itself: the on-device filter, the bounded fallback, the cache (t16). + + The status vocabulary is the point. A capture that carries the resolver + section is an ANSWER — cached even when it holds no candidates — while + anything else (a shell that raised, no output, an error line, a filtered + answer we cannot read) is NOT an answer, is NOT cached, and is retried. + That is the discovery-cache defect's rule applied to intents, and it is + why ``has_activity_resolver_table`` is asked before the parser is + believed. + """ + poller = self._poller + if poller is None: # pragma: no cover - the caller checked this + return IntentScanResult(package=package, status=INTENT_SCAN_ADB_UNAVAILABLE) + delivered_something = False + for command in ( + intent_scan_command(package), + intent_scan_fallback_command(package), + ): + try: + answer = await poller.shell(command) + except Exception as err: # noqa: BLE001 - the shell's failure IS the finding + _LOGGER.debug("intent scan for %s failed (%s)", package, err) + continue + if not answer or not answer.strip(): + continue + delivered_something = True + if not has_activity_resolver_table(answer): + # The device answered, but not with the resolver section (a + # "Unable to find package" line, a truncated answer): worth one + # bounded fallback attempt, and never a cached answer. + continue + report = parse_intent_report(answer) + entry = self._intents.put( + package, report.candidates, truncated=report.truncated + ) + _LOGGER.debug( + "intent scan for %s answered: %d candidate(s), state=%s, " + "truncated=%s", + package, + len(entry.candidates), + entry.state, + entry.truncated, + ) + return IntentScanResult( + package=package, + status=INTENT_SCAN_FRESH if report.candidates else INTENT_SCAN_FRESH_EMPTY, + candidates=entry.candidates, + entry_state=entry.state, + age_seconds=0.0, + truncated=entry.truncated, + ) + return IntentScanResult( + package=package, + status=INTENT_SCAN_UNREADABLE if delivered_something else INTENT_SCAN_ADB_UNAVAILABLE, + ) + def last_known_map(self) -> dict[str, MediaSessionInfo]: """TTL-filtered snapshot of the per-package last-known cache.""" result: dict[str, MediaSessionInfo] = {} diff --git a/custom_components/nvidia_shield/ha_adb_common/__init__.py b/custom_components/nvidia_shield/ha_adb_common/__init__.py index 724e0b6..d9ff5c1 100644 --- a/custom_components/nvidia_shield/ha_adb_common/__init__.py +++ b/custom_components/nvidia_shield/ha_adb_common/__init__.py @@ -24,6 +24,30 @@ from .installed_apps import ( parse_package_list, ) from .intent import PlaybackIntent +from .intents import ( + INTENT_CANDIDATE_LIMIT, + INTENT_ENTRY_CANDIDATES, + INTENT_ENTRY_EMPTY, + INTENT_SCAN_ADB_DISABLED, + INTENT_SCAN_ADB_UNAVAILABLE, + INTENT_SCAN_CACHED, + INTENT_SCAN_FRESH, + INTENT_SCAN_FRESH_EMPTY, + INTENT_SCAN_INVALID_PACKAGE, + INTENT_SCAN_TTL, + INTENT_SCAN_UNREADABLE, + INTERNAL_MARKERS, + IntentCandidate, + IntentScanCache, + IntentScanResult, + activity_resolver_section, + has_activity_resolver_table, + intent_scan_command, + intent_scan_fallback_command, + is_likely_internal, + parse_activity_resolver_table, + parse_intent_report, +) from .keys import ensure_adb_key from .logcat_events import foreground_from_logcat from .media_session import ( @@ -41,8 +65,30 @@ __all__ = [ "ATTR_MEDIA_DAG_ROLE", "ATTR_MEDIA_DAG_RUNNING_APPS", "AdbClientBase", + "INTENT_CANDIDATE_LIMIT", + "INTENT_ENTRY_CANDIDATES", + "INTENT_ENTRY_EMPTY", + "INTENT_SCAN_ADB_DISABLED", + "INTENT_SCAN_ADB_UNAVAILABLE", + "INTENT_SCAN_CACHED", + "INTENT_SCAN_FRESH", + "INTENT_SCAN_FRESH_EMPTY", + "INTENT_SCAN_INVALID_PACKAGE", + "INTENT_SCAN_TTL", + "INTENT_SCAN_UNREADABLE", + "INTERNAL_MARKERS", + "IntentCandidate", + "IntentScanCache", + "IntentScanResult", "MediaSessionInfo", "PlaybackIntent", + "activity_resolver_section", + "has_activity_resolver_table", + "intent_scan_command", + "intent_scan_fallback_command", + "is_likely_internal", + "parse_activity_resolver_table", + "parse_intent_report", "ensure_adb_key", "parse_all_media_sessions", "parse_app_label", diff --git a/custom_components/nvidia_shield/ha_adb_common/intents.py b/custom_components/nvidia_shield/ha_adb_common/intents.py new file mode 100644 index 0000000..fb73766 --- /dev/null +++ b/custom_components/nvidia_shield/ha_adb_common/intents.py @@ -0,0 +1,692 @@ +"""Intent discovery from an app's Activity Resolver Table (t16). + +The approved intents feature (``docs/INTENTS_PLAN.md`` §2) starts here: the user +scans ONE installed app and we answer with the intents that app advertises. The +source of truth is the ``Activity Resolver Table:`` section Android prints in +``pm dump ``. + +WHY THE DUMP IS FILTERED ON THE DEVICE. On the user's Shield (captured +2026-10-12) ``pm dump com.netflix.ninja`` is ~2241 lines of WHOLE dumpsys — +``DUMP OF SERVICE package:`` and then activity/meminfo/procstats/usagestats/ +batterystats — for ONE app. Shipping that over ADB to read a few dozen resolver +lines is the same mistake ``| head -300`` made for app labels (t66, where the +truncation was misread as "this app has no label"), so the command built here +pipes the dump through ``sed`` ON THE DEVICE and only the resolver section +crosses the wire: + + pm dump 2>/dev/null \ + | sed -n "/Activity Resolver Table/,/Service Resolver Table/p" \ + | head -n + +with a bounded ``grep -A … -m 1 -F`` fallback (``intent_scan_fallback_command``) +for a device whose toolbox cannot do sed ranges. The package id is always +``shlex.quote``d — a package name is the only thing in either command that could +carry shell metacharacters, and quoting it is what makes the command safe — and +everything else in the command is a fixed constant from this file. + +WHAT THE CANDIDATES CARRY AND WHY. Per §3's data model: component, action, +categories, and the data (scheme, authority, path). A filter that carries several +schemes, authorities or paths is expanded into ONE CANDIDATE PER COMBINATION, +because each of those is a concrete match — a URI the app really accepts — and a +picker row that mixes two schemes cannot be turned into a URI at all. Candidates +are deduplicated on the whole tuple, so the expansion costs nothing when the +app's filters coincide. + +WHAT THIS MODULE DOES NOT MODEL: MIME types. The resolver key of a MIME filter +(``video/*:``) is not one of §3's fields, and inventing a data URI out of it +would be worse than reporting the filter through its component, action and +categories — which is what happens. Nothing is dropped for being unmodelled. + +MARK — READ THIS BEFORE TRUSTING A SHAPE. The parser is proven against a +TRANSCRIBED capture (``tests/fixtures/pm_dump_netflix_resolver.txt``), not +against a device: no real Shield was available, and the raw capture text is not +in this repository either. The entry/detail shapes below are what the capture +records (see the fixture's own notes for which lines are verbatim and which are +reconstructed), and every parser branch is deliberately tolerant of a shape we +have never seen: an unrecognised line degrades to "no candidate from this line", +never to a wrong candidate. + +NOT TO BE CONFUSED WITH ``.intent``: that module holds ``PlaybackIntent``, the +optimistic-UI dataclass, not an Android intent. §7.4 of the plan suggests renaming +it to ``playback_intent.py``; that churn is not part of this unit, so the two names +sit side by side here and this note is the signpost. +""" + +from __future__ import annotations + +import re +import shlex +import time +from collections.abc import Iterator +from dataclasses import dataclass + +#: ``Activity Resolver Table:`` — the section that opens the package's resolver +#: tables, and the two headings that END it on the observed capture (the receiver +#: table follows the activity one, then the service table). +ACTIVITY_TABLE_HEADER = "Activity Resolver Table:" +RECEIVER_TABLE_HEADER = "Receiver Resolver Table:" +SERVICE_TABLE_HEADER = "Service Resolver Table:" +PROVIDER_TABLE_HEADER = "Provider Resolver Table:" + +#: The end of the activity section: whichever of these comes FIRST. Stopping at +#: the first one can only ever cut the section short, never extend it into +#: another table (the plan's sed uses the service heading; a capture where the +#: receiver table is absent still ends here). +_SECTION_ENDS = ( + RECEIVER_TABLE_HEADER, + SERVICE_TABLE_HEADER, + PROVIDER_TABLE_HEADER, +) + +#: The two patterns the ON-DEVICE sed range is built from. The colons are dropped +#: deliberately: the shipped command is the plan's own line +#: (``sed -n '/Activity Resolver Table/,/Service Resolver Table/p'``), which is +#: what has to be recognisable in a shell and in a review, and both patterns match +#: the same printed heading. They are constants — never caller text — so quoting +#: them in the command below cannot be broken from outside. +SED_RANGE_START = "Activity Resolver Table" +SED_RANGE_END = "Service Resolver Table" + +#: How long a scan that RAN is reused, per package. An app's advertised intents +#: only change when the app does, so this is a cost bound rather than a staleness +#: guess: one shell an hour for a package the user keeps looking at. +INTENT_SCAN_TTL = 3600.0 + +#: The transfer bound of both commands (``head -n`` / ``grep -A``). A real +#: resolver section is tens of lines; this is what keeps a pathological table +#: from becoming a wall of text on the wire. +INTENT_SCAN_LINE_LIMIT = 400 + +#: The cap on returned candidates, applied AFTER dedup and ranking. The cap is +#: disclosed (``IntentParseResult.truncated`` / ``IntentScanResult.truncated``), +#: so a capped answer never looks complete. +INTENT_CANDIDATE_LIMIT = 200 + +#: One entry line: an optional group key, the component's hex id, the COMPONENT +#: and — for an activity filter — the filter's own id. Both shapes the capture +#: records match, with or without the group key in front: +#: +#: cast: deca7fa com.netflix.ninja/.MainActivity filter fc483a1 +#: deca7fa com.netflix.ninja/.MainActivity filter fc483a2 +_ENTRY_RE = re.compile( + r"^(?P.*?)\s*(?P[0-9a-fA-F]{6,})\s+" + r"(?P[A-Za-z_][\w.]*/[\w.$]+)" + r"(?:\s+filter\s+(?P[0-9a-fA-F]+))?\s*$" +) + +#: A group heading on its own line (``Schemes:``, ``Non-Data Actions:``, and the +#: action-keyed groups inside it). Used for ONE fallback only: when a filter +#: carries no ``Action:`` detail, the key that grouped it is the action the +#: device printed — never a guess about a key that is not action-shaped. +_GROUP_KEY_RE = re.compile(r"^\s+(?P[A-Za-z_][\w.$]*)\s*:\s*$") + +#: A filter detail line. The five fields §3's model carries; ``Type:`` (a MIME +#: type) and anything else are read off the line and ignored — see the module +#: docstring on what is not modelled. +_DETAIL_RE = re.compile( + r"^\s*(?PAction|Category|Scheme|Authority|Path)\s*:\s*(?P.+?)\s*$" +) + +#: ``Authority: "www.netflix.com": -1`` — the authority, not its weight. +_AUTHORITY_RE = re.compile(r'^"(?P[^"]*)"\s*(?::\s*-?\d+)?$') + +#: ``Path: "PatternMatcher{GLOB: /title.*}"`` — the matcher kind and its value. +#: Both separators Android prints are accepted (``GLOB: /x`` and ``LITERAL, /x``). +_PATH_MATCHER_RE = re.compile( + r"^PatternMatcher\{(?P[A-Za-z_]+)\s*[:,]?\s*(?P.*)\}$" +) + +#: The marker families ``docs/INTENTS_PLAN.md`` §2 names as internal-looking. +#: They are FLAGGED and never deleted: the user decides what is worth +#: registering, and a candidate we quietly removed is one they can never +#: register. Matching is on the action and the component (the capture's own +#: noise is action-shaped, e.g. ``com.netflix.action.DEBUG_PUSH_NOTIFICATION``). +INTERNAL_MARKERS = ( + "DEBUG_", + "androidx.", + "com.google.firebase.", + "ConstraintProxy", +) + +#: A cache entry's state (t16). ONLY a scan that RAN writes an entry, and the +#: entry then says which of these two answers it holds — the distinction the +#: discovery-cache defect taught us, made structural instead of remembered. +INTENT_ENTRY_CANDIDATES = "candidates" +INTENT_ENTRY_EMPTY = "empty" + +#: A scan result's status. ``answered`` is True for the first three only: the +#: rest mean "we do not know", which is NOT the same as "this app advertises +#: nothing" and must never be shown as an empty list of intents. +INTENT_SCAN_FRESH = "fresh" +INTENT_SCAN_FRESH_EMPTY = "fresh_empty" +INTENT_SCAN_CACHED = "cached" +INTENT_SCAN_ADB_DISABLED = "adb_disabled" +INTENT_SCAN_ADB_UNAVAILABLE = "adb_unavailable" +INTENT_SCAN_UNREADABLE = "unreadable" +INTENT_SCAN_INVALID_PACKAGE = "invalid_package" + +#: The statuses whose candidates are an ANSWER from the device. +_ANSWERED_STATUSES = ( + INTENT_SCAN_FRESH, + INTENT_SCAN_FRESH_EMPTY, + INTENT_SCAN_CACHED, +) + + +def _unquote(value: str | None) -> str: + """Strip one layer of matching quotes, if any (``"x"`` -> ``x``).""" + text = (value or "").strip() + if len(text) >= 2 and text[0] == text[-1] and text[0] in ("'", '"'): + return text[1:-1].strip() + return text + + +def _authority_value(raw: str) -> str | None: + """The authority in an ``Authority:`` detail, or None. + + Android prints ``Authority: "www.netflix.com": -1`` (quoted, then the filter's + score); a bare ``Authority: www.netflix.com`` is accepted too. + """ + text = (raw or "").strip() + match = _AUTHORITY_RE.match(text) + if match: + return match.group("value") or None + return _unquote(text) or None + + +def _path_value(raw: str) -> tuple[str | None, str | None]: + """``(matcher_kind, path)`` from a ``Path:`` detail. + + ``"PatternMatcher{GLOB: /title.*}"`` -> ``("GLOB", "/title.*")``; + ``"PatternMatcher{LITERAL, /browse}"`` -> ``("LITERAL", "/browse")``; anything + that is not a PatternMatcher is returned as a bare path with no kind. + """ + text = _unquote(raw) + match = _PATH_MATCHER_RE.match(text) + if match: + return match.group("kind").upper(), match.group("value").strip() + return None, text or None + + +def looks_like_action(key: str | None) -> bool: + """Is a group key an ACTION rather than a sub-heading (t16)? + + ``android.intent.action.MAIN`` and ``com.netflix.action.DIAL_START`` are + action-shaped; ``Schemes``, ``Non-Data Actions`` and the ``cast:``/``http:`` + scheme keys are not. Only used when a filter printed no ``Action:`` detail. + """ + text = (key or "").strip().rstrip(":") + return bool(text) and "." in text and " " not in text + + +def is_likely_internal(action: str | None, component: str | None = None) -> bool: + """Does this candidate LOOK like the app's own internal plumbing (t16)? + + A heuristic, and labelled as one: the four families the plan names are + matched anywhere in the action or the component, and the flag only moves a + candidate to the END of the ordering (``rank_candidates``) — it never removes + it, because the user is the one who decides what is worth registering. + """ + action_text = action or "" + component_text = component or "" + return any( + marker in action_text or marker in component_text + for marker in INTERNAL_MARKERS + ) + + +@dataclass(frozen=True) +class IntentCandidate: + """One intent an app advertises, as its resolver table describes it (t16). + + ``component`` is the ``package/Activity`` the filter belongs to; ``action`` + is the intent action (never invented: see ``looks_like_action``); the data + fields are one concrete match, so ``scheme``/``authority``/``path`` together + describe a URI the app accepts. ``path_kind`` is Android's matcher + (``GLOB``/``LITERAL``/``PREFIX``) and is kept because ``/title.*`` is a glob, + not a path: a caller that treats it as literal builds the wrong URI. + """ + + component: str + action: str | None = None + categories: tuple[str, ...] = () + scheme: str | None = None + authority: str | None = None + path: str | None = None + path_kind: str | None = None + likely_internal: bool = False + + @property + def key(self) -> tuple: + """The dedup key: everything the candidate means, nothing about its row.""" + return ( + self.component, + self.action, + self.categories, + self.scheme, + self.authority, + self.path, + self.path_kind, + ) + + def as_dict(self) -> dict: + """A plain dict, for the options page, diagnostics and JSON.""" + return { + "component": self.component, + "action": self.action, + "categories": list(self.categories), + "scheme": self.scheme, + "authority": self.authority, + "path": self.path, + "path_kind": self.path_kind, + "likely_internal": self.likely_internal, + } + + +@dataclass(frozen=True) +class IntentParseResult: + """What one capture yielded: the candidates, and whether the cap applied.""" + + candidates: tuple[IntentCandidate, ...] = () + found: int = 0 + truncated: bool = False + + +def has_activity_resolver_table(output: str | None) -> bool: + """Does this capture carry the resolver section at all (t16)? + + THE distinction this module exists to keep: a capture WITH the heading that + holds no entries is an ANSWER ("this app advertises nothing we can read"), + while a capture WITHOUT it is not an answer at all — the transport handed us + something else (an error line, a truncated answer, nothing). Only the former + may be cached (``IntentScanCache``); the latter must be retried. + """ + return bool(output) and ACTIVITY_TABLE_HEADER in output + + +def activity_resolver_section(output: str | None) -> str | None: + """The ``Activity Resolver Table:`` section of a capture, or None. + + Bounded by the FIRST of the receiver/service/provider headings, so a capture + that carries the whole dumpsys cannot leak another table's filters (a + receiver's actions are not an activity's intents). None when the section is + absent — see ``has_activity_resolver_table`` for why that is not "no intents". + """ + if not output: + return None + start = output.find(ACTIVITY_TABLE_HEADER) + if start < 0: + return None + body = output[start + len(ACTIVITY_TABLE_HEADER):] + end = len(body) + for heading in _SECTION_ENDS: + found = body.find(heading) + if 0 <= found < end: + end = found + return body[:end] + + +@dataclass +class _Entry: + """A filter line being read: its component, its key, its own filter id.""" + + component: str + key: str | None = None + filter_id: str | None = None + + @property + def group_action(self) -> str | None: + """The group key as an action, when it is action-shaped (else None).""" + key = self.key + return key.rstrip(":") if looks_like_action(key) else None + + +def _iter_filters(section: str) -> Iterator[tuple[_Entry, list[tuple[str, str]]]]: + """Yield ``(entry, details)`` for every filter line in the section. + + Details belong to the filter line they follow, so the walk is line-order and + resets on each new entry. A group key on its own line is remembered for the + action fallback (``_Entry.group_action``). + """ + entry: _Entry | None = None + details: list[tuple[str, str]] = [] + group_key: str | None = None + for line in section.splitlines(): + match = _ENTRY_RE.match(line) + if match: + if entry is not None: + yield entry, details + key = (match.group("key") or "").strip() or group_key + entry = _Entry( + component=match.group("component"), + key=key, + filter_id=match.group("filter_id"), + ) + details = [] + continue + detail = _DETAIL_RE.match(line) + if detail is not None: + if entry is not None: + details.append((detail.group("field"), detail.group("value"))) + continue + heading = _GROUP_KEY_RE.match(line) + if heading is not None: + group_key = heading.group("key") + if entry is not None: + yield entry, details + + +def _data_specs(details: list[tuple[str, str]]) -> list[tuple]: + """The ``(scheme, authority, path, path_kind)`` combinations of one filter. + + ``Scheme:`` opens a group; the ``Authority:``/``Path:`` lines that follow it + belong to it (that is the nesting the capture prints). A filter with none of + the three yields a single all-None spec: "this filter matches on its action + alone", which is exactly what a Non-Data Action is. + """ + groups: list[list] = [] + for field, raw in details: + if field == "Scheme": + groups.append([_unquote(raw) or None, [], []]) + continue + if field not in ("Authority", "Path"): + continue + if not groups: + groups.append([None, [], []]) + target = groups[-1] + if field == "Authority": + value = _authority_value(raw) + if value: + target[1].append(value) + else: + target[2].append(_path_value(raw)) + + specs: list[tuple] = [] + for scheme, authorities, paths in groups: + for authority in authorities or [None]: + for path_kind, path in paths or [(None, None)]: + specs.append((scheme, authority, path, path_kind)) + return specs + + +def _candidates_for(entry: _Entry, details: list[tuple[str, str]]) -> list[IntentCandidate]: + """Every candidate one filter line yields (action x data combination).""" + actions: list[str | None] = [] + categories: list[str] = [] + for field, raw in details: + if field == "Action": + value = _unquote(raw) + if value and value not in actions: + actions.append(value) + elif field == "Category": + value = _unquote(raw) + if value and value not in categories: + categories.append(value) + if not actions: + # No Action: detail. The group key is the action ONLY when it is + # action-shaped; otherwise the candidate keeps action None rather than + # inventing one from a sub-heading. + actions = [entry.group_action] + + specs = _data_specs(details) or [(None, None, None, None)] + candidates: list[IntentCandidate] = [] + for action in actions: + for scheme, authority, path, path_kind in specs: + candidates.append( + IntentCandidate( + component=entry.component, + action=action, + categories=tuple(categories), + scheme=scheme, + authority=authority, + path=path, + path_kind=path_kind, + likely_internal=is_likely_internal(action, entry.component), + ) + ) + return candidates + + +def rank_candidates( + candidates: list[IntentCandidate] | tuple[IntentCandidate, ...], + *, + limit: int = INTENT_CANDIDATE_LIMIT, +) -> list[IntentCandidate]: + """Deduplicate, order deterministically, and cap (t16). + + Order: candidates a user is likely to want first (``likely_internal`` last), + then the dedup key itself — action, scheme, authority, path, component, + categories — so the order is total and identical on a re-scan, which is what + makes it safe for a UI to key rows by position. + + The cap (``limit``, default ``INTENT_CANDIDATE_LIMIT``) keeps a pathological + table from becoming a wall of rows; it is applied AFTER dedup, drops the + least interesting rows first (internal-looking ones) and is DISCLOSED by the + caller (``IntentParseResult.truncated``), never silent. + """ + unique: dict[tuple, IntentCandidate] = {} + for candidate in candidates: + unique.setdefault(candidate.key, candidate) + ordered = sorted( + unique.values(), + key=lambda c: ( + c.likely_internal, + c.action or "", + c.scheme or "", + c.authority or "", + c.path or "", + c.component, + c.categories, + ), + ) + return ordered[:limit] + + +def parse_intent_report( + output: str | None, + *, + limit: int = INTENT_CANDIDATE_LIMIT, +) -> IntentParseResult: + """Parse a resolver capture into a report: candidates, count, capped or not. + + An absent section yields an EMPTY report, not an answer — callers must ask + ``has_activity_resolver_table`` (the manager does) before treating "no + candidates" as "this app advertises nothing". + """ + if not output: + return IntentParseResult() + section = activity_resolver_section(output) + if section is None: + return IntentParseResult() + + raw: list[IntentCandidate] = [] + for entry, details in _iter_filters(section): + raw.extend(_candidates_for(entry, details)) + ranked = rank_candidates(raw, limit=limit) + deduped = len({candidate.key for candidate in raw}) + return IntentParseResult( + candidates=tuple(ranked), + found=deduped, + truncated=deduped > len(ranked), + ) + + +def parse_activity_resolver_table( + output: str | None, + *, + limit: int = INTENT_CANDIDATE_LIMIT, +) -> list[IntentCandidate]: + """The candidates in a ``pm dump`` resolver capture (t16). + + Deduplicated on (component, action, categories, scheme, authority, path), + deterministically ordered and capped — see ``rank_candidates``. An absent + section returns an empty list: this function reports WHAT IS THERE, and the + caller decides whether the capture was an answer at all (the manager checks + ``has_activity_resolver_table`` first — a failed scan is not an answer and is + never cached as one). + """ + return list(parse_intent_report(output, limit=limit).candidates) + + +def intent_scan_command(package: str, *, line_limit: int = INTENT_SCAN_LINE_LIMIT) -> str: + """The ON-DEVICE filter: only the resolver section crosses the wire (t16). + + ``pm dump `` is a whole dumpsys (~2241 lines for one app on the user's + Shield), so the range is applied by the device's own ``sed`` and the answer is + additionally bounded, which is what makes the transfer proportional to the + resolver table rather than to the dump. + + SAFETY: the package id is the only caller-supplied text in the command and it + is always ``shlex.quote``d; the sed script is built from the two constants + above (no interpolation of anything untrusted), so a package id containing + shell metacharacters cannot add a command — the fallback builder is quoted the + same way. A caller must still refuse an id that is not package-shaped: the + manager does (``is_launchable``). + """ + sed_range = f"/{SED_RANGE_START}/,/{SED_RANGE_END}/p" + return ( + f"pm dump {shlex.quote(package)} 2>/dev/null | " + f'sed -n "{sed_range}" | ' + f"head -n {int(line_limit)}" + ) + + +def intent_scan_fallback_command( + package: str, + *, + line_limit: int = INTENT_SCAN_LINE_LIMIT, +) -> str: + """The bounded ``grep -A`` fallback for a device without sed ranges (t16). + + ``pm dump 2>/dev/null | grep -A -m 1 -F 'Activity Resolver Table:'`` + — the same transfer bound as the primary command, applied with the one tool + every Android toolbox has. It is only ever a SECOND attempt: the manager runs + it when the primary answer did not contain the section at all, so a device + whose sed works never pays for it. + """ + return ( + f"pm dump {shlex.quote(package)} 2>/dev/null | " + f"grep -A {int(line_limit)} -m 1 -F {shlex.quote(ACTIVITY_TABLE_HEADER)}" + ) + + +@dataclass +class IntentCacheEntry: + """One cached ANSWER: its state, its candidates and when it was written.""" + + state: str + candidates: tuple[IntentCandidate, ...] + stamp: float + truncated: bool = False + + +class IntentScanCache: + """A per-package TTL cache of scans that RAN (t16). + + The rule the discovery-cache defect taught us is structural here: ``put`` is + the ONLY writer, it is called only by a scan that ran, and it records which + of the two answers it holds — ``INTENT_ENTRY_CANDIDATES`` or + ``INTENT_ENTRY_EMPTY``. A scan whose transport failed never reaches ``put``, + so nothing is cached and the next request retries instead of serving a + fabricated "this app advertises nothing" for the whole TTL. + + A stale entry is DROPPED when it is read (never served), so callers see + either an answer inside its TTL or nothing at all. + """ + + def __init__(self, *, ttl: float = INTENT_SCAN_TTL, clock=time.monotonic): + self._ttl = ttl + self._clock = clock + self._entries: dict[str, IntentCacheEntry] = {} + + @property + def ttl(self) -> float: + return self._ttl + + def get(self, package: str) -> IntentCacheEntry | None: + """The fresh entry for a package, or None (dropping it when stale).""" + entry = self._entries.get(package) + if entry is None: + return None + if self._clock() - entry.stamp > self._ttl: + self._entries.pop(package, None) + return None + return entry + + def put( + self, + package: str, + candidates: list[IntentCandidate] | tuple[IntentCandidate, ...], + *, + truncated: bool = False, + ) -> IntentCacheEntry: + """Record a scan that RAN: its candidates, or the empty ANSWER. + + The state is derived, not passed: an empty candidate list means the + device answered with none (``INTENT_ENTRY_EMPTY``) — an answer worth + caching — while a non-empty list is ``INTENT_ENTRY_CANDIDATES``. + """ + saved = tuple(candidates) + entry = IntentCacheEntry( + state=INTENT_ENTRY_CANDIDATES if saved else INTENT_ENTRY_EMPTY, + candidates=saved, + stamp=self._clock(), + truncated=truncated, + ) + self._entries[package] = entry + return entry + + def state(self, package: str | None) -> str | None: + """The cached entry's state, or None when no fresh answer is cached.""" + if not package: + return None + entry = self.get(package) + return entry.state if entry is not None else None + + def age(self, package: str) -> float | None: + """How long ago the fresh entry was written, or None.""" + entry = self.get(package) + return None if entry is None else self._clock() - entry.stamp + + def clear(self) -> None: + """Forget every entry (tests, and a device switch).""" + self._entries.clear() + + +@dataclass(frozen=True) +class IntentScanResult: + """The answer to one scan request (t16), status included. + + ``candidates`` is only ever an ANSWER when ``answered`` is True. Every other + status means we do not know, which is why they all carry no candidates: an + ADB-down scan must not look like an app with no intents. + """ + + package: str + status: str + candidates: tuple[IntentCandidate, ...] = () + entry_state: str | None = None + age_seconds: float | None = None + truncated: bool = False + + @property + def answered(self) -> bool: + """Did the DEVICE answer? False means "unknown", never "no intents".""" + return self.status in _ANSWERED_STATUSES + + @property + def has_candidates(self) -> bool: + """Did the device answer WITH intents?""" + return bool(self.candidates) + + def as_dict(self) -> dict: + """A plain dict, for the options page, diagnostics and JSON.""" + return { + "package": self.package, + "status": self.status, + "answered": self.answered, + "entry_state": self.entry_state, + "age_seconds": self.age_seconds, + "truncated": self.truncated, + "candidates": [candidate.as_dict() for candidate in self.candidates], + } diff --git a/tests/fixtures/pm_dump_netflix_resolver.txt b/tests/fixtures/pm_dump_netflix_resolver.txt new file mode 100644 index 0000000..eab9e6d --- /dev/null +++ b/tests/fixtures/pm_dump_netflix_resolver.txt @@ -0,0 +1,101 @@ +DUMP OF SERVICE package: +Activity Resolver Table: + Schemes: + cast: deca7fa com.netflix.ninja/.MainActivity filter fc483a1 + Action: "com.google.cast.action.START" + Category: "android.intent.category.DEFAULT" + Scheme: "cast" + http: deca7fa com.netflix.ninja/.MainActivity filter fc483a1 + Action: "android.intent.action.VIEW" + Category: "android.intent.category.DEFAULT" + Category: "android.intent.category.BROWSABLE" + Category: "android.intent.category.LEANBACK_LAUNCHER" + Category: "android.intent.category.LAUNCHER" + Scheme: "http" + Authority: "www.netflix.com": -1 + Path: "PatternMatcher{GLOB: /title.*}" + Path: "PatternMatcher{GLOB: /watch.*}" + Path: "PatternMatcher{LITERAL, /browse}" + Path: "PatternMatcher{LITERAL, /home}" + Path: "PatternMatcher{GLOB: /deeplink.*}" + https: deca7fa com.netflix.ninja/.MainActivity filter fc483a1 + Action: "android.intent.action.VIEW" + Category: "android.intent.category.DEFAULT" + Category: "android.intent.category.BROWSABLE" + Category: "android.intent.category.LEANBACK_LAUNCHER" + Category: "android.intent.category.LAUNCHER" + Scheme: "https" + Authority: "www.netflix.com": -1 + Path: "PatternMatcher{GLOB: /title.*}" + Path: "PatternMatcher{GLOB: /watch.*}" + Path: "PatternMatcher{LITERAL, /browse}" + Path: "PatternMatcher{LITERAL, /home}" + Path: "PatternMatcher{GLOB: /deeplink.*}" + netflix: deca7fa com.netflix.ninja/.MainActivity filter fc483a1 + Action: "android.intent.action.VIEW" + Category: "android.intent.category.DEFAULT" + Scheme: "netflix" + nflx: deca7fa com.netflix.ninja/.MainActivity filter fc483a1 + Action: "android.intent.action.VIEW" + Category: "android.intent.category.DEFAULT" + Scheme: "nflx" + Non-Data Actions: + android.intent.action.MAIN: + deca7fa com.netflix.ninja/.MainActivity filter fc483a2 + Action: "android.intent.action.MAIN" + Category: "android.intent.category.LEANBACK_LAUNCHER" + Category: "android.intent.category.LAUNCHER" + com.netflix.action.NETFLIX_KEY_START: + deca7fa com.netflix.ninja/.MainActivity filter fc483a3 + Action: "com.netflix.action.NETFLIX_KEY_START" + com.netflix.action.DIAL_START: + deca7fa com.netflix.ninja/.MainActivity filter fc483a4 + Action: "com.netflix.action.DIAL_START" + com.netflix.action.DEBUG_PUSH_NOTIFICATION: + deca7fa com.netflix.ninja/.debug.DebugReceiver filter fc483a5 + Action: "com.netflix.action.DEBUG_PUSH_NOTIFICATION" + com.netflix.action.DEBUG_FORCE_EXIT: + deca7fa com.netflix.ninja/.debug.DebugReceiver filter fc483a6 + Action: "com.netflix.action.DEBUG_FORCE_EXIT" +Receiver Resolver Table: + Non-Data Actions: + androidx.work.DiagnosticsReportReceiver: + deca7fa com.netflix.ninja/androidx.work.impl.diagnostics.DiagnosticsReceiver + com.google.firebase.MESSAGING_EVENT: + deca7fa com.netflix.ninja/com.google.firebase.messaging.FirebaseMessagingService + com.netflix.action.DEBUG_PUSH_NOTIFICATION: + deca7fa com.netflix.ninja/.push.DebugPushReceiver +Service Resolver Table: + Non-Data Actions: + com.google.firebase.INSTANCE_ID_EVENT: + deca7fa com.netflix.ninja/com.google.firebase.iid.FirebaseInstanceIdService +Provider Resolver Table: + Authorities: + com.netflix.ninja.androidx-startup: + deca7fa com.netflix.ninja/androidx.startup.InitializationProvider +Permissions: + Permission [com.netflix.ninja.C2D_MESSAGE]: + sourcePackage=com.netflix.ninja +Packages: + Package [com.netflix.ninja] (7f1a2b3c): + userId=10123 + pkg=Package{7f1a2b3c com.netflix.ninja} + nonLocalizedLabel=null + labelRes=0x7f0f0d31 + versionCode=61102 minSdk=21 targetSdk=29 + versionName=6.11.2 build 3413 + codePath=/data/app/com.netflix.ninja-1/base.apk + ConstraintProxy.BatteryBroadcastReceiver= + com.netflix.ninja/androidx.work.impl.constraints.trackers.BatteryChargingProxy + Activity Resolver Table: (nothing else listed for this package) +DUMP OF SERVICE activity: + ACTIVITY MANAGER PENDING INTENTS (dumpsys activity pending) + (nothing) +DUMP OF SERVICE meminfo: + Applications Memory Usage (in Kilobytes): +DUMP OF SERVICE procstats: + Process Stats: +DUMP OF SERVICE usagestats: + In-memory rollup: +DUMP OF SERVICE batterystats: + Battery Stats: diff --git a/tests/ha_api_drift.py b/tests/ha_api_drift.py index fb3ffbe..b25c406 100644 --- a/tests/ha_api_drift.py +++ b/tests/ha_api_drift.py @@ -141,9 +141,9 @@ HA_CALL_OWNERS: dict[str, tuple[str, str]] = { } #: Non-vacuity floors — the ONE source of truth for the standalone entry point -#: and the pytest wrapper (t39 F1). They sit below the current walk (25 modules, -#: 76 imports, 206 relative imports, 57 checked arguments, 6 awaited HA calls, -#: 1322 statement blocks) so a broken or mis-targeted walk fails loudly instead of +#: and the pytest wrapper (t39 F1). They sit below the current walk (26 modules, +#: 76 imports, 242 relative imports, 57 checked arguments, 6 awaited HA calls, +#: 1417 statement blocks) so a broken or mis-targeted walk fails loudly instead of #: printing zeros and exiting 0. FLOORS: dict[str, int] = { # t41: the sweep is recursive, so the floor sits above the top-level-only @@ -154,7 +154,12 @@ FLOORS: dict[str, int] = { # `ATTR_ACTIVITY`, the transport-state accessors and the shared app-directory # helpers all arrive as imports, and none of them may vanish unnoticed. "imports": 70, - "relative_imports": 195, + # t16: raised 190 -> 195 -> 235, again just under the walk. The intent + # discovery module adds the package's exports to `ha_adb_common`'s surface and + # is imported through it by the manager, so the relative-import count is exactly + # where a new module must show up — a floor left behind would let the manager's + # import of it disappear unnoticed. + "relative_imports": 235, # t62 F2: raised from 20 to just under the current 46. A floor's job is to # notice an API OWNER going away, not only a broken walk: at 20, deleting # `async_show_menu` from HA_CALL_OWNERS left 40 checked arguments and still @@ -177,7 +182,10 @@ FLOORS: dict[str, int] = { # floor here means a pass that silently stopped parsing cannot look green. # t15: raised 500 -> 1200 -> 1300, just under the walk, for the same reason the # imports floor moved: the exhaustion pass covers the new surfaces too. - "statement_blocks": 1300, + # t16: raised 1200 -> 1300 -> 1400: the parser, the cache and the manager's + # scan path are new statement blocks, and the unreachable pass must keep + # covering them. + "statement_blocks": 1400, } #: The interpreter floor Home Assistant itself declares: PyPI metadata for diff --git a/tests/test_intents.py b/tests/test_intents.py new file mode 100644 index 0000000..2400522 --- /dev/null +++ b/tests/test_intents.py @@ -0,0 +1,850 @@ +"""tests/test_intents.py — intent discovery from the resolver table (t16). + +THE FIXTURE IS A TRANSCRIPTION, NOT IMAGINATION. ``tests/fixtures/ +pm_dump_netflix_resolver.txt`` reproduces the ``pm dump com.netflix.ninja`` +capture the user took on 2026-10-12 (the structure is recorded in +``docs/INTENTS_PLAN.md`` §1 and the ticket). No real Shield was available while +this was written, and the raw ~2241-line capture text is NOT in this repository +either (the same limit t13 recorded for its own Netflix fixture), so the fixture +is a faithful reproduction of the RECORDED structure rather than the capture +itself. Being precise about that, because a parser "proven" against invented +output is worth nothing: + +* VERBATIM (quoted in the ticket/plan, reproduced character for character as + they appear there): the section heading ``Activity Resolver Table:``; the + bound ``Receiver Resolver Table:`` / ``Service Resolver Table:``; the entry + form ``deca7fa com.netflix.ninja/.MainActivity filter fc483a1`` (hex id, + component, ``filter`` id); the detail lines ``Action:``, ``Category:``, + ``Scheme:``, ``Authority: "www.netflix.com": -1`` and + ``Path: "PatternMatcher{GLOB: /title.*}"``; the cast entry + (``com.google.cast.action.START`` + ``DEFAULT``); the ``http``/``https`` + entries (``android.intent.action.VIEW`` + ``DEFAULT`` + ``LEANBACK_LAUNCHER`` + + ``LAUNCHER`` + ``BROWSABLE``) with the ``www.netflix.com`` authority and the + ``/title.*``, ``/watch.*``, ``/browse``, ``/home``, ``/deeplink.*`` paths; the + ``nflx:``/``netflix:`` schemes; the ``Non-Data Actions:`` group with + ``android.intent.action.MAIN`` (``LEANBACK_LAUNCHER`` + ``LAUNCHER``), + ``com.netflix.action.NETFLIX_KEY_START`` and ``com.netflix.action.DIAL_START``; + and the noise actions ``com.netflix.action.DEBUG_PUSH_NOTIFICATION`` / + ``com.netflix.action.DEBUG_FORCE_EXIT``. +* RECONSTRUCTED from the recorded structure (NOT claimed as capture text): the + whole-dumpsys envelope lines (``DUMP OF SERVICE package:``/``activity``/ + ``meminfo``/``procstats``/``usagestats``/``batterystats``), the exit order of + the receiver/service/provider tables, the package block + (``nonLocalizedLabel=null``, ``labelRes=0x7f0f0d31`` — the shapes t13's fixture + already records), and the exact INDENTATION (the capture is not in the repo, so + the nesting depth is the shape ``IntentResolver`` documents). +* THE SIZE. The capture is ~2241 lines; the fixture is 101, so the "only the + section crosses the wire" test PADS the fixture to the observed size with + filler (``_whole_dump``) rather than pretending a 135-line file is a 2241-line + dump. The capture's other 2000+ lines are four service sections and per-package + blocks, none of which the resolver run reads. +* Filler is always labelled, and one test below ("the file is pure capture + text") asserts the fixture holds no prose of ours. + +""" + +from __future__ import annotations + +import asyncio +import shlex +from pathlib import Path +from types import SimpleNamespace + +import pytest +from nsh.connection import ShieldConnectionManager +from nsh.const import CONF_ADB_ENABLED +from nsh.ha_adb_common import ( + INTENT_CANDIDATE_LIMIT, + INTENT_ENTRY_CANDIDATES, + INTENT_ENTRY_EMPTY, + INTENT_SCAN_ADB_DISABLED, + INTENT_SCAN_ADB_UNAVAILABLE, + INTENT_SCAN_CACHED, + INTENT_SCAN_FRESH, + INTENT_SCAN_FRESH_EMPTY, + INTENT_SCAN_INVALID_PACKAGE, + INTENT_SCAN_UNREADABLE, + INTERNAL_MARKERS, + IntentScanCache, + activity_resolver_section, + has_activity_resolver_table, + intent_scan_command, + intent_scan_fallback_command, + is_likely_internal, + parse_activity_resolver_table, + parse_intent_report, +) + +FIXTURE = Path(__file__).resolve().parent / "fixtures" / "pm_dump_netflix_resolver.txt" +NETFLIX = "com.netflix.ninja" +NETFLIX_DUMP = FIXTURE.read_text(encoding="utf-8") + +#: The components the capture's ACTIVITY table names (both real: the app's main +#: activity and its debug receiver). +MAIN_ACTIVITY = "com.netflix.ninja/.MainActivity" +DEBUG_RECEIVER = "com.netflix.ninja/.debug.DebugReceiver" + +#: The five paths on the http/https filter, with the matcher kind the capture +#: prints next to each (a GLOB is not a path — see the candidate docstring). +NETFLIX_PATHS = ( + ("/title.*", "GLOB"), + ("/watch.*", "GLOB"), + ("/browse", "LITERAL"), + ("/home", "LITERAL"), + ("/deeplink.*", "GLOB"), +) + +#: The 18 candidates the capture's activity table yields: 5 paths x 2 schemes for +#: the VIEW filter, the cast/netflix/nflx entries, and the five Non-Data Actions. +EXPECTED_CANDIDATES = 18 + + +def _whole_dump(padding_lines: int = 2200) -> str: + """The fixture padded to the OBSERVED capture size (~2241 lines). + + The filler is explicitly not capture text: it is what the four trailing + service sections and the per-package blocks look like at scale, and it exists + so the transfer test measures the real problem rather than a toy one. Lines + are shaped so a naive parser could pick them up if the on-device filter were + missing (they look like resolver entries for another package). + """ + filler = "\n".join( + f" Package [com.example.filler{n}] (7f0f{n:04d}):\n" + f" versionName=1.0.{n} codePath=/data/app/com.example.filler{n}\n" + f"Activity Resolver Table:\n" + f" Schemes:\n" + f" custom{n}: deca7fa com.example.filler{n}/.OtherActivity " + f"filter fc{n:04d}\n" + f" Action: \"com.example.filler{n}.ACTION\"\n" + for n in range(padding_lines // 6) + ) + return f"{NETFLIX_DUMP}\n{filler}\n" + + +def _on_device_sed_range(dump: str) -> str: + """What the WIRE carries once the device applies the sed range (t16). + + A fake that handed the parser the whole dump would only prove the parser + tolerates it — not that the manager never had to ship it. This emulates the + shipped pipeline: the sed range first, then the ``head -n`` bound. + """ + start = dump.find("Activity Resolver Table:") + assert start >= 0 + end = len(dump) + for heading in ("Receiver Resolver Table:", "Service Resolver Table:"): + found = dump.find(heading, start) + if 0 <= found < end: + end = found + section = dump[start:end] + lines = section.splitlines()[:400] + return "\n".join(lines) + "\n" + + +class _IntentsPoller: + """A poller whose shell answers the resolver commands like the device would. + + ``answers`` maps a substring of the command to the answer, so a test can put + a specific shape on the wire (the section, an error line, nothing at all) and + know which command produced it. Every command is recorded: the cost claims in + this file are assertions on that list, not on intentions. + """ + + def __init__(self, *, answers=None, default=None, raise_on=None, connected=True): + self.answers = dict(answers or {}) + self.default = default + self.raise_on = raise_on + self.connected = connected + self.shell_calls: list[str] = [] + + async def shell(self, command: str): + self.shell_calls.append(command) + for needle, answer in self.answers.items(): + if needle in command: + if isinstance(answer, Exception): + raise answer + return answer + if self.raise_on is not None and self.raise_on in command: + raise RuntimeError("adb shell failed") + return self.default + + @property + def section_commands(self) -> list[str]: + """The commands that actually asked for the resolver section.""" + return [cmd for cmd in self.shell_calls if "Activity Resolver Table" in cmd] + + +def _entry(*, adb_enabled=True): + return SimpleNamespace( + options={CONF_ADB_ENABLED: adb_enabled}, + data={}, + title="Shield", + entry_id="e1", + ) + + +def _manager(poller, *, adb_enabled=True): + return ShieldConnectionManager( + hass=None, entry=_entry(adb_enabled=adb_enabled), poller=poller + ) + + +# ── the fixture itself ───────────────────────────────────────────────────── + + +class TestTheFixtureIsHonest: + """The fixture is a transcription — and it has to LOOK like one.""" + + def test_it_is_a_whole_dumpsys_not_just_the_section(self): + # the shape the ticket records: the package dump, then the four trailing + # service sections. A fixture that held only the resolver table would + # make the on-device filter look pointless. + assert "DUMP OF SERVICE package:" in NETFLIX_DUMP + for service in ("activity", "meminfo", "procstats", "usagestats", "batterystats"): + assert f"DUMP OF SERVICE {service}:" in NETFLIX_DUMP, service + + def test_it_holds_no_prose_of_ours(self): + # the honesty notes live HERE, not in the capture: a fixture with our + # commentary inside it could not be compared with real device output + assert "t16" not in NETFLIX_DUMP + assert "NOTE" not in NETFLIX_DUMP + assert "reconstruct" not in NETFLIX_DUMP.lower() + + def test_the_section_bounds_the_capture_records_are_present(self): + assert "Receiver Resolver Table:" in NETFLIX_DUMP + assert "Service Resolver Table:" in NETFLIX_DUMP + assert NETFLIX_DUMP.count("Activity Resolver Table:") == 2 # + the package block + + def test_the_fixture_stays_a_trimmed_reproduction(self): + # the docstring states the fixture's size; keeping it small is what makes + # the padding note in _whole_dump honest rather than rhetorical + assert 80 < len(NETFLIX_DUMP.splitlines()) < 200 + def test_the_capture_is_far_bigger_than_its_resolver_section(self): + whole = _whole_dump() + filtered = _on_device_sed_range(whole) + + assert len(whole.splitlines()) > 2000, "the observed size is ~2241 lines" + assert len(filtered.splitlines()) < 120, "only the resolver section crosses" + assert "com.example.filler1" not in filtered, "the filter really filters" + + +# ── the parser ──────────────────────────────────────────────────────────── + +@pytest.fixture(scope="module") +def candidates(): + """The capture's candidates, parsed once for the whole module.""" + return parse_activity_resolver_table(NETFLIX_DUMP) + + +class TestTheParserFindsTheCapturedEntries: + """Every entry the ticket names is found, with the fields it carries.""" + + def test_the_cast_entry(self, candidates): + cast = [ + c + for c in candidates + if c.action == "com.google.cast.action.START" and c.scheme == "cast" + ] + assert len(cast) == 1 + assert cast[0].component == MAIN_ACTIVITY + assert cast[0].categories == ("android.intent.category.DEFAULT",) + + def test_the_http_and_https_entries(self, candidates): + for scheme in ("http", "https"): + view = [ + c + for c in candidates + if c.action == "android.intent.action.VIEW" and c.scheme == scheme + ] + assert len(view) == len(NETFLIX_PATHS), scheme + assert {c.authority for c in view} == {"www.netflix.com"} + assert {c.component for c in view} == {MAIN_ACTIVITY} + # the categories the capture prints on both, in capture order + assert view[0].categories == ( + "android.intent.category.DEFAULT", + "android.intent.category.BROWSABLE", + "android.intent.category.LEANBACK_LAUNCHER", + "android.intent.category.LAUNCHER", + ) + + def test_the_path_globs_are_kept_with_their_matcher_kind(self, candidates): + view = [ + c + for c in candidates + if c.action == "android.intent.action.VIEW" and c.scheme == "https" + ] + assert {(c.path, c.path_kind) for c in view} == set(NETFLIX_PATHS) + + def test_the_custom_schemes(self, candidates): + by_scheme = {c.scheme for c in candidates} + assert {"netflix", "nflx"} <= by_scheme + for scheme in ("netflix", "nflx"): + entry = [c for c in candidates if c.scheme == scheme] + assert len(entry) == 1 + assert entry[0].action == "android.intent.action.VIEW" + assert entry[0].authority is None and entry[0].path is None + + def test_the_non_data_actions(self, candidates): + by_action = {c.action for c in candidates if not c.scheme} + assert { + "android.intent.action.MAIN", + "com.netflix.action.NETFLIX_KEY_START", + "com.netflix.action.DIAL_START", + } <= by_action + main = [c for c in candidates if c.action == "android.intent.action.MAIN"] + assert len(main) == 1 + assert main[0].categories == ( + "android.intent.category.LEANBACK_LAUNCHER", + "android.intent.category.LAUNCHER", + ) + assert main[0].scheme is None and main[0].authority is None + + def test_the_authority_keeps_its_name_and_drops_its_weight(self, candidates): + assert "www.netflix.com" in {c.authority for c in candidates} + assert not any( + c.authority and ":" in c.authority for c in candidates + ), "the filter's score (-1) is not part of the authority" + + def test_the_two_components_are_the_captures(self, candidates): + assert {c.component for c in candidates} == {MAIN_ACTIVITY, DEBUG_RECEIVER} + + def test_the_count_is_exactly_the_capture_18(self, candidates): + assert len(candidates) == EXPECTED_CANDIDATES + + def test_deduplication_is_real(self, candidates): + keys = [c.key for c in candidates] + assert len(set(keys)) == len(keys) == EXPECTED_CANDIDATES + + def test_the_receiver_service_and_provider_tables_are_not_read(self, candidates): + text = " ".join(f"{c.component} {c.action}" for c in candidates) + for foreign in ( + "androidx.work.impl.diagnostics.DiagnosticsReceiver", + "com.google.firebase.messaging.FirebaseMessagingService", + "com.google.firebase.iid.FirebaseInstanceIdService", + "androidx.startup.InitializationProvider", + "com.netflix.ninja/.push.DebugPushReceiver", + ): + assert foreign not in text, foreign + + +class TestRankingAndHonesty: + """Order, the cap, and the noise flag — never a silent deletion.""" + + def test_the_order_is_deterministic(self): + first = parse_activity_resolver_table(NETFLIX_DUMP) + second = parse_activity_resolver_table(NETFLIX_DUMP) + + assert [c.key for c in first] == [c.key for c in second] + + def test_noise_is_flagged_and_kept(self): + candidates = parse_activity_resolver_table(NETFLIX_DUMP) + debug = [c for c in candidates if "DEBUG_" in (c.action or "")] + + assert len(debug) == 2, "both DEBUG_* actions from the capture are present" + assert all(c.likely_internal for c in debug), "and both are flagged" + assert candidates[-2:] == sorted( + debug, key=lambda c: c.action or "" + ), "flagged candidates are ranked last, not hidden" + + def test_useful_candidates_come_first(self): + candidates = parse_activity_resolver_table(NETFLIX_DUMP) + + assert candidates[0].action == "android.intent.action.MAIN" + assert not candidates[0].likely_internal + + def test_the_cap_is_applied_after_dedup_and_disclosed(self): + report = parse_intent_report(NETFLIX_DUMP, limit=3) + + assert len(report.candidates) == 3 + assert report.found == EXPECTED_CANDIDATES + assert report.truncated is True + + def test_without_a_cap_nothing_is_truncated(self): + report = parse_intent_report(NETFLIX_DUMP) + + assert report.found == len(report.candidates) == EXPECTED_CANDIDATES + assert report.truncated is False + assert INTENT_CANDIDATE_LIMIT > EXPECTED_CANDIDATES + + def test_the_four_marker_families_are_the_plans(self): + # the ticket's list, pinned so a marker cannot be quietly dropped + assert INTERNAL_MARKERS == ( + "DEBUG_", + "androidx.", + "com.google.firebase.", + "ConstraintProxy", + ) + + def test_the_other_marker_families_are_flagged_not_dropped(self): + """The three families NOT in this capture's activity table. + + On the capture they appear in the receiver/provider tables (which the + parser does not read), so they are proven on a labelled SYNTHETIC section + rather than smuggled into the transcription. + """ + section = ( + "Activity Resolver Table:\n" + " Non-Data Actions:\n" + " androidx.work.DiagnosticsReportReceiver:\n" + " deca7fa com.netflix.ninja/androidx.work.impl.\n" + "DiagnosticsReceiver filter fc483b1\n" + " Action: \"androidx.work.Diagnostics.REPORT\"\n" + " com.google.firebase.MESSAGING_EVENT:\n" + " deca7fa com.netflix.ninja/.PushService filter fc483b2\n" + " Action: \"com.google.firebase.MESSAGING_EVENT\"\n" + " com.netflix.action.NORMAL:\n" + " deca7fa com.netflix.ninja/.MainActivity filter fc483b3\n" + " Action: \"ConstraintProxy.Normal\"\n" + "Service Resolver Table:\n" + ) + candidates = parse_activity_resolver_table(section) + + assert len(candidates) == 3, "nothing was deleted" + assert all(c.likely_internal for c in candidates) + assert is_likely_internal("com.netflix.action.NORMAL", None) is False + assert is_likely_internal(None, "com.x/ConstraintProxy$Foo") is True + + +class TestDegradation: + """A shape we have never seen degrades to NO candidate, never a wrong one.""" + + def test_an_absent_section_is_not_an_answer(self): + for output in (None, "", "Unable to find package: com.netflix.ninja\n", " random"): + assert has_activity_resolver_table(output) is False + assert parse_activity_resolver_table(output) == [] + + def test_a_section_with_no_entries_is_an_answer(self): + empty = ( + "Activity Resolver Table:\n" + "Service Resolver Table:\n" + " Non-Data Actions:\n" + " com.x.ACTION:\n" + " deca7fa com.x/.Y filter fc1\n" + " Action: \"com.x.ACTION\"\n" + ) + # the section IS there and holds nothing: an answer, and the service + # table's filter must not leak into it + assert has_activity_resolver_table(empty) is True + assert parse_activity_resolver_table(empty) == [] + + def test_a_group_key_is_the_action_only_when_it_is_action_shaped(self): + action_keyed = ( + "Activity Resolver Table:\n" + " Non-Data Actions:\n" + " com.netflix.action.KEY_ONLY:\n" + " deca7fa com.netflix.ninja/.MainActivity filter fc1\n" + "Service Resolver Table:\n" + ) + scheme_keyed = ( + "Activity Resolver Table:\n" + " Schemes:\n" + " cast: deca7fa com.netflix.ninja/.MainActivity filter fc2\n" + "Service Resolver Table:\n" + ) + + assert [c.action for c in parse_activity_resolver_table(action_keyed)] == [ + "com.netflix.action.KEY_ONLY" + ] + # "cast" is a scheme key, not an action: no action is invented from it + assert [c.action for c in parse_activity_resolver_table(scheme_keyed)] == [None] + assert [c.scheme for c in parse_activity_resolver_table(scheme_keyed)] == [None] + + def test_a_filter_without_a_component_yields_nothing(self): + noise = ( + "Activity Resolver Table:\n" + " Schemes:\n" + " https:\n" + " Action: \"android.intent.action.VIEW\"\n" + " Scheme: \"acme\"\n" + "Service Resolver Table:\n" + ) + # no entry line at all -> nothing to attach those details to + assert parse_activity_resolver_table(noise) == [] + + def test_the_section_helper_stops_at_the_first_table(self): + section = activity_resolver_section(NETFLIX_DUMP) + + assert section is not None + assert section.startswith("\n Schemes:") + assert "Receiver Resolver Table:" not in section + assert activity_resolver_section("nothing here") is None + + +# ── the command shape and its safety ────────────────────────────────────── + + +class TestTheOnDeviceFilter: + """The dump is filtered on the device — and the command is safe.""" + + def test_the_primary_command_is_a_device_side_range(self): + command = intent_scan_command(NETFLIX) + + assert command.startswith("pm dump ") + assert "sed -n" in command + # the plan's own line, recognisable in a review + assert "/Activity Resolver Table/,/Service Resolver Table/p" in command + assert "head -n 400" in command + assert "head -300" not in command, "the t66 truncation mistake, not repeated" + assert "2>/dev/null" in command + + def test_the_fallback_command_is_a_bounded_grep(self): + command = intent_scan_fallback_command(NETFLIX) + + assert "grep -A 400 -m 1 -F" in command + assert "'Activity Resolver Table:'" in command + assert "sed" not in command + + def test_both_commands_quote_the_package(self): + for command in (intent_scan_command(NETFLIX), intent_scan_fallback_command(NETFLIX)): + assert f"pm dump {shlex.quote(NETFLIX)} " in command + + @pytest.mark.parametrize( + "package", + [ + "com.evil; touch /tmp/owned", + "com.evil && reboot", + "com.evil`id`", + "com.evil$(rm -rf /)", + "com.evil|cat /etc/passwd", + "com.evil'quote", + 'com.evil"quote', + "com.evil\nreboot", + ], + ) + def test_a_hostile_package_cannot_add_a_command(self, package): + for command in (intent_scan_command(package), intent_scan_fallback_command(package)): + tokens = shlex.split(command) + # the shell sees the package as exactly ONE token, verbatim: nothing + # it contains can become a command, an operator or a redirection + assert tokens[2] == package, tokens + assert tokens[0] == "pm" and tokens[1] == "dump" + for forbidden in ("touch", "rm", "reboot", "id", "cat", "bash", "sh"): + assert forbidden not in tokens, (package, forbidden, tokens) + # ...and it only ever appears inside a quoted word of the raw command + # ...and the package is quoted by shlex, byte for byte + assert f"pm dump {shlex.quote(package)} " in command + + def test_the_transfer_is_proportional_to_the_section(self): + whole = _whole_dump() + assert len(whole.splitlines()) > 2000 + + # the emulated device-side filter, then the parser + filtered = _on_device_sed_range(whole) + candidates = parse_activity_resolver_table(filtered) + + assert len(candidates) == EXPECTED_CANDIDATES + assert len(filtered) < len(whole) / 10, "the wire carries the section, not the dump" + + +# ── the cache: an answer versus a failure ───────────────────────────────── + + +class TestTheCacheKeepsTheTwoApart: + """Only a scan that RAN writes an entry — and the entry says which answer.""" + + def test_a_scan_that_found_candidates_says_so(self): + clock = [1000.0] + cache = IntentScanCache(clock=lambda: clock[0]) + candidates = parse_activity_resolver_table(NETFLIX_DUMP) + + entry = cache.put(NETFLIX, candidates) + + assert entry.state == INTENT_ENTRY_CANDIDATES + assert cache.state(NETFLIX) == INTENT_ENTRY_CANDIDATES + assert len(cache.get(NETFLIX).candidates) == EXPECTED_CANDIDATES + + def test_a_scan_that_ran_and_found_nothing_is_an_answer(self): + cache = IntentScanCache() + + entry = cache.put(NETFLIX, []) + + assert entry.state == INTENT_ENTRY_EMPTY + assert entry.candidates == () + assert cache.state(NETFLIX) == INTENT_ENTRY_EMPTY, ( + "an empty answer is cached: the device answered and the answer is 'none'" + ) + + def test_a_failed_scan_is_not_cached_at_all(self): + cache = IntentScanCache() + + # a failure never reaches put() — that is the whole design — so there is + # nothing to read back, not even an empty answer + assert cache.get(NETFLIX) is None + assert cache.state(NETFLIX) is None + + def test_a_stale_entry_is_dropped_rather_than_served(self): + clock = [1000.0] + cache = IntentScanCache(ttl=60.0, clock=lambda: clock[0]) + cache.put(NETFLIX, parse_activity_resolver_table(NETFLIX_DUMP)) + + clock[0] += 61.0 + + assert cache.get(NETFLIX) is None + assert cache.state(NETFLIX) is None + + def test_the_ttl_is_the_module_constant(self): + assert IntentScanCache().ttl > 0 + + def test_the_cap_travels_with_the_entry(self): + clock = [1000.0] + cache = IntentScanCache(clock=lambda: clock[0]) + report = parse_intent_report(NETFLIX_DUMP, limit=2) + + entry = cache.put(NETFLIX, report.candidates, truncated=report.truncated) + + assert entry.truncated is True + assert len(entry.candidates) == 2 + + +# ── the manager: one package, on demand, over ADB ───────────────────────── + + +class TestTheManagerScan: + """The scan is explicit, single, cached, and honest about failure.""" + + async def test_a_scan_returns_the_candidates_and_caches_them(self): + poller = _IntentsPoller(default=NETFLIX_DUMP) + manager = _manager(poller) + + first = await manager.async_scan_intents(NETFLIX) + + assert first.status == INTENT_SCAN_FRESH + assert first.answered is True + assert len(first.candidates) == EXPECTED_CANDIDATES + assert first.entry_state == INTENT_ENTRY_CANDIDATES + assert len(poller.section_commands) == 1, "one shell for one app" + + second = await manager.async_scan_intents(NETFLIX) + + assert second.status == INTENT_SCAN_CACHED + assert second.age_seconds is not None and second.age_seconds >= 0 + assert [c.key for c in second.candidates] == [c.key for c in first.candidates] + assert len(poller.shell_calls) == 1, "the second request cost nothing" + assert manager.intent_scan_state(NETFLIX) == INTENT_ENTRY_CANDIDATES + + async def test_force_rescans(self): + poller = _IntentsPoller(default=NETFLIX_DUMP) + manager = _manager(poller) + await manager.async_scan_intents(NETFLIX) + + again = await manager.async_scan_intents(NETFLIX, force=True) + + assert again.status == INTENT_SCAN_FRESH + assert len(poller.section_commands) == 2 + + async def test_an_empty_answer_is_cached_and_is_not_a_failure(self): + empty = "Activity Resolver Table:\nService Resolver Table:\n" + poller = _IntentsPoller(default=empty) + manager = _manager(poller) + + first = await manager.async_scan_intents(NETFLIX) + + assert first.status == INTENT_SCAN_FRESH_EMPTY + assert first.answered is True, "the device answered: it advertises nothing" + assert first.candidates == () + assert first.entry_state == INTENT_ENTRY_EMPTY + + second = await manager.async_scan_intents(NETFLIX) + + assert second.status == INTENT_SCAN_CACHED + assert second.entry_state == INTENT_ENTRY_EMPTY + assert len(poller.shell_calls) == 1, "a definitive empty answer is not re-asked" + + async def test_a_shell_that_raises_is_not_cached_and_is_retried(self): + poller = _IntentsPoller(raise_on="Activity Resolver Table") + manager = _manager(poller) + + first = await manager.async_scan_intents(NETFLIX) + + assert first.status == INTENT_SCAN_ADB_UNAVAILABLE + assert first.answered is False + assert first.candidates == () + assert manager.intent_scan_state(NETFLIX) is None, "a failure is not an answer" + + second = await manager.async_scan_intents(NETFLIX) + + assert second.status == INTENT_SCAN_ADB_UNAVAILABLE + assert len(poller.shell_calls) > 1, "the next request really retried" + + async def test_no_output_is_not_an_answer(self): + poller = _IntentsPoller(default="") + manager = _manager(poller) + + result = await manager.async_scan_intents(NETFLIX) + + assert result.status == INTENT_SCAN_ADB_UNAVAILABLE + assert result.answered is False + assert manager.intent_scan_state(NETFLIX) is None + + async def test_an_unreadable_answer_falls_back_then_says_so(self): + poller = _IntentsPoller( + answers={ + "sed -n": "Unable to find package: com.netflix.ninja\n", + "grep": "", + } + ) + manager = _manager(poller) + + result = await manager.async_scan_intents(NETFLIX) + + assert result.status == INTENT_SCAN_UNREADABLE + assert result.answered is False + assert result.candidates == () + assert len(poller.shell_calls) == 2, "the bounded grep fallback was tried" + assert "grep" in poller.shell_calls[-1] + assert manager.intent_scan_state(NETFLIX) is None + + async def test_the_fallback_recovers_a_device_without_sed_ranges(self): + poller = _IntentsPoller( + answers={ + "sed -n": "", # a toolbox that cannot do ranges answers nothing + "grep": NETFLIX_DUMP, + } + ) + manager = _manager(poller) + + result = await manager.async_scan_intents(NETFLIX) + + assert result.status == INTENT_SCAN_FRESH + assert len(result.candidates) == EXPECTED_CANDIDATES + assert len(poller.shell_calls) == 2 + assert manager.intent_scan_state(NETFLIX) == INTENT_ENTRY_CANDIDATES + + async def test_adb_disabled_and_unreachable_are_told_apart(self): + disabled = _manager(None, adb_enabled=False) + enabled_but_passive = _manager(None, adb_enabled=True) + disconnected = _manager(_IntentsPoller(default=NETFLIX_DUMP, connected=False)) + disconnected._poller.connected = False + + for manager, expected in ( + (disabled, INTENT_SCAN_ADB_DISABLED), + (enabled_but_passive, INTENT_SCAN_ADB_UNAVAILABLE), + (disconnected, INTENT_SCAN_ADB_UNAVAILABLE), + ): + result = await manager.async_scan_intents(NETFLIX) + assert result.status == expected + assert result.answered is False + assert result.candidates == (), "no empty list pretending there are none" + + assert disconnected._poller.shell_calls == [], "nothing was asked" + + async def test_a_malformed_package_never_reaches_a_shell(self): + poller = _IntentsPoller(default=NETFLIX_DUMP) + manager = _manager(poller) + + for bad in ("", " ", "notapackage", "not a package; rm -rf /"): + result = await manager.async_scan_intents(bad) + assert result.status == INTENT_SCAN_INVALID_PACKAGE + assert result.answered is False + + assert poller.shell_calls == [] + + async def test_one_request_scans_one_package_and_nothing_else(self): + poller = _IntentsPoller( + answers={ + NETFLIX: NETFLIX_DUMP, + "com.other.app": NETFLIX_DUMP, + } + ) + manager = _manager(poller) + + await manager.async_scan_intents(NETFLIX) + + assert len(poller.shell_calls) == 1 + assert NETFLIX in poller.shell_calls[0] + assert "com.other" not in poller.shell_calls[0] + + await manager.async_scan_intents("com.other.app") + + assert len(poller.shell_calls) == 2, "the second package is its own scan" + assert manager.intent_scan_state(NETFLIX) == INTENT_ENTRY_CANDIDATES + assert manager.intent_scan_state("com.other.app") == INTENT_ENTRY_CANDIDATES + + async def test_concurrent_requests_for_one_package_share_a_dump(self): + release = asyncio.Event() + + class _SlowPoller(_IntentsPoller): + async def shell(self, command): + await release.wait() + return await super().shell(command) + + poller = _SlowPoller(default=NETFLIX_DUMP) + manager = _manager(poller) + + first = asyncio.create_task(manager.async_scan_intents(NETFLIX)) + await asyncio.sleep(0) # let the first scan reach the shell + second = asyncio.create_task(manager.async_scan_intents(NETFLIX)) + await asyncio.sleep(0) + release.set() + one, two = await asyncio.gather(first, second) + + assert len(poller.section_commands) == 1, "one dump, two callers" + assert [c.key for c in one.candidates] == [c.key for c in two.candidates] + + async def test_the_watchdog_never_scans_intents(self): + """Discovery is EXPLICIT: no cycle of the manager may start a scan.""" + poller = _IntentsPoller(default=NETFLIX_DUMP) + manager = _manager(poller) + + await manager._watchdog_cycle(0.0) + + assert poller.section_commands == [], "the watchdog must not scan" + assert manager.intent_scan_state(NETFLIX) is None + + async def test_the_result_serialises_for_the_ui(self): + poller = _IntentsPoller(default=NETFLIX_DUMP) + manager = _manager(poller) + + payload = (await manager.async_scan_intents(NETFLIX)).as_dict() + + assert payload["status"] == INTENT_SCAN_FRESH + assert payload["answered"] is True + assert payload["truncated"] is False + assert len(payload["candidates"]) == EXPECTED_CANDIDATES + assert payload["candidates"][0]["component"] == MAIN_ACTIVITY + + async def test_the_ttl_expires_in_the_manager_too(self): + """The manager's cache is the module's, on the manager's clock (t16).""" + now = [1000.0] + poller = _IntentsPoller(default=NETFLIX_DUMP) + manager = ShieldConnectionManager( + hass=None, + entry=_entry(), + poller=poller, + clock=lambda: now[0], + ) + + await manager.async_scan_intents(NETFLIX) + cached = await manager.async_scan_intents(NETFLIX) + assert cached.status == INTENT_SCAN_CACHED + + now[0] += manager.intent_cache.ttl + 1.0 + + assert manager.intent_scan_state(NETFLIX) is None, "a stale answer is dropped" + again = await manager.async_scan_intents(NETFLIX) + assert again.status == INTENT_SCAN_FRESH + assert len(poller.section_commands) == 2, "and the device is asked again" + + async def test_an_enormous_table_is_capped_and_says_so(self): + """The cap is disclosed end to end, never a silently shortened answer.""" + entries = "".join( + f" com.example.action{n:03d}:\n" + f" deca7fa com.example.app/.Activity{n:03d} filter fc{n:04d}\n" + f" Action: \"com.example.action{n:03d}\"\n" + for n in range(INTENT_CANDIDATE_LIMIT + 25) + ) + section = ( + "Activity Resolver Table:\n" + " Non-Data Actions:\n" + f"{entries}" + "Service Resolver Table:\n" + ) + poller = _IntentsPoller(default=section) + manager = _manager(poller) + + result = await manager.async_scan_intents(NETFLIX) + + assert result.status == INTENT_SCAN_FRESH + assert result.answered is True + assert len(result.candidates) == INTENT_CANDIDATE_LIMIT + assert result.truncated is True, "a capped answer never looks complete" + assert manager.intent_cache.get(NETFLIX).truncated is True