"""t57: installed-app discovery — parsers, the bounded sweep, and the surfaces.

No real Shield was available, so the command OUTPUT here is the part that is
fixture-based, not observed: the shapes are the documented Android ones
(``pm list packages`` prints ``package:<id>``; ``resolve-activity --brief``
prints the verbose intent then ``package/activity``; ``pm dump`` prints
``application-label: '...'``). The parsers are deliberately tolerant because
that is the property that matters when a real device's output differs in
whitespace or ordering, and every tolerance is pinned below.
"""

from __future__ import annotations

import asyncio
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import entity_stublib  # noqa: E402
import pytest  # noqa: E402
from entity_stublib import load_entity_modules  # noqa: E402


@pytest.fixture(autouse=True)
def scoped_entity_stubs():
    """Container for the HA stubs + loaded entity modules (t7 discipline).

    Without the teardown the stubs this file installs leak into whichever test
    file runs next — tests/test_harness_scoping.py exists because that mistake
    is invisible until an unrelated file fails.
    """
    snapshot = entity_stublib.snapshot_modules()
    entity_stublib.install()
    yield load_entity_modules()
    entity_stublib.restore_modules(snapshot)

from nsh.connection import ShieldConnectionManager  # noqa: E402
from nsh.const import (  # noqa: E402
    APP_ICON_FALLBACKS,
    APP_LABEL_EXIT_MARKER,
    APP_LABEL_GREP_SHAPES,
    APP_LIST_TTL,
    CONF_APP_NAME_MAP,
)
from nsh.ha_adb_common import (  # noqa: E402
    icon_for_package,
    installed_packages,
    is_launchable,
    parse_active_launcher,
    parse_package_list,
)

# ── realistic device output ────────────────────────────────────────────────

#: `pm list packages -3` on the user's Shield (t66). These are the apps the
#: report listed, which is also why they are the curated table's rows: on THIS
#: device the firmware names none of them, so the table — not the device — is where
#: the names come from. `org.xbmc.kodi` and `com.example.handmade` are kept as the
#: packages NOTHING can name (no device label, no curated row), so the fall-through
#: to "" is pinned on the same list.
USER_PACKAGES = (
    "package:ar.tvplayer.tv\n"
    "package:be.digitalia.mediasession2mqtt\n"
    "package:com.amazon.amazonvideo.livingroom\n"
    "package:com.esaba.downloader\n"
    "package:com.github.damontecres.wholphin\n"
    "package:com.google.android.sambadocumentsprovider\n"
    "package:com.google.android.speech.pumpkin\n"
    "package:com.google.android.tts\n"
    "package:com.google.android.tvlauncher\n"
    "package:com.google.android.youtube.tv\n"
    "package:com.hyperion.grabber\n"
    "package:com.lonelycatgames.Xplore\n"
    "package:com.nendo.argosy\n"
    "package:com.netflix.ninja\n"
    "package:com.plexapp.mediaserver.smb\n"
    "package:com.retroarch\n"
    "package:com.spocky.projengmenu\n"
    "package:dev.jdtech.jellyfin\n"
    "package:dev.trooped.tvquickbars\n"
    "package:flar2.homebutton\n"
    "package:tv.pluto.android\n"
    "package:org.xbmc.kodi\n"
    "package:com.example.handmade\n"
)

#: `cmd package resolve-activity --brief -c android.intent.category.HOME` —
#: the verbose header then the component, exactly as Android prints it.
LAUNCHER_RESOLVED = (
    "priority=0 preferredOrder=0 match=0x108000 specificIndex=-1 isDefault=false\n"
    "com.google.android.tvlauncher/com.google.android.apps.tv.launcherx.MainActivity\n"
)

#: `pm dump` for an app whose manifest and locale variants both exist. The
#: parser takes the first default-locale label and ignores the localized ones.
DUMP_TIVIMATE = (
    "Package [ar.tvplayer.tv] (7f1a2b3c):\n"
    "  application-label:'TiviMate'\n"
    "  application-label-de:'TiviMate DE'\n"
    "  application-label-fr:'TiviMate FR'\n"
    "  versionName=4.7.0\n"
)

#: A dump that RAN and carries no label at all — several system/OEM packages
#: look like this. It is a definitive negative, not a failed fetch.
DUMP_LABEL_LESS = (
    "Package [com.example.handmade] (aa11bb22):\n"
    "  versionName=1.0\n"
    "  codePath=/data/app/com.example.handmade\n"
)

