ha-nvidia-shield/tests/test_adb_off_gates_every_device_path.py

"""t23/O1 — with ADB switched off, NO path reaches the device.

The t22 review measured the harm: with a poller injected into an entry whose own
option said ``adb_enabled: False``, the installed-apps sweep sent **4** real
shells and a label probe sent **1** more. Both gated on the POLLER's absence,
which is a property of the setup code (``_build_poller_factory`` returns None when
ADB is off) — exactly the indirect guarantee the t21 repair had already proven
insufficient for the intent scan.

This file drives EVERY path in the connection manager that can touch the device
and asserts on the TRAFFIC (shells, polls, commands, connects), because the
traffic is the harm — a correct-looking status string is not the point. The
device side is a scripted double that would answer everything: no real Shield is
involved, so what is proven is that nothing was sent, not what firmware would do.
"""

from __future__ import annotations

import asyncio
import contextlib
from types import SimpleNamespace

import pytest
from nsh.connection import (
    FIELD_AWAKE,
    FIELD_FOREGROUND,
    FIELD_RUNNING_APPS,
    FIELD_SESSIONS,
    ShieldConnectionManager,
)
from nsh.const import CONF_ADB_ENABLED, CONF_LOGCAT_EVENTS
from nsh.ha_adb_common import (
    INTENT_SCAN_ADB_DISABLED,
    INTENT_SCAN_ADB_UNAVAILABLE,
    INTENT_SCAN_FRESH,
    INTENT_SCAN_FRESH_EMPTY,
)

PKG = "com.netflix.ninja"

#: A minimal answer that IS a resolver section, so a scan that were allowed to run
#: would count as a real answer (nothing here fails for a second reason).
RESOLVER_SECTION = (
    "Activity Resolver Table:\n"
    "  Schemes:\n"
    "    https:\n"
    "      deca7fa com.netflix.ninja/.MainActivity filter fc1\n"
    '      Action: "android.intent.action.VIEW"\n'
    "Service Resolver Table:\n"
)


class WillingPoller:
    """A poller that answers EVERY call and records every one (t23/O1).

    Nothing here refuses: if a path is still gated by the wrong fact, this double
    will happily talk to "the device" and the shell/command count proves it.
    """

    def __init__(self):
        self.connected = True
        self.awake = True
        self.foreground = "app.a"
        self.audio_dump = "  package=app.a\n"
        self.shell_calls: list[str] = []
        self.poll_calls: list[str] = []
        self.media_calls: list[str] = []
        self.digit_calls: list[str] = []
        self.foreground_sends: list[str] = []
        self.enter_calls = 0
        self.connect_calls = 0
        self.disconnect_calls = 0
        self.sessions = [SimpleNamespace(package="app.a", playing=True)]

    # ── the transport ──
    async def connect(self) -> bool:
        self.connect_calls += 1
        self.connected = True
        return True

    async def disconnect(self) -> None:
        self.disconnect_calls += 1
        self.connected = False

    async def health_check(self) -> bool:
        return self.connected

    # ── the heartbeat ──
    async def is_awake(self) -> bool | None:
        self.poll_calls.append("is_awake")
        return self.awake

    async def poll_media_sessions(self, foreground_package=None, head_lines=200):
        self.poll_calls.append("poll_media_sessions")
        return self.sessions[0] if self.sessions else None

    @property
    def all_sessions(self):
        return self.sessions

    @property
    def last_dump(self):
        return self.audio_dump

    async def poll_running_apps(self) -> list[dict]:
        self.poll_calls.append("poll_running_apps")
        return [{"package": "app.a", "name": "App A", "icon": None}]

    async def poll_foreground(self) -> str | None:
        self.poll_calls.append("poll_foreground")
        return self.foreground

    # ── anything a shell can reach ──
    async def shell(self, command: str):
        self.shell_calls.append(command)
        if "Activity Resolver Table" in command:
            return RESOLVER_SECTION
        if "pm list packages" in command:
            return f"package:{PKG}\n"
        if "resolve-activity" in command:
            return f"{PKG}/.MainActivity\n"
        if "pm dump" in command:
            return 'Application Label: "Netflix"\n'
        return "ok\n"

    async def send_media_command(self, command: str) -> bool:
        self.media_calls.append(command)
        return True

    async def bring_to_foreground(self, package: str) -> bool:
        self.foreground_sends.append(package)
        return True

    async def send_digit_keyevent(self, digit: str) -> bool:
        self.digit_calls.append(digit)
        return True

    async def send_enter_keyevent(self) -> bool:
        self.enter_calls += 1
        return True

    @property
    def config(self):
        return {"host": "shield.invalid", "port": 5555, "key_path": "/tmp/does-not-exist"}

    @property
    def traffic(self) -> list[str]:
        """EVERY device-touching call, in one list — the harm in one place (t23)."""
        return [
            *(f"shell:{call}" for call in self.shell_calls),
            *(f"poll:{call}" for call in self.poll_calls),
            *(f"media:{call}" for call in self.media_calls),
            *(f"digit:{digit}" for digit in self.digit_calls),
            *(f"foreground:{pkg}" for pkg in self.foreground_sends),
            *(["enter"] * self.enter_calls),
            *(["connect"] * self.connect_calls),
        ]


