"""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. WHAT "NOTHING IS CACHED" MEANS EXACTLY (t24/O4): the ONE write a disabled path can still make is a retry DEBT — ``_app_label_failed.add(package)`` in ``_dump_label`` — and a debt is not an answer: the label cache (``_app_labels``) stays untouched, ``app_label`` keeps returning None, and the debt is settled the moment the probe finally runs with ADB back on. Asserted in ``test_the_only_write_while_off_is_a_retry_debt_not_an_answer``. WHAT THIS FILE DOES NOT GATE (t25/O1): the paths where "ADB off must mean no traffic" is deliberately NOT enforced, each listed in :data:`CARVE_OUTS` with its mitigation and with the test that asserts that mitigation — the Pair ADB button's press and ``ShieldConnectionManager.test_connect``. Naming them is the point: an audit that silently excludes a path is not an audit, and a carve-out whose assertion does not exist is prose. """ from __future__ import annotations import asyncio import contextlib from pathlib import Path 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)), ("logcat open", lambda m: _open_logcat(m)), ("logcat ladder", lambda m: m._run_logcat_stream()), ("watchdog start", lambda m: _start_watchdog(m)), ] #: NAMED CARVE-OUTS (t25/O1) — the device paths this audit does NOT gate, each with #: what it does, why that is safe, and the assertion that proves the mitigation. A #: carve-out is a claim like every other claim here, so #: ``test_every_carve_out_names_a_real_assertion`` checks that the named assertion #: actually exists in the tree: a carve-out cannot point at a phantom. CARVE_OUTS = ( { "name": "Pair ADB button press (button.py::NvidiaShieldPairAdbButton)", "what_it_does": ( "builds its OWN AdbMediaSessionPoller and connects it, and never reads " "the entry's ADB option — MEASURED ctor/connect/disconnect 1/1/1 with " "the option OFF, identical to option-ON (driven, not quoted)" ), "why_safe": ( "there is no production route to the press with ADB off: " "button.async_setup_entry adds that entity ONLY when the option is on, " "so a disabled entry has no button to press; every options save runs " "through NvidiaShieldOptionsFlow(OptionsFlowWithReload), which reloads " "the entry and re-runs that setup, so turning ADB off takes the button " "away; and a press is user-initiated — the same class as the " "test_connect pairing probe below" ), "asserted_by": ( "tests/test_intents_registration.py", "test_no_pair_adb_button_while_adb_is_off", ), "measured_by": ( "tests/test_intents_registration.py", "test_the_press_does_not_consult_the_option_and_that_is_pinned", ), "not_the_same_as": ( "the INTENT button, which CAN exist for an ADB-off entry and is NOT a " "carve-out: it sends through run_registered_intent, which resolves the " "poller through the gated adb_poller accessor (media_player.py:2317) and " "refuses with `intents_need_adb` — zero traffic" ), }, { "name": "ShieldConnectionManager.test_connect (the config-flow pairing probe)", "what_it_does": ( "opens a throwaway AdbMediaSessionPoller and connects, before any entry " "exists, consulting no option at all" ), "why_safe": ( "the setup menu offers that step only while the setup's own ADB option " "is on (the menu row is built from _user_data[CONF_ADB_ENABLED]), and " "the probe IS the user's explicit pairing action" ), "asserted_by": ( "tests/test_adb_off_gates_every_device_path.py", "test_the_pairing_probe_is_offered_only_while_the_option_is_on", ), }, ) 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() def _open_logcat(manager): """The logcat OPEN path: it raises while disabled, so swallow it here. ``test_the_logcat_open_refuses_with_the_disabled_message`` asserts the message; this driver only has to prove the raise cost no traffic. """ with contextlib.suppress(RuntimeError): manager._open_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 ) REPO_ROOT = Path(__file__).resolve().parent.parent def _source(relative_path: str) -> str: return (REPO_ROOT / relative_path).read_text(encoding="utf-8") class TestTheNamedCarveOuts: """Every exclusion is a NAMED claim with an assertion behind it (t25/O1).""" @pytest.mark.parametrize("carve_out", CARVE_OUTS, ids=[c["name"] for c in CARVE_OUTS]) def test_every_carve_out_names_a_real_assertion(self, carve_out): """No phantom carve-outs: the named assertion must exist in the tree.""" for key in ("asserted_by", "measured_by"): pointer = carve_out.get(key) if pointer is None: continue relative, name = pointer assert f"def {name}(" in _source(relative), ( f"{carve_out['name']} claims {key}={relative}::{name}, which does " "not exist" ) assert carve_out["what_it_does"], "a carve-out states what it does" assert carve_out["why_safe"], "a carve-out states its mitigation" @pytest.mark.parametrize("carve_out", CARVE_OUTS, ids=[c["name"] for c in CARVE_OUTS]) def test_every_carve_out_is_outside_the_audited_table(self, carve_out): """A carve-out is an EXCLUSION — it must not also claim to be driven.""" driven = {name for name, _ in AUDITED_PATHS} assert all( carve_out["name"] not in driven_path for driven_path in driven ), carve_out["name"] def test_the_pairing_probe_is_offered_only_while_the_option_is_on(self): """`test_connect` has no entry to ask, so the SETUP MENU is its gate.""" config_flow = _source("custom_components/nvidia_shield/config_flow.py") menu_guard = ( "if self._user_data.get(CONF_ADB_ENABLED) and not getattr(\n" ' self, "_paired_adb", False\n' " ):\n" ' options.append("adb_pair")' ) assert menu_guard in config_flow, ( "the adb_pair row (the only route to the pairing probe) must stay " "behind the setup's ADB option" ) class TestTheLogcatPathsTheAuditNames: """The audit text names these two paths — so they are DRIVEN here (t25/O2).""" async def test_the_logcat_open_refuses_with_the_disabled_message(self): poller = WillingPoller() manager = _manager(poller, adb_enabled=False) with pytest.raises(RuntimeError, match="ADB is switched off"): manager._open_logcat_stream() assert poller.traffic == [], "the refusal cost no traffic" async def test_the_logcat_ladder_returns_without_counting_a_failure(self): """A local option must not look like a device failure (t23, pinned t25).""" poller = WillingPoller() manager = _manager(poller, adb_enabled=False) await manager._run_logcat_stream() # returns at once: nothing to consume assert manager._logcat_failures == [], "no failure was recorded" assert manager._logcat_disabled is False, "and the stream was not disabled" assert poller.traffic == [] async def test_five_real_logcat_failures_still_auto_disable(self): """THE ANTI-OVERCORRECTION PIN: the ladder itself still works when ADB is on.""" poller = WillingPoller() manager = _manager(poller, adb_enabled=True) async def failing_pass(): raise RuntimeError("stream ended (device closed)") async def no_sleep(_seconds): return None manager._logcat_pass = failing_pass manager._logcat_ladder_sleep = no_sleep await manager._run_logcat_stream() assert manager._logcat_disabled is True, "five real failures still trip it" assert manager._entry_options_logcat() is True, "the option itself is unchanged" class TestTheOneWriteWhileOffIsDebt: """t25/O4: "nothing is cached" is exact because the one write is a DEBT.""" async def test_the_only_write_while_off_is_a_retry_debt_not_an_answer(self): poller = WillingPoller() manager = _manager(poller, adb_enabled=False) # the one disabled call that reaches the probe directly (the watchdog drain # and the public lookup are gated above it, so this is the route the review # measured the write on) await manager._fetch_app_label(PKG) assert manager._app_labels == {}, "no label was CACHED — that is the claim" assert manager._app_label_failed == {PKG}, "only the retry DEBT was recorded" assert manager.app_label(PKG) is None, "and the lookup still answers nothing" assert manager.app_label_inflight == set() assert poller.traffic == [], "the debt cost no traffic" manager.entry.options[CONF_ADB_ENABLED] = True # ADB is turned back on await manager._fetch_app_label(PKG) # the owed retry finally runs assert manager._app_labels.get(PKG) is not None, "the probe ran and cached" assert manager._app_label_failed == set(), "and the debt was settled" class TestAMissingOptionFailsClosed: """Where the `entry=None`-means-ADB-off behaviour is asserted (t25).""" async def test_an_entry_with_no_adb_option_is_disabled(self): """No option recorded = OFF, so the harness change matched production.""" poller = WillingPoller() entry = SimpleNamespace(options={}, data={}, title="Shield", entry_id="e1") manager = ShieldConnectionManager(hass=None, entry=entry, poller=poller) assert manager._adb_disabled() is True assert manager._adb_active() is False assert manager.adb_poller is None assert await manager.send_media_command("play") is False assert await manager.async_refresh_installed_apps(force=True) == [] assert manager.installed_apps() == [] assert poller.traffic == [] async def test_a_manager_with_no_entry_at_all_is_disabled_too(self): """The old harness shape (`entry=None`) is the same fact.""" poller = WillingPoller() manager = ShieldConnectionManager(hass=None, entry=None, poller=poller) assert manager._adb_disabled() is True assert await manager.send_digit_key("5") is False assert poller.traffic == [] def test_the_production_default_is_read_the_same_way(self): """`__init__.py` gates the poller on the same missing-option default.""" entry_source = _source("custom_components/nvidia_shield/__init__.py") assert "if not entry.options.get(CONF_ADB_ENABLED, False):" in entry_source, ( "production treats a missing option as OFF, exactly like the manager" )