#: The REAL device shape, trimmed: `pm dump com.netflix.ninja` on the user's Shield
#: (captured 2026-10-12; the observed structure is recorded in
#: ``docs/INTENTS_PLAN.md`` §1). It is the evidence behind this work — the raw
#: ~2241-line capture text is NOT in the repository, so this is a faithful
#: reproduction of the recorded structure, not the capture itself.
#:
#: WHAT IS REPRODUCED: the whole-dumpsys envelope (`DUMP OF SERVICE package:`,
#: then `activity`, `meminfo`, `procstats`, `usagestats`, `batterystats`), the
#: package section opening with `Activity Resolver Table:` followed by the
#: receiver/service tables, Netflix's real entries (`cast:`, `http`/`https` with
#: `www.netflix.com` path globs, `nflx:`, `netflix:`, `MAIN` +
#: `LEANBACK_LAUNCHER`) and the capture's own noise (`com.netflix.action.DEBUG_*`,
#: `androidx.*`, `com.google.firebase.*`, `ConstraintProxy*`) — and, the point of
#: the whole exercise, NO `application-label` line anywhere: grepping the capture
#: for `application-label` found nothing.
#:
#: (`nonLocalizedLabel=null`, `labelRes=0x7f0f0d31`) — the capture's record covers
#: the resolver tables, not this block. They are here because they are exactly the
#: shapes the on-device search must NOT read as a name: `null` is not a name, and a
#: resource id is not a name either. The bodies of the four trailing service
#: sections are trimmed to one line each (the capture is ~2241 lines; this is ~80).
DUMP_NETFLIX_REAL = (
    "DUMP OF SERVICE package:\n"
    "Activity Resolver Table:\n"
    "  Full MIME Types:\n"
    "    video/*: 7f0f4c12 com.netflix.ninja/.ui.MainActivity filter 7f0f0a11\n"
    "      Action: \"android.intent.action.VIEW\"\n"
    "      Category: \"android.intent.category.DEFAULT\"\n"
    "      Category: \"android.intent.category.BROWSABLE\"\n"
    "      Scheme: \"https\"\n"
    "        Authority: \"www.netflix.com\": -1\n"
    "        Path: \"PatternMatcher{LITERAL,/title}\"\n"
    "      Scheme: \"http\"\n"
    "        Authority: \"www.netflix.com\": -1\n"
    "    nflx: 7f0f4c12 com.netflix.ninja/.ui.MainActivity filter 7f0f0a12\n"
    "      Action: \"android.intent.action.VIEW\"\n"
    "      Category: \"android.intent.category.DEFAULT\"\n"
    "      Scheme: \"nflx\"\n"
    "    netflix: 7f0f4c12 com.netflix.ninja/.ui.MainActivity filter 7f0f0a13\n"
    "      Action: \"android.intent.action.VIEW\"\n"
    "      Scheme: \"netflix\"\n"
    "  Non-Data Actions:\n"
    "    android.intent.action.MAIN:\n"
    "      7f0f4c12 com.netflix.ninja/.ui.MainActivity filter 7f0f0a14\n"
    "        Action: \"android.intent.action.MAIN\"\n"
    "        Category: \"android.intent.category.LEANBACK_LAUNCHER\"\n"
    "    cast:\n"
    "      7f0f4c12 com.netflix.ninja/.ui.MainActivity filter 7f0f0a15\n"
    "        Action: \"android.intent.action.VIEW\"\n"
    "        Scheme: \"cast\"\n"
    "    com.netflix.action.DEBUG_FORCE_EXIT:\n"
    "      7f0f4c12 com.netflix.ninja/.DebugReceiver filter 7f0f0a16\n"
    "        Action: \"com.netflix.action.DEBUG_FORCE_EXIT\"\n"
    "Receiver Resolver Table:\n"
    "  Non-Data Actions:\n"
    "    androidx.work.DiagnosticsReportReceiver:\n"
    "      7f0f4c12 com.netflix.ninja/androidx.work.impl.diagnostics.DiagnosticsReceiver\n"
    "    com.google.firebase.MESSAGING_EVENT:\n"
    "      7f0f4c12 com.netflix.ninja/com.google.firebase.messaging.FirebaseMessagingService\n"
    "    com.netflix.action.DEBUG_PUSH_NOTIFICATION:\n"
    "      7f0f4c12 com.netflix.ninja/.push.DebugPushReceiver\n"
    "Service Resolver Table:\n"
    "  Non-Data Actions:\n"
    "    com.google.firebase.INSTANCE_ID_EVENT:\n"
    "      7f0f4c12 com.netflix.ninja/com.google.firebase.iid.FirebaseInstanceIdService\n"
    "Provider Resolver Table:\n"
    "  Authorities:\n"
    "    com.netflix.ninja.androidx-startup:\n"
    "      7f0f4c12 com.netflix.ninja/androidx.startup.InitializationProvider\n"
    "Permissions:\n"
    "  Permission [com.netflix.ninja.C2D_MESSAGE]:\n"
    "    sourcePackage=com.netflix.ninja\n"
    "Packages:\n"
    "  Package [com.netflix.ninja] (7f1a2b3c):\n"
    "    userId=10123\n"
    "    pkg=Package{7f1a2b3c com.netflix.ninja}\n"
    # the two INFERRED lines (see the note above): the shapes the search must skip
    "    nonLocalizedLabel=null\n"
    "    labelRes=0x7f0f0d31\n"
    "    versionCode=61102 minSdk=21 targetSdk=29\n"
    "    versionName=6.11.2 build 3413\n"
    "    codePath=/data/app/com.netflix.ninja-1/base.apk\n"
    "    ConstraintProxy.BatteryBroadcastReceiver=\n"
    "      com.netflix.ninja/androidx.work.impl.constraints.trackers."
    "ConstraintProxy$BatteryChargingProxy\n"
    "  Activity Resolver Table: (nothing else listed for this package)\n"
    "DUMP OF SERVICE activity:\n"
    "  ACTIVITY MANAGER PENDING INTENTS (dumpsys activity pending)\n"
    "    (nothing)\n"
    "DUMP OF SERVICE meminfo:\n"
    "  Applications Memory Usage (in Kilobytes):\n"
    "DUMP OF SERVICE procstats:\n"
    "  Process Stats:\n"
    "DUMP OF SERVICE usagestats:\n"
    "  In-memory rollup:\n"
    "DUMP OF SERVICE batterystats:\n"
    "  Battery Stats:\n"
)