def _entry(*, adb_enabled: bool, **options):
    """The entry's OWN options decide — this is the fact every path must ask."""
    return SimpleNamespace(
        options={
            CONF_ADB_ENABLED: adb_enabled,
            # ON, so the logcat path is skipped for the OPTION's reason (ADB off)
            # rather than for its own (a stream nobody asked for).
            CONF_LOGCAT_EVENTS: True,
            **options,
        },
        data={},
        title="Shield",
        entry_id="e1",
    )


def _manager(poller, *, adb_enabled: bool) -> ShieldConnectionManager:
    return ShieldConnectionManager(
        hass=None, entry=_entry(adb_enabled=adb_enabled), poller=poller
    )


#: THE AUDIT (t23/O1), as data: every path in the manager that can reach the
#: device, with the call that drives it. Each is run against a manager whose
#: option says ADB is off, and the poller's traffic must stay EMPTY.
AUDITED_PATHS = [
    ("intent scan", lambda m: m.async_scan_intents(PKG)),
    ("intent scan (forced)", lambda m: m.async_scan_intents(PKG, force=True)),
    ("installed-apps refresh", lambda m: m.async_refresh_installed_apps(force=True)),
    ("installed-apps refresh (non-forced)", lambda m: m.async_refresh_installed_apps()),
    ("app-label lookup", lambda m: _sync_label(m)),
    ("app-label drain (watchdog side)", lambda m: _drain(m)),
    ("adb text entry", lambda m: m.send_adb_input_text("hello")),
    ("media key send", lambda m: m.send_media_command("play")),
    ("bring to foreground", lambda m: m.bring_to_foreground(PKG)),
    ("digit key send", lambda m: m.send_digit_key("5")),
    ("enter key send", lambda m: m.send_enter_key()),
    ("watchdog cycle", lambda m: m._watchdog_cycle(5.0)),
    ("heartbeat refresh", lambda m: m._refresh()),
    ("logcat ensure", lambda m: _logcat(m)),
    ("watchdog start", lambda m: _start_watchdog(m)),
]


async def _maybe_await(value):
    """Drive an audited path: some are coroutines, some are plain calls (t23)."""
    if asyncio.iscoroutine(value):
        return await value
    return value


def _sync_label(manager):
    return manager.app_label(PKG)


async def _settle_label_tasks(manager):
    """Await any label fetch the drain started, so its shell WOULD have happened."""
    for task in list(manager._tasks):
        if task.get_name().startswith("app_label"):
            with contextlib.suppress(Exception):
                await task


async def _drain(manager):
    """The render path queues a label fetch, then the watchdog drains the queue."""
    manager.app_label(PKG)
    manager._drain_app_label_requests()
    await _settle_label_tasks(manager)


def _logcat(manager):
    manager._ensure_logcat_stream()


async def _start_watchdog(manager):
    await manager.async_start()
    await asyncio.sleep(0.01)  # would be one cycle if the loop had started
    await manager.async_stop()


class TestEveryAuditedPathSendsNothing:
    """ADB off + a poller that WOULD answer = zero traffic on every path."""

    @pytest.mark.parametrize(
        ("name", "drive"), AUDITED_PATHS, ids=[name for name, _ in AUDITED_PATHS]
    )
    async def test_the_path_does_not_touch_the_device(self, name, drive):
        poller = WillingPoller()
        manager = _manager(poller, adb_enabled=False)

        await _maybe_await(drive(manager))
        await asyncio.sleep(0)  # let any task the path started really run

        assert poller.traffic == [], f"{name} reached the device: {poller.traffic}"

    async def test_a_full_cycle_of_every_path_still_sends_nothing(self):
        """One manager, every path in sequence — the whole cycle, not one call."""
        poller = WillingPoller()
        manager = _manager(poller, adb_enabled=False)

        # 1. the intent scan (t21's repair, asked again one layer deeper)
        assert (await manager.async_scan_intents(PKG)).status == INTENT_SCAN_ADB_DISABLED
        assert (
            await manager.async_scan_intents(PKG, force=True)
        ).status == INTENT_SCAN_ADB_DISABLED

        # 2. the installed-apps sweep — the 4-shell path the review measured
        assert await manager.async_refresh_installed_apps(force=True) == []
        assert manager.installed_apps() == []

        # 3. the app-label probe — the 1-shell path, both entry points
        assert manager.app_label(PKG) is None
        manager._drain_app_label_requests()
        await _settle_label_tasks(manager)

        # 4. text entry and every key/command delegation
        assert await manager.send_adb_input_text("hello") is False
        assert await manager.send_media_command("play") is False
        assert await manager.bring_to_foreground(PKG) is False
        assert await manager.send_digit_key("5") is False
        assert await manager.send_enter_key() is False

        # 5. the heartbeat, the watchdog and the logcat stream
        await manager._watchdog_cycle(5.0)
        await _start_watchdog(manager)
        manager._ensure_logcat_stream()
        await asyncio.sleep(0)

        # 6. the public accessor the OUT-OF-FILE consumers (adb_command service,
        #    screen capture, intent shell) read — it must refuse too
        assert manager.adb_poller is None

        assert poller.traffic == [], poller.traffic

    async def test_nothing_is_cached_when_the_cycle_is_refused(self):
        """No cache write either: an unpermitted scan is not an answer."""
        poller = WillingPoller()
        manager = _manager(poller, adb_enabled=False)

        await manager.async_scan_intents(PKG)
        await manager.async_refresh_installed_apps(force=True)
        manager.app_label(PKG)
        manager._drain_app_label_requests()
        await manager._refresh()

        assert manager.intent_scan_state(PKG) is None, "no intent answer"
        assert manager.installed_apps() == [], "no app table"
        assert manager.installed_apps_stale is True, "nothing was stamped fresh"
        assert manager.app_label_requests == set(), "nothing queued"
        assert manager.app_label_inflight == set(), "nothing in flight"
        assert manager.store.get(FIELD_AWAKE) is None, "no wake reading"
        assert manager.store.get(FIELD_SESSIONS, []) == []
        assert manager.store.get(FIELD_RUNNING_APPS, []) == []
        assert manager.store.get(FIELD_FOREGROUND) is None
        assert manager.awake is None


