ha-nvidia-shield/tests/test_intents.py

"""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"
        )