def _on_device_grep(dump: str) -> str:
    """What the WIRE carries once the fetch greps on the device (t66).

    The real command is ``pm dump <pkg> | grep -E <shapes> | head -40``, so the
    2241-line dump never crosses ADB: only the matching lines do, plus the exit
    marker the manager relies on. A fake that returned the whole dump would only
    prove that the PARSER tolerates it — not that the manager never had to.
    """
    shapes = APP_LABEL_GREP_SHAPES.split("|")
    hits = [line for line in dump.splitlines() if any(s in line for s in shapes)]
    return "\n".join(hits[:40]) + f"\n{APP_LABEL_EXIT_MARKER}0\n"


class _DiscoveryPoller:
    """A poller whose shell answers the discovery commands from fixtures."""

    def __init__(self, *, packages=USER_PACKAGES, launcher=LAUNCHER_RESOLVED,
                 dumps=None, failure=False, on_device_grep=False):
        self.connected = True
        self.connected_override = True
        self.packages = packages
        self.launcher = launcher
        self.dumps = dict(dumps or {})
        self.failure = failure
        #: t66: True = the fixture values are RAW device dumps and the fake applies
        #: the on-device grep before answering, like the real command does.
        self.on_device_grep = on_device_grep
        self.shell_calls: list[str] = []

    @property
    def is_connected(self):
        return self.connected

    async def shell(self, cmd: str):
        self.shell_calls.append(cmd)
        if self.failure:
            return None
        if "pm list packages" in cmd:
            return self.packages
        if "resolve-activity" in cmd:
            return self.launcher
        if "pm dump" in cmd:
            import shlex

            for package, dump in self.dumps.items():
                if shlex.quote(package) in cmd or package in cmd:
                    return _on_device_grep(dump) if self.on_device_grep else dump
            return "__NSH_EXIT:0\n"  # ran, no label
        return None


def _manager(poller):
    return ShieldConnectionManager(hass=None, entry=None, poller=poller)


class TestPackageListParsing:
    """`pm list packages -3`."""

    def test_real_output_parses_one_package_per_line(self):
        # the order of `pm list packages -3` is preserved, one id per line
        assert parse_package_list(USER_PACKAGES) == [
            line.removeprefix("package:") for line in USER_PACKAGES.splitlines()
        ]

    def test_noise_and_duplicates_are_tolerated(self):
        noisy = (
            "adb: connecting to 10.0.0.9:5555\n"
            "package:com.a\n"
            "\n"
            "package:com.a\n"
            "package:com.b   \n"
            "some warning line\n"
        )
        assert parse_package_list(noisy) == ["com.a", "com.b"]

    def test_empty_and_none_are_an_empty_list(self):
        assert parse_package_list(None) == []
        assert parse_package_list("") == []

    def test_a_package_id_is_sanity_checked(self):
        assert is_launchable("com.example.app")
        assert is_launchable("ar.tvplayer.tv")
        assert not is_launchable("")
        assert not is_launchable("notapackage")
        assert not is_launchable("com..b")


class TestLauncherParsing:
    """`cmd package resolve-activity --brief -c ...HOME`."""

    def test_the_component_after_the_verbose_header_is_the_launcher(self):
        assert parse_active_launcher(LAUNCHER_RESOLVED) == "com.google.android.tvlauncher"

    def test_activity_only_output_still_parses(self):
        assert parse_active_launcher("com.google.android.tvlauncher/.MainActivity") == (
            "com.google.android.tvlauncher"
        )

    def test_nothing_resolved_is_none(self):
        assert parse_active_launcher("No activity found") is None
        assert parse_active_launcher(None) is None

    def test_the_tv_settings_entry_point_is_not_a_launcher(self):
        # a HOME resolution can land on the settings app's entry point; that is
        # not the app a user opens
        output = "com.android.tv.settings/.MainActivity"
        assert parse_active_launcher(output) is None


class TestInstalledPackagesMerge:
    """User apps plus the active launcher, de-duplicated."""

    def test_launcher_is_added_last(self):
        # t66: the user's own list contains the launcher, so the append-last
        # property is pinned on a list that does NOT already have it
        user_only = USER_PACKAGES.replace("package:com.google.android.tvlauncher\n", "")
        packages = installed_packages(user_only, LAUNCHER_RESOLVED)
        assert packages[-1] == "com.google.android.tvlauncher"
        assert packages.count("com.google.android.tvlauncher") == 1
        assert packages[0] == "ar.tvplayer.tv"

    def test_a_launcher_already_listed_keeps_its_place(self):
        packages = installed_packages(USER_PACKAGES, LAUNCHER_RESOLVED)
        assert packages.count("com.google.android.tvlauncher") == 1
        assert packages == parse_package_list(USER_PACKAGES)

    def test_a_third_party_launcher_is_not_duplicated(self):
        user = "package:com.example.launcher\npackage:com.a\n"
        launcher = "com.example.launcher/.MainActivity"
        assert installed_packages(user, launcher) == ["com.example.launcher", "com.a"]

    def test_no_launcher_still_yields_the_user_apps(self):
        assert installed_packages(USER_PACKAGES, None) == parse_package_list(USER_PACKAGES)


class TestIconHeuristic:
    """Icons are a curated map plus a generic fallback — and always a string."""

    def test_a_known_package_gets_its_curated_icon(self):
        assert icon_for_package("com.netflix.ninja", APP_ICON_FALLBACKS) == "mdi:netflix"

    def test_an_unknown_package_gets_the_generic_icon(self):
        assert icon_for_package("com.example.handmade", APP_ICON_FALLBACKS) == "mdi:television"

    def test_the_result_is_always_a_string(self):
        for package in ("", None, "com.example.handmade", "ar.tvplayer.tv"):
            icon = icon_for_package(package, APP_ICON_FALLBACKS)
            assert isinstance(icon, str) and icon


