"""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 TestATruncatedValueIsNotFabricated: """t18/F2: a capture cut mid-value must not yield a PARTIAL value. Android prints quoted detail values (``Scheme: "https"``, ``Authority: "www.netflix.com": -1``, ``Path: "PatternMatcher{GLOB: /title.*}"``). A dump truncated inside one of them used to produce ``"ht`` / ``"www.exam`` — values that LOOK real, which ``_candidate_data`` would then turn into a URL the app cannot match. Not an injection (everything stays quoted when used), which is exactly why nothing else caught it: the corrupt value must be distinguishable from a real one, so an unterminated quoted value is DROPPED. """ #: Every quoted detail cut before its closing quote — the reviewer's cases, #: plus a cut in the middle of the Path matcher. TRUNCATED = ( "Activity Resolver Table:\n" " Schemes:\n" " https:\n" " deca7fa com.netflix.ninja/.MainActivity filter fc1\n" ' Action: "android.intent.action.VIEW"\n' ' Scheme: "ht\n' ' Authority: "www.exam\n' ' Path: "PatternMatcher{GLOB: /title\n' "Service Resolver Table:\n" ) def test_the_partial_values_are_dropped_not_kept(self): [candidate] = parse_activity_resolver_table(self.TRUNCATED) assert candidate.scheme is None, "a half scheme is not a scheme" assert candidate.authority is None, "a half host is not a host" assert candidate.path is None, "a half pattern is not a path" assert candidate.path_kind is None def test_the_intact_details_survive_and_nothing_carries_a_quote(self): [candidate] = parse_activity_resolver_table(self.TRUNCATED) assert candidate.action == "android.intent.action.VIEW", ( "the value that WAS complete is still read" ) assert candidate.component == "com.netflix.ninja/.MainActivity" for value in (candidate.scheme, candidate.authority, candidate.path): assert value is None or not value.startswith(("'", '"')), value def test_an_unterminated_value_at_the_very_end_of_the_input(self): """The last line can be the cut one — the end of input is not a terminator.""" cut = ( "Activity Resolver Table:\n" " Schemes:\n" " https:\n" " deca7fa com.netflix.ninja/.MainActivity filter fc1\n" ' Scheme: "ht' ) [candidate] = parse_activity_resolver_table(cut) assert candidate.scheme is None assert candidate.authority is None assert candidate.path is None def test_a_lone_quote_is_dropped_too(self): cut = ( "Activity Resolver Table:\n" " Schemes:\n" " https:\n" " deca7fa com.netflix.ninja/.MainActivity filter fc1\n" ' Scheme: "\n' ) [candidate] = parse_activity_resolver_table(cut) assert candidate.scheme is None def test_a_complete_quoted_value_is_still_read(self): """The anti-overcorrection pin: dropping must not eat good values.""" intact = ( "Activity Resolver Table:\n" " Schemes:\n" " https:\n" " deca7fa com.netflix.ninja/.MainActivity filter fc1\n" ' Action: "android.intent.action.VIEW"\n' ' Scheme: "https"\n' ' Authority: "www.netflix.com": -1\n' ' Path: "PatternMatcher{LITERAL, /browse}"\n' "Service Resolver Table:\n" ) [candidate] = parse_activity_resolver_table(intact) assert candidate.scheme == "https" assert candidate.authority == "www.netflix.com" assert candidate.path == "/browse" assert candidate.path_kind == "LITERAL" 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 class TestAdbDisabledNeverTouchesTheDevice: """t18/F1: an entry whose OPTION says ADB is off must not scan at all. The review measured the defect: with ``adb_enabled: False`` and a CONNECTED poller, the scan fell straight through to the shell loop — ``pm dump com.x …`` was really sent, one candidate was CACHED, and the status read ``fresh`` (a silent poller made it ``adb_unavailable``, so even the label depended on the wrong thing). "ADB is switched off" is a fact about the ENTRY, so the option is checked once at the top and returns before any shell or cache touch. """ async def test_a_held_poller_does_not_override_the_option(self): poller = _IntentsPoller(default=NETFLIX_DUMP) # connected, would answer manager = _manager(poller, adb_enabled=False) result = await manager.async_scan_intents(NETFLIX) assert result.status == INTENT_SCAN_ADB_DISABLED assert result.answered is False assert result.candidates == () assert poller.shell_calls == [], "the device was never touched" assert manager.intent_scan_state(NETFLIX) is None, "nothing was cached" async def test_a_forced_scan_is_refused_the_same_way(self): poller = _IntentsPoller(default=NETFLIX_DUMP) manager = _manager(poller, adb_enabled=False) assert (await manager.async_scan_intents(NETFLIX)).status == ( INTENT_SCAN_ADB_DISABLED ) forced = await manager.async_scan_intents(NETFLIX, force=True) assert forced.status == INTENT_SCAN_ADB_DISABLED assert poller.shell_calls == [], "force does not override the option either" assert manager.intent_scan_state(NETFLIX) is None async def test_a_silent_poller_cannot_change_the_status(self): """The review's second half: the LABEL must come from the option.""" silent_but_disabled = _manager(_IntentsPoller(default=None), adb_enabled=False) assert (await silent_but_disabled.async_scan_intents(NETFLIX)).status == ( INTENT_SCAN_ADB_DISABLED ) assert silent_but_disabled._poller.shell_calls == [] # and the mirror image: ADB ON with a silent poller is "could not ask", # never "switched off" — the two have different fixes silent_but_enabled = _manager(_IntentsPoller(default=None), adb_enabled=True) assert (await silent_but_enabled.async_scan_intents(NETFLIX)).status == ( INTENT_SCAN_ADB_UNAVAILABLE ) async def test_a_cached_answer_cannot_leak_out_of_a_disabled_entry(self): """ADB off is answered BEFORE the cache is read — no stale prefill.""" poller = _IntentsPoller(default=NETFLIX_DUMP) manager = _manager(poller) first = await manager.async_scan_intents(NETFLIX) assert first.status == INTENT_SCAN_FRESH and first.candidates manager.entry.options[CONF_ADB_ENABLED] = False # the user turns ADB off after = await manager.async_scan_intents(NETFLIX) assert after.status == INTENT_SCAN_ADB_DISABLED assert after.candidates == (), "a disabled entry offers no scan answer at all" assert len(poller.shell_calls) == 1, "only the original, permitted scan ran" async def test_the_no_poller_control_keeps_its_own_status(self): manager = _manager(None, adb_enabled=True) result = await manager.async_scan_intents(NETFLIX) assert result.status == INTENT_SCAN_ADB_UNAVAILABLE, ( "no poller is not the same fact as the option being off" )