class TestTheWatchdogNeverConnects:
    """A TCP connect is device traffic too — the loop must not even start."""

    async def test_the_watchdog_does_not_build_a_poller_or_connect(self):
        ours = WillingPoller()
        built = []

        def factory():
            built.append(1)
            return ours

        manager = ShieldConnectionManager(
            hass=None,
            entry=_entry(adb_enabled=False),
            poller_factory=factory,
        )
        await manager.async_start()
        await asyncio.sleep(0.02)
        await manager.async_stop()

        assert built == [], "the factory was never asked for a poller"
        assert ours.connect_calls == 0, "and nothing was connected"
        assert ours.traffic == []

    async def test_a_direct_cycle_does_not_connect_either(self):
        poller = WillingPoller()
        poller.connected = False
        manager = _manager(poller, adb_enabled=False)

        await manager._watchdog_cycle(5.0)

        assert poller.connect_calls == 0
        assert poller.traffic == []


class TestTheOptionIsACheckNotALatch:
    """Turning ADB back on resumes normal work — no remembered decision."""

    async def test_the_same_manager_resumes_when_the_option_flips(self):
        poller = WillingPoller()
        manager = _manager(poller, adb_enabled=False)

        assert await manager.async_refresh_installed_apps(force=True) == []
        assert manager.app_label(PKG) is None
        assert await manager.send_media_command("play") is False
        assert poller.traffic == [], "nothing while it is off"

        manager.entry.options[CONF_ADB_ENABLED] = True  # the user turns ADB on

        rows = await manager.async_refresh_installed_apps(force=True)
        assert rows, "the sweep really ran once ADB is on"
        assert manager.app_label("com.other.app") is None  # queued, not sent yet
        assert manager.app_label_requests == {"com.other.app"}
        assert await manager.send_media_command("play") is True
        assert await manager.send_digit_key("5") is True
        assert manager.adb_poller is poller
        assert poller.shell_calls, "the device was reached again"
        assert poller.media_calls == ["play"]
        assert poller.digit_calls == ["5"]

    async def test_the_intent_scan_resumes_too(self):
        poller = WillingPoller()
        manager = _manager(poller, adb_enabled=False)
        assert (await manager.async_scan_intents(PKG)).status == INTENT_SCAN_ADB_DISABLED

        manager.entry.options[CONF_ADB_ENABLED] = True

        result = await manager.async_scan_intents(PKG)
        assert result.status in (INTENT_SCAN_FRESH, INTENT_SCAN_FRESH_EMPTY)
        assert result.answered is True
        assert manager.intent_scan_state(PKG) is not None, "the answer is cached now"


class TestTheOtherStatusesAreUntouched:
    """ADB ON with no poller is 'unavailable', never 'disabled' (t21's split)."""

    async def test_no_poller_reads_as_unavailable(self):
        manager = _manager(None, adb_enabled=True)

        assert (await manager.async_scan_intents(PKG)).status == (
            INTENT_SCAN_ADB_UNAVAILABLE
        )
        assert await manager.async_refresh_installed_apps(force=True) == []
        assert manager.installed_apps_status == "unknown"
        assert await manager.send_media_command("play") is False

    async def test_adb_off_and_no_poller_are_told_apart(self):
        disabled = _manager(WillingPoller(), adb_enabled=False)
        unbuilt = _manager(None, adb_enabled=True)

        assert (await disabled.async_scan_intents(PKG)).status == (
            INTENT_SCAN_ADB_DISABLED
        )
        assert (await unbuilt.async_scan_intents(PKG)).status == (
            INTENT_SCAN_ADB_UNAVAILABLE
        )