class TestDiscoverySweep:
    """The bounded sweep on the connection manager."""

    async def test_a_sweep_lists_labels_and_icons_every_app(self):
        poller = _DiscoveryPoller(dumps={
            "ar.tvplayer.tv": DUMP_TIVIMATE,
            "com.example.handmade": DUMP_LABEL_LESS,
        })
        manager = _manager(poller)

        rows = await manager.async_refresh_installed_apps()

        packages = [row["package"] for row in rows]
        assert "ar.tvplayer.tv" in packages
        assert "com.netflix.ninja" in packages
        assert "com.google.android.tvlauncher" in packages  # the launcher is there
        by_pkg = {row["package"]: row for row in rows}
        assert by_pkg["ar.tvplayer.tv"]["name"] == "TiviMate"  # from the dump
        assert by_pkg["ar.tvplayer.tv"]["icon"] == "mdi:television-classic"
        # t66: nothing can name `com.example.handmade` (the dump carries no label
        # and it is not in the curated table), so its name is "" — the id is
        # already published once in `package`, and a name that repeats it is the
        # `pkg=pkg` row the user reported from the apps editor. The row itself is
        # still offered: it must not vanish from the table.
        assert by_pkg["com.example.handmade"]["name"] == ""
        assert by_pkg["com.example.handmade"]["package"] == "com.example.handmade"
        # every row has a string name ("" when nothing names it) and a real icon
        for row in rows:
            assert isinstance(row["name"], str)
            assert isinstance(row["icon"], str) and row["icon"]

    async def test_the_live_bug_the_discovered_list_is_a_superset(self):
        """The complaint in one assertion: every app the user has is offered, and
        each one is NAMED — on this device none of them is named by the firmware.
        """
        from nsh.const import DEFAULT_APP_DIRECTORY

        poller = _DiscoveryPoller(dumps={"ar.tvplayer.tv": DUMP_TIVIMATE})
        manager = _manager(poller)

        rows = await manager.async_refresh_installed_apps()
        packages = {row["package"] for row in rows}
        by_pkg = {row["package"]: row for row in rows}

        # the curated table is exactly the app set the report listed
        assert set(DEFAULT_APP_DIRECTORY) <= packages
        # ...and every one of those apps reads by its NAME, from the curated map,
        # with the device naming nothing (t66)
        for package, name in DEFAULT_APP_DIRECTORY.items():
            assert name
            assert by_pkg[package]["name"], package
            if package != "ar.tvplayer.tv":  # this one's dump DOES carry a label
                assert by_pkg[package]["name"] == name, package
        assert by_pkg["ar.tvplayer.tv"]["name"] == "TiviMate"  # the device wins here
        assert by_pkg["org.xbmc.kodi"]["name"] == ""  # known to nobody
        assert "com.google.android.tvlauncher" in packages  # + the launcher
        # a system app the device never lists as -3 is NOT invented here
        assert "com.android.systemui" not in packages

    async def test_the_sweep_issues_the_documented_commands(self):
        poller = _DiscoveryPoller(dumps={"ar.tvplayer.tv": DUMP_TIVIMATE})
        manager = _manager(poller)

        await manager.async_refresh_installed_apps()

        assert "pm list packages -3" in poller.shell_calls
        assert any("resolve-activity" in cmd for cmd in poller.shell_calls)
        assert "android.intent.category.HOME" in " ".join(poller.shell_calls)

    async def test_labels_are_fetched_in_a_bounded_burst(self):
        """A full sweep must not serialise one shell at a time, and must not flood."""
        poller = _DiscoveryPoller()
        manager = _manager(poller)
        concurrent = {"now": 0, "peak": 0, "total": 0}

        real_shell = poller.shell

        async def _counted(cmd):
            if "pm dump" in cmd:
                concurrent["now"] += 1
                concurrent["total"] += 1
                concurrent["peak"] = max(concurrent["peak"], concurrent["now"])
                try:
                    await asyncio.sleep(0)  # yield so overlap is observable
                    return await real_shell(cmd)
                finally:
                    concurrent["now"] -= 1
            return await real_shell(cmd)

        poller.shell = _counted
        rows = await manager.async_refresh_installed_apps()

        assert len(rows) >= 5
        assert concurrent["total"] >= 5
        # bounded: never more than the configured workers in flight at once
        from nsh.const import APP_SWEEP_CONCURRENCY

        assert concurrent["peak"] <= APP_SWEEP_CONCURRENCY
        assert concurrent["peak"] > 1  # ...and genuinely concurrent, not serial

    async def test_a_cached_table_is_reused_and_costs_no_shell(self):
        poller = _DiscoveryPoller(dumps={"ar.tvplayer.tv": DUMP_TIVIMATE})
        manager = _manager(poller)

        first = await manager.async_refresh_installed_apps()
        calls = len(poller.shell_calls)
        second = await manager.async_refresh_installed_apps()

        assert second == first
        assert len(poller.shell_calls) == calls  # cached: zero new shells
        assert manager.installed_apps_stale is False

    async def test_force_refreshes_even_a_fresh_table(self):
        poller = _DiscoveryPoller()
        manager = _manager(poller)

        await manager.async_refresh_installed_apps()
        calls = len(poller.shell_calls)
        await manager.async_refresh_installed_apps(force=True)

        assert len(poller.shell_calls) > calls

    async def test_a_stale_table_is_not_served(self):
        now = [1000.0]
        poller = _DiscoveryPoller()
        manager = ShieldConnectionManager(
            hass=None, entry=None, poller=poller, clock=lambda: now[0]
        )
        await manager.async_refresh_installed_apps()
        assert manager.installed_apps()

        now[0] += APP_LIST_TTL + 1
        assert manager.installed_apps() == []
        assert manager.installed_apps_stale is True

    async def test_the_watchdog_never_sweeps(self):
        """Acceptance: nothing discovery-related runs on a state poll.

        The watchdog cycle is what a poll drives. It may drain ONE label fetch
        (the t24/t55 rule for the live entity path), but it must never list
        packages or fan out into a discovery burst.
        """
        poller = _DiscoveryPoller()
        manager = _manager(poller)

        await manager._watchdog_cycle(5.0)
        await asyncio.sleep(0)

        assert not [c for c in poller.shell_calls if "pm list packages" in c]
        assert not [c for c in poller.shell_calls if "resolve-activity" in c]
        assert manager.installed_apps() == []

    async def test_concurrent_callers_share_one_sweep(self):
        poller = _DiscoveryPoller()
        manager = _manager(poller)
        list_calls = {"n": 0}
        real_shell = poller.shell

        async def _counted(cmd):
            if "pm list packages" in cmd:
                list_calls["n"] += 1
                await asyncio.sleep(0.01)
            return await real_shell(cmd)

        poller.shell = _counted
        results = await asyncio.gather(
            manager.async_refresh_installed_apps(force=True),
            manager.async_refresh_installed_apps(force=True),
            manager.async_refresh_installed_apps(force=True),
        )

        assert list_calls["n"] == 1  # one sweep, shared
        assert results[0] == results[1] == results[2]


