Not-Expected-Yet/ha-nvidia-shield
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

feat(intents): discover an app's intents from its activity resolver table (t16)

Browse Source 
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 <pkg> 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).
This commit is contained in:
James Mackay committed 2026-10-12 12:27:01 +01:00
1 parent 14c168c25e
commit 91d9fb3883
6 files changed
+1869 -5
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

custom_components/nvidia_shield/connection.py
+167
View File
@@ -45,6 +45,7 @@ from .const import (
APP_LIST_TTL, APP_LIST_TTL,
APP_SWEEP_CONCURRENCY, APP_SWEEP_CONCURRENCY,
APP_SWEEP_SPACING, APP_SWEEP_SPACING,
CONF_ADB_ENABLED,
CONF_APP_NAME_MAP, CONF_APP_NAME_MAP,
CONF_LOGCAT_EVENTS, CONF_LOGCAT_EVENTS,
DEFAULT_APP_DIRECTORY, DEFAULT_APP_DIRECTORY,
@@ -59,13 +60,26 @@ from .const import (
SYSTEM_MEDIA_PACKAGES, SYSTEM_MEDIA_PACKAGES,
) )
from .ha_adb_common import ( 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, MediaSessionInfo,
foreground_from_logcat, foreground_from_logcat,
has_activity_resolver_table,
icon_for_package, icon_for_package,
installed_packages, installed_packages,
intent_scan_command,
intent_scan_fallback_command,
is_launchable, is_launchable,
parse_app_label, parse_app_label,
parse_audio_playback, parse_audio_playback,
parse_intent_report,
resolve_package_name, resolve_package_name,
) )
@@ -292,6 +306,16 @@ class ShieldConnectionManager:
self._logcat_ladder_sleep = asyncio.sleep self._logcat_ladder_sleep = asyncio.sleep
# Injectable clock keeps the TTL expiry tests deterministic. # Injectable clock keeps the TTL expiry tests deterministic.
self._clock = clock 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._tasks: list[asyncio.Task] = []
self._stopping = False self._stopping = False
self._failures = 0 self._failures = 0
@@ -1503,6 +1527,149 @@ class ShieldConnectionManager:
self._create_task(self._fetch_app_label(package), f"app_label_{package}") 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]: def last_known_map(self) -> dict[str, MediaSessionInfo]:
"""TTL-filtered snapshot of the per-package last-known cache.""" """TTL-filtered snapshot of the per-package last-known cache."""
result: dict[str, MediaSessionInfo] = {} result: dict[str, MediaSessionInfo] = {}
custom_components/nvidia_shield/ha_adb_common/__init__.py
+46
View File
@@ -24,6 +24,30 @@ from .installed_apps import (
parse_package_list, parse_package_list,
) )
from .intent import PlaybackIntent 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 .keys import ensure_adb_key
from .logcat_events import foreground_from_logcat from .logcat_events import foreground_from_logcat
from .media_session import ( from .media_session import (
@@ -41,8 +65,30 @@ __all__ = [
"ATTR_MEDIA_DAG_ROLE", "ATTR_MEDIA_DAG_ROLE",
"ATTR_MEDIA_DAG_RUNNING_APPS", "ATTR_MEDIA_DAG_RUNNING_APPS",
"AdbClientBase", "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", "MediaSessionInfo",
"PlaybackIntent", "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", "ensure_adb_key",
"parse_all_media_sessions", "parse_all_media_sessions",
"parse_app_label", "parse_app_label",
custom_components/nvidia_shield/ha_adb_common/intents.py
+692
View File
@@ -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 <package>``.
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 <pkg> 2>/dev/null \
| sed -n "/Activity Resolver Table/,/Service Resolver Table/p" \
| head -n <cap>
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<key>.*?)\s*(?P<hex_id>[0-9a-fA-F]{6,})\s+"
r"(?P<component>[A-Za-z_][\w.]*/[\w.$]+)"
r"(?:\s+filter\s+(?P<filter_id>[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<key>[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*(?P<field>Action|Category|Scheme|Authority|Path)\s*:\s*(?P<value>.+?)\s*$"
)
#: ``Authority: "www.netflix.com": -1`` — the authority, not its weight.
_AUTHORITY_RE = re.compile(r'^"(?P<value>[^"]*)"\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<kind>[A-Za-z_]+)\s*[:,]?\s*(?P<value>.*)\}$"
)
#: 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 <pkg>`` 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 <pkg> 2>/dev/null | grep -A <cap> -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],
}
tests/fixtures/pm_dump_netflix_resolver.txt
Vendored
+101
View File
@@ -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:
tests/ha_api_drift.py
+13 -5
View File
@@ -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 #: 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, #: and the pytest wrapper (t39 F1). They sit below the current walk (26 modules,
#: 76 imports, 206 relative imports, 57 checked arguments, 6 awaited HA calls, #: 76 imports, 242 relative imports, 57 checked arguments, 6 awaited HA calls,
#: 1322 statement blocks) so a broken or mis-targeted walk fails loudly instead of #: 1417 statement blocks) so a broken or mis-targeted walk fails loudly instead of
#: printing zeros and exiting 0. #: printing zeros and exiting 0.
FLOORS: dict[str, int] = { FLOORS: dict[str, int] = {
# t41: the sweep is recursive, so the floor sits above the top-level-only # 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 # `ATTR_ACTIVITY`, the transport-state accessors and the shared app-directory
# helpers all arrive as imports, and none of them may vanish unnoticed. # helpers all arrive as imports, and none of them may vanish unnoticed.
"imports": 70, "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 # 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 # 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 # `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. # 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 # 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. # 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 #: The interpreter floor Home Assistant itself declares: PyPI metadata for
tests/test_intents.py
+850
View File
@@ -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