class TestDiscoveryDegradesGracefully:
    """ADB off, down or mute: an empty list and a note, never a crash or a loop.

    t62 F1: "could not ask" and "asked, nothing there" are different answers.
    Only the second one is cached; the first is retried.
    """

    async def test_no_poller_is_an_empty_list(self):
        manager = ShieldConnectionManager(hass=None, entry=None, poller=None)

        rows = await manager.async_refresh_installed_apps()

        assert rows == []
        assert manager.installed_apps() == []

    async def test_a_disconnected_poller_is_an_empty_list(self):
        poller = _DiscoveryPoller()
        manager = _manager(poller)
        manager._poller = poller
        poller.connected = False
        # the manager's own connected flag is what the sweep reads
        assert manager.connected is False

        rows = await manager.async_refresh_installed_apps()

        assert rows == []
        assert poller.shell_calls == []  # nothing was even attempted

    async def test_a_mute_transport_is_spaced_but_not_stamped(self):
        """t62 F1: a failed sweep is retried — but not once per state poll."""
        now = [1000.0]
        poller = _DiscoveryPoller(failure=True)
        manager = ShieldConnectionManager(
            hass=None, entry=None, poller=poller, clock=lambda: now[0]
        )

        rows = await manager.async_refresh_installed_apps()

        assert rows == []
        assert poller.shell_calls  # it tried
        assert manager.installed_apps_status == "unknown"  # NOT an empty answer
        # the next poll inside the window does NOT re-shell (no hammering)..
        calls = len(poller.shell_calls)
        assert await manager.async_refresh_installed_apps() == []
        assert len(poller.shell_calls) == calls
        # ..but nothing was stamped as an answer, so once the window passes the
        # sweep runs again. Pre-fix it waited out APP_LIST_TTL (6 h) instead.
        now[0] += manager.APP_SWEEP_RETRY_SECONDS + 1
        assert await manager.async_refresh_installed_apps() == []
        assert len(poller.shell_calls) > calls

    async def test_a_sweep_that_could_not_run_is_not_a_fresh_answer(self):
        """t62 F1 (red proof): startup with ADB down must not BE the answer.

        The entity's one startup sweep runs with the poller down. Once ADB comes
        back, ``homeassistant.update_entity`` (``sensor.async_update``) must
        recover the table. Pre-fix the startup sweep stamped an empty answer, so
        this call issued ZERO shells and kept reporting 0/"pending" for
        APP_LIST_TTL = 6 h.
        """
        poller = _DiscoveryPoller(dumps={"ar.tvplayer.tv": DUMP_TIVIMATE})
        poller.connected = False
        manager = _manager(poller)

        assert await manager.async_refresh_installed_apps() == []  # the startup
        assert poller.shell_calls == []  # it was not even attempted
        assert manager.installed_apps_stale is True  # ...so it is not an answer
        assert manager.installed_apps_status == "unknown"

        poller.connected = True  # ADB is back
        rows = await manager.async_refresh_installed_apps()

        assert rows, "update_entity must recover the table once ADB returns"
        assert "pm list packages -3" in poller.shell_calls
        assert manager.installed_apps_status == "ok"

    async def test_a_genuinely_empty_answer_is_ok(self):
        """t62 F1: 'ok' must be reachable for a really empty device."""
        poller = _DiscoveryPoller(packages="", launcher="")
        manager = _manager(poller)

        assert await manager.async_refresh_installed_apps() == []

        assert poller.shell_calls  # the commands RAN and answered
        assert manager.installed_apps_stale is False  # stamped: a real answer
        assert manager.installed_apps_status == "ok"

    async def test_the_accessor_never_shells(self):
        poller = _DiscoveryPoller()
        manager = _manager(poller)

        assert manager.installed_apps() == []
        assert poller.shell_calls == []


class TestDiscoveryUsesTheCachedLabelPath:
    """Acceptance: labels come from the pm-dump path AND its 24 h cache."""

    async def test_a_second_sweep_reuses_the_cached_labels(self):
        """The 24 h cache is what makes a re-sweep cheap — negatives included.

        A label-less package is a DEFINITIVE answer, so the second sweep must
        not re-dump it. The check goes through the cache directly, because the
        display accessor (`app_label`) cannot tell "no label" from "not asked".
        """
        poller = _DiscoveryPoller(dumps={"ar.tvplayer.tv": DUMP_TIVIMATE})
        manager = _manager(poller)

        await manager.async_refresh_installed_apps()
        dumps = [c for c in poller.shell_calls if "pm dump" in c]
        assert dumps  # the first sweep labelled everything it could
        assert len(dumps) >= 5  # ...and it did dump for the unlabelled ones
        # t66: and then it STOPPED. On this device 5 probes in a row find no label,
        # so the sweep does not spend one shell per package — the packaged-in
        # in-flight window is the only slack above the probe limit.
        from nsh.const import APP_LABEL_PROBE_LIMIT, APP_SWEEP_CONCURRENCY

        assert len(dumps) <= APP_LABEL_PROBE_LIMIT + APP_SWEEP_CONCURRENCY + 1
        assert len(dumps) < len(USER_PACKAGES.splitlines())

        # the cache now answers for EVERY package, so a forced re-sweep lists
        # again but dumps nothing at all (the 24 h label cache, not the table TTL)
        manager._installed_apps = None  # forget the table, keep the labels
        rows = await manager.async_refresh_installed_apps(force=True)

        assert [c for c in poller.shell_calls if "pm dump" in c] == dumps
        assert manager.app_label("ar.tvplayer.tv") == "TiviMate"
        by_pkg = {row["package"]: row for row in rows}
        assert by_pkg["ar.tvplayer.tv"]["name"] == "TiviMate"
        # nothing names kodi (it is in no curated row, and the memo'd device says
        # "no label"): the row shows the id ONCE, in `package`, and name ""
        assert by_pkg["org.xbmc.kodi"]["name"] == ""

    async def test_a_failed_label_is_retried_by_the_next_sweep(self):
        """UNIT 0 semantics at scale: a transient failure is not permanent."""
        # a SMALL app set, with the failing package FIRST: it is the first probe the
        # sweep makes, so its failure is recorded before the device-level memo can
        # form — which is exactly the state the retry guarantee is about
        packages = "package:org.xbmc.kodi\n" + "".join(
            f"package:com.example.app{i}\n" for i in range(6)
        )
        poller = _DiscoveryPoller(
            packages=packages, dumps={"org.xbmc.kodi": "application-label: 'Kodi'"}
        )
        manager = _manager(poller)
        real_shell = poller.shell
        state = {"fail_kodi": True}

        async def _flaky(cmd):
            if state["fail_kodi"] and "pm dump" in cmd and "kodi" in cmd:
                poller.shell_calls.append(cmd)
                return None  # the transport delivered nothing
            return await real_shell(cmd)

        poller.shell = _flaky
        rows = await manager.async_refresh_installed_apps()
        by_pkg = {row["package"]: row for row in rows}
        assert by_pkg["org.xbmc.kodi"]["name"] == ""  # fell back (nothing names it)
        assert manager.app_label("org.xbmc.kodi") is None  # NOT cached

        # t66: the device-level memo is set by now (5 label-less answers), so the
        # sweep will not fan out again — but kodi's FAILED probe is still owed its
        # retry, and the app on screen is what drives it (t57 UNIT 0 survives).
        assert manager.app_labels_absent is True
        assert manager._probe_still_owed("org.xbmc.kodi") is True
        assert await manager._dump_label("org.xbmc.kodi", on_demand=True) is None

        state["fail_kodi"] = False
        await manager._fetch_app_label("org.xbmc.kodi")  # the on-demand retry
        assert manager.app_label("org.xbmc.kodi") == "Kodi"
        manager._installed_apps = None
        rows2 = await manager.async_refresh_installed_apps()
        by_pkg2 = {row["package"]: row for row in rows2}
        assert by_pkg2["org.xbmc.kodi"]["name"] == "Kodi"  # recovered


class TestInstalledAppsSensor:
    """The dashboard surface: state = count, attribute = the table."""

    def _sensor(self, manager, options=None):
        sensor_module = load_entity_modules()["sensor"]
        entry = type("E", (), {})()
        entry.entry_id = "e1"
        entry.options = dict(options or {})
        entry.runtime_data = manager
        entry.title = "mdarcy"
        sensor = sensor_module.NvidiaShieldAppsSensor.__new__(
            sensor_module.NvidiaShieldAppsSensor
        )
        sensor.hass = entity_stublib.FakeHass()
        sensor._entry = entry
        return sensor

    async def test_state_is_the_count_and_the_attribute_is_the_table(self):
        poller = _DiscoveryPoller(dumps={"ar.tvplayer.tv": DUMP_TIVIMATE})
        manager = _manager(poller)
        await manager.async_refresh_installed_apps()
        sensor = self._sensor(manager)

        assert sensor.native_value == len(manager.installed_apps())
        attrs = sensor.extra_state_attributes
        assert attrs["discovery"] == "ok"
        assert any(
            app["package"] == "ar.tvplayer.tv" and app["name"] == "TiviMate"
            for app in attrs["installed_apps"]
        )

    def test_no_discovery_yet_is_an_empty_table_with_a_reason(self):
        poller = _DiscoveryPoller()
        manager = _manager(poller)
        sensor = self._sensor(manager)

        assert sensor.native_value == 0
        attrs = sensor.extra_state_attributes
        assert attrs["installed_apps"] == []
        assert attrs["discovery"] in ("pending", "not_connected", "adb_disabled")

    async def test_an_empty_discovery_is_ok_not_pending(self):
        """t62 F1: a sweep that RAN and found nothing is the answer 'ok'."""
        poller = _DiscoveryPoller(packages="", launcher="")
        manager = _manager(poller)
        await manager.async_refresh_installed_apps()
        sensor = self._sensor(manager, {"adb_enabled": True})

        assert sensor.native_value == 0
        assert sensor.extra_state_attributes == {
            "installed_apps": [],
            "discovery": "ok",
        }

    async def test_a_sweep_that_could_not_run_is_unknown_not_pending(self):
        """t62 F1: 'pending' used to cover both "running" and "could not run"."""
        poller = _DiscoveryPoller(failure=True)
        manager = _manager(poller)
        await manager.async_refresh_installed_apps()
        sensor = self._sensor(manager, {"adb_enabled": True})

        assert sensor.extra_state_attributes == {
            "installed_apps": [],
            "discovery": "unknown",
        }

    def test_adb_disabled_is_reported_as_such(self):
        poller = _DiscoveryPoller()
        manager = _manager(poller)
        poller.connected = False
        sensor = self._sensor(manager, {"adb_enabled": False})

        attrs = sensor.extra_state_attributes

        assert attrs == {"installed_apps": [], "discovery": "adb_disabled"}

    async def test_the_sensor_refresh_is_a_no_op_without_a_manager(self):
        sensor = self._sensor(None)
        sensor._entry.runtime_data = None

        assert sensor.native_value == 0
        assert sensor.extra_state_attributes["installed_apps"] == []
        await sensor.async_update()  # must not raise


class TestDiscoveredAppsStillRespectTheUsersMap:
    """The user's own map is an OVERRIDE, and discovery fills the rest (t57)."""

    async def test_a_mapped_package_keeps_the_users_name_not_the_device_label(self):
        poller = _DiscoveryPoller(dumps={"ar.tvplayer.tv": DUMP_TIVIMATE})
        manager = _manager(poller)
        await manager.async_refresh_installed_apps()

        # the shared resolver is what the sensor and the player use; discovery
        # only supplies the device label underneath the user's map
        from nsh.media_player import resolve_app_name

        options = {CONF_APP_NAME_MAP: {"ar.tvplayer.tv": "Telly"}}
        assert (
            resolve_app_name(
                "ar.tvplayer.tv", options[CONF_APP_NAME_MAP], manager.app_label
            )
            == "Telly"
        )
        # ...and the device label is what discovery put UNDER it, so dropping
        # the map entry still resolves to a real name rather than a package id
        assert (
            resolve_app_name("ar.tvplayer.tv", {}, manager.app_label) == "TiviMate"
        )


def test_the_label_exit_marker_never_becomes_a_name():
    """The marker is a transport detail; it must not leak into a label."""
    from nsh.ha_adb_common import parse_application_label

    assert parse_application_label("__NSH_EXIT:0\n") is None
    assert parse_application_label(DUMP_TIVIMATE + "__NSH_EXIT:0\n") == "TiviMate"


def test_importing_the_discovery_module_needs_no_home_assistant():
    """The parsers live in ha_adb_common and stay importable without HA."""
    import nsh.ha_adb_common.installed_apps as module

    assert module.__doc__ and "pm list packages -3" in module.__doc__
    assert not [n for n in dir(module) if n.startswith("hass")]


class TestTheRealDeviceShape:
    """t66: the user's own `pm dump com.netflix.ninja` shape, and what it implies.

    Three properties are pinned against the fixture, because each one was a live
    defect: the capture carries no label the parser can mistake for a name; the
    fetch must not truncate the dump nor keep shelling every package once the
    device has said it has no labels; and the name the user then SEES comes from
    the curated table.
    """

    def test_no_label_is_extracted_from_the_whole_capture(self):
        from nsh.ha_adb_common import parse_app_label, parse_application_label

        # the capture's shape, in one place: a whole dumpsys (not a stub), whose
        # package section is a resolver table rather than a manifest block
        assert DUMP_NETFLIX_REAL.startswith("DUMP OF SERVICE package:\n")
        assert "Activity Resolver Table:" in DUMP_NETFLIX_REAL
        assert "DUMP OF SERVICE batterystats:" in DUMP_NETFLIX_REAL
        assert len(DUMP_NETFLIX_REAL.splitlines()) > 60

        # (a) NOTHING in it is a name — this is the "no label" the report saw, and
        # it must be a parse result, not a truncation artefact
        assert "application-label" not in DUMP_NETFLIX_REAL
        assert parse_application_label(DUMP_NETFLIX_REAL) is None
        assert parse_app_label(DUMP_NETFLIX_REAL) is None

    def test_the_inferred_shapes_are_not_mistaken_for_names(self):
        """The two ApplicationInfo lines the capture's record does not cover."""
        from nsh.ha_adb_common import parse_label_res, parse_non_localized_label

        assert parse_non_localized_label(DUMP_NETFLIX_REAL) is None  # =null
        assert parse_label_res(DUMP_NETFLIX_REAL) is None  # a resource id
        assert parse_non_localized_label("nonLocalizedLabel=Netflix") == "Netflix"
        assert parse_label_res('labelRes="Netflix"') == "Netflix"
        assert parse_label_res("labelRes=0x7f0f0d31") is None

    async def test_the_search_runs_on_the_device_and_the_dump_is_never_truncated(self):
        poller = _DiscoveryPoller(
            dumps={"com.netflix.ninja": DUMP_NETFLIX_REAL}, on_device_grep=True
        )
        manager = _manager(poller)

        # nothing names this package, and the probe still RAN: a definitive answer
        assert await manager._dump_label("com.netflix.ninja") == ""

        commands = [c for c in poller.shell_calls if "pm dump" in c]
        assert len(commands) == 1
        assert "head -300" not in commands[0]  # the defect: it cut the dump short
        assert "grep -E" in commands[0]  # the fix: filtered before the wire
        for shape in ("application-label", "nonLocalizedLabel", "labelRes"):
            assert shape in commands[0], shape

    async def test_one_shell_per_package_then_the_device_is_believed(self):
        poller = _DiscoveryPoller(
            dumps={"com.netflix.ninja": DUMP_NETFLIX_REAL}, on_device_grep=True
        )
        manager = _manager(poller)

        rows = await manager.async_refresh_installed_apps()
        dumps = [c for c in poller.shell_calls if "pm dump" in c]

        # (b) the sweep did NOT spend one shell per package: on this device 5
        # label-less answers in a row are the device-level conclusion, and only the
        # packages already in flight can add to that count
        from nsh.const import APP_LABEL_PROBE_LIMIT, APP_SWEEP_CONCURRENCY

        assert len(dumps) <= APP_LABEL_PROBE_LIMIT + APP_SWEEP_CONCURRENCY + 1
        assert len(dumps) < len(USER_PACKAGES.splitlines())
        assert manager.app_labels_absent is True
        assert "Netflix" in {row["name"] for row in rows}

        # a forced re-sweep costs ZERO further probes: every package is cached
        manager._installed_apps = None
        await manager.async_refresh_installed_apps(force=True)
        assert [c for c in poller.shell_calls if "pm dump" in c] == dumps

    async def test_the_curated_name_is_what_surfaces(self):
        """(c) the name the user sees, on every surface that shows one."""
        from nsh.const import DEFAULT_APP_DIRECTORY
        from nsh.ha_adb_common import resolve_package_name
        from nsh.media_player import resolve_app_name

        poller = _DiscoveryPoller(
            dumps={"com.netflix.ninja": DUMP_NETFLIX_REAL}, on_device_grep=True
        )
        manager = _manager(poller)
        rows = await manager.async_refresh_installed_apps()
        by_pkg = {row["package"]: row for row in rows}

        # the discovery table
        assert by_pkg["com.netflix.ninja"]["name"] == "Netflix"

        # the chain the entities call: no user map, and the device named nothing
        assert resolve_app_name("com.netflix.ninja", {}, manager.app_label) == "Netflix"
        assert (
            resolve_package_name(
                "com.netflix.ninja", device_label="", curated=DEFAULT_APP_DIRECTORY
            )
            == "Netflix"
        )

        # ...and the installed-apps sensor publishes the same table
        sensor_module = load_entity_modules()["sensor"]
        sensor = sensor_module.NvidiaShieldAppsSensor.__new__(
            sensor_module.NvidiaShieldAppsSensor
        )
        entry = type("E", (), {})()
        entry.entry_id = "e1"
        entry.options = {}
        entry.runtime_data = manager
        entry.title = "mdarcy"
        sensor.hass = entity_stublib.FakeHass()
        sensor._entry = entry
        assert any(
            app["package"] == "com.netflix.ninja" and app["name"] == "Netflix"
            for app in sensor.extra_state_attributes["installed_apps"]
        )


def test_the_name_chain_is_one_order_everywhere():
    """user map > device label > curated > raw id, in ONE pure function (t66)."""
    from nsh.ha_adb_common import resolve_package_name

    curated = {"com.curated": "Curated", "com.both": "Curated Both"}
    user = {"com.mapped": "Mapped", "com.both": "Mapped Both"}

    assert resolve_package_name("com.mapped", user_map=user, curated=curated) == "Mapped"
    assert resolve_package_name("com.curated", device_label="From Device") == "From Device"
    assert resolve_package_name("com.curated", curated=curated) == "Curated"
    assert resolve_package_name("com.both", user_map=user, curated=curated) == "Mapped Both"
    # the device label sits BELOW the user map and ABOVE the curated table
    assert (
        resolve_package_name("com.both", user_map={}, device_label="D", curated=curated)
        == "D"
    )
    # the id is shown ONCE, and an empty package has no name at all
    assert resolve_package_name("com.nobody", curated=curated) == "com.nobody"
    assert resolve_package_name("", curated=curated) is None
    assert resolve_package_name(None) is None
    # a blank device label is not a name (it is what a memo'd device returns)
    assert resolve_package_name("com.curated", device_label="", curated=curated) == "Curated"
    assert resolve_package_name("com.curated", device_label="  ", curated=curated) == "Curated"


def test_the_device_label_lookup_is_lazy():
    """A mapped app must not cost a device probe (t66).

    ``media_player.resolve_app_name`` passes its ``app_label`` accessor, which
    ENQUEUES an on-demand fetch on a miss: calling it for an app the user already
    named would spend a shell to learn something nobody reads.
    """
    from nsh.ha_adb_common import resolve_package_name

    calls: list[str] = []

    def _device_label(package):
        calls.append(package)
        return None

    assert (
        resolve_package_name(
            "com.mapped", user_map={"com.mapped": "Mapped"}, device_label=_device_label
        )
        == "Mapped"
    )
    assert calls == []  # the device was never asked

    assert resolve_package_name("com.other", device_label=_device_label) == "com.other"
    assert calls == ["com.other"]  # ...and it IS asked when the map misses