From 7ed982801ab3e5b70e2a987f92587662756212d1 Mon Sep 17 00:00:00 2001 From: James Mackay Date: Mon, 12 Oct 2026 13:46:55 +0100 Subject: [PATCH] test(adb-off audit): name the carve-outs, drive the paths the audit names (t25) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The t24 review's four observations on the t23 ADB-off audit. No production code changes: button.py is untouched, deliberately (see O1). O1 — THE PAIR ADB CARVE-OUT, ACCEPTED AND NAMED. The button's press builds its own AdbMediaSessionPoller and connects without ever reading the entry's option. I searched for a production route that reaches the press with ADB off and found NONE, so no gate was added: * `button.async_setup_entry` is the ONLY place that entity is created, and it creates it only while the option is on (button.py:34-35); * every options save runs through `NvidiaShieldOptionsFlow(OptionsFlowWithReload)`, which reloads the entry and re-runs that setup — so turning ADB off takes the button away (config_flow.py:1340-1353); * the press is user-initiated in any case — the same class as the already declared `test_connect` pairing probe. It is now a NAMED carve-out in the audit's `CARVE_OUTS` table, with what it does, why that is safe, and the assertions behind both. The measurement is DRIVEN rather than quoted: the press is executed with the option OFF and ON and asserted to cost the same 1 ctor / 1 connect / 1 disconnect, so the carve-out's substance is a test result and not prose. A meta test checks every carve-out's named assertion really exists, so a carve-out cannot point at a phantom. MUTATION-CHECKED: making `async_setup_entry` add the entity unconditionally fails three of the new tests. The intent button is explicitly NOT a carve-out (it can exist for a disabled entry and refuses through the t23 accessor gate) — stated in the same entry. O2 — THE AUDIT NOW EXECUTES WHAT IT NAMES. `_open_logcat_stream` and the logcat ladder are driven both ways: two new rows in the parametrized `AUDITED_PATHS` table (so the disabled case costs zero traffic) plus dedicated assertions — the open RAISES "ADB is switched off" in a `pytest.raises(match=...)`, and the ladder RETURNS at once with `_logcat_failures == []` and `_logcat_disabled is False`, i.e. a local option is not booked as a device failure. The anti-overcorrection pin is kept in the same class: with ADB ON, five real ladder failures DO auto-disable. MUTATION-CHECKED: removing either t23 logcat gate fails three of these tests. O3 — THE ROTTABLE FIGURE IS GONE. The statement_blocks bullet no longer states a count ("currently 28" was stale within one unit of work); the band (~2%) is the policy, the current walk lives in the header the sweep prints, and the comment now says why no band carries a count. The relative_imports bullet lost its illustrative figure for the same reason. O4 — THE ONE WRITE IS NAMED AS A DEBT. The audit's module docstring now states exactly what "nothing is cached" means: the single write a disabled path can still make is `_app_label_failed.add(package)` in `_dump_label` — a retry DEBT, not an answer. It is pinned by a test: with ADB off the debt is recorded while `_app_labels` stays EMPTY and the lookup still answers None; with ADB back on the probe runs, the label is cached and the debt is settled (the discard that the code already does). FAIL-CLOSED, CONFIRMED AND LOCATED. The harness change (`entry=None` now reads as ADB off) matches production because both read the option the same way: the manager returns "disabled" for a missing option, and `__init__.py` builds no poller for the same missing option — asserted by two new tests (an entry with `options={}`, and a manager with no entry at all) and by a source assertion on `__init__.py`'s own `if not entry.options.get(CONF_ADB_ENABLED, False):`. No production path is affected: production always constructs the manager WITH the entry, and for an entry whose option is on nothing changed at all. No regressions: 942 passed / 5 skipped (V1), 943 / 4 (V2) — the +19 are the new tests. ruff clean, drift exit 0, the real-HA pattern collects and runs its 3 files. The device side remains a scripted double: the claim proven is that nothing was sent, not what firmware would do. --- tests/ha_api_drift.py | 22 +- tests/test_adb_off_gates_every_device_path.py | 234 ++++++++++++++++++ tests/test_intents_registration.py | 162 ++++++++++++ 3 files changed, 410 insertions(+), 8 deletions(-) diff --git a/tests/ha_api_drift.py b/tests/ha_api_drift.py index 8138e34..db98168 100644 --- a/tests/ha_api_drift.py +++ b/tests/ha_api_drift.py @@ -158,14 +158,20 @@ HA_CALL_OWNERS: dict[str, tuple[str, str]] = { #: sweep green) is what the 2 catches. This is the project's long-standing #: canary (t62 wrote it as "two below the count"), now stated once instead of #: being implied by each raise. -#: * relative_imports: ~2.5%, currently 7. The count is dominated by package- -#: relative imports that legitimately churn when files move; a collapse (an -#: unparsed module, a stale glob) is orders of magnitude, not 7. -#: * statement_blocks: ~2%, currently 28. This counter moves in BULK with every -#: edit, which is exactly why a fixed small margin would be wrong here (the -#: t19 complaint: at 1300 vs 1322 a 23-statement refactor tripped it). A 2% -#: band leaves a normal refactor alone while a parse failure still shows up as -#: a large drop. +#: * relative_imports: ~2.5%. The count is dominated by package-relative imports +#: that legitimately churn when files move; a collapse (an unparsed module, a +#: stale glob) is orders of magnitude, not a handful. +#: * statement_blocks: ~2%. This counter moves in BULK with every edit, which is +#: exactly why a fixed small margin would be wrong here (the t19 complaint: at +#: 1300 vs 1322 a 23-statement refactor tripped it). A 2% band leaves a normal +#: refactor alone while a parse failure still shows up as a large drop. +#: +#: No band states a COUNT: the current walk is in the header above and printed by +#: every sweep run, so a figure in this prose can only rot — a previous revision +#: carried exactly such a figure for statement_blocks and it was stale within one +#: unit of work (t24/O3). The BANDS are the policy and are deliberately the only +#: numbers here; a floor, once re-derived from a re-measured walk, is named next to +#: the counter that owns it. #: * awaited_calls: 1 (measured 6, floor 5 — DELIBERATE, see below). Stated #: plainly: with six awaited HA calls a percentage margin is meaningless, so #: this floor is the weakest one here. It is still worth having — a walk that diff --git a/tests/test_adb_off_gates_every_device_path.py b/tests/test_adb_off_gates_every_device_path.py index 5740ba1..3e5d40f 100644 --- a/tests/test_adb_off_gates_every_device_path.py +++ b/tests/test_adb_off_gates_every_device_path.py @@ -12,12 +12,27 @@ 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 @@ -197,10 +212,68 @@ AUDITED_PATHS = [ ("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): @@ -231,6 +304,16 @@ 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 @@ -410,3 +493,154 @@ class TestTheOtherStatusesAreUntouched: 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" + ) diff --git a/tests/test_intents_registration.py b/tests/test_intents_registration.py index 3344c95..754f11c 100644 --- a/tests/test_intents_registration.py +++ b/tests/test_intents_registration.py @@ -38,6 +38,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parent)) import entity_stublib # noqa: E402 from entity_stublib import load_entity_modules # noqa: E402 from nsh.const import ( # noqa: E402 + CONF_ADB_ENABLED, CONF_REGISTERED_INTENTS, INTENT_FIELD_ACTION, INTENT_FIELD_ACTIVITY, @@ -1370,3 +1371,164 @@ def test_a_truncated_capture_prefills_no_url(modules): assert "www.exam" not in values[INTENT_FIELD_DATA] assert values[INTENT_FIELD_ACTION] == "android.intent.action.VIEW" assert values[INTENT_FIELD_ACTIVITY] == "com.netflix.ninja/.MainActivity" + +# ── t25/O1: the Pair ADB BUTTON is a NAMED carve-out of the ADB-off audit ── + + +class TestThePairAdbCarveOut: + """Why the ADB-off audit deliberately does NOT gate the Pair ADB press. + + `NvidiaShieldPairAdbButton.async_press` builds its OWN poller and connects — + it never asks the entry's ADB option (the t24 review measured + ctor/connect/disconnect 1/1/1 with the option OFF, identical to option-ON). + That is accepted, not overlooked, because there is no production route to the + press with ADB off: + + * `button.async_setup_entry` adds that entity ONLY for an entry whose option + is on, so a disabled entry has NO button to press (asserted below); + * every options save goes through `NvidiaShieldOptionsFlow(OptionsFlowWithReload)`, + which reloads the entry — and the reload re-runs that setup, so turning ADB + off takes the button away; + * the press itself is USER-INITIATED (somebody had to press it), which is the + same class as the `ShieldConnectionManager.test_connect` pairing probe. + + If a future change ever makes that entity exist for a disabled entry, this + class fails — which is the point: the carve-out is a claim like any other. + """ + + async def _setup(self, modules, options): + import importlib + + entry = SimpleNamespace( + entry_id="e1", options=options, title="Shield", runtime_data=None + ) + added: list = [] + button = importlib.import_module("nsh.button") + await button.async_setup_entry(None, entry, added.extend) + return added + + @staticmethod + def _pair_adb(added): + return [entity for entity in added if getattr(entity, "_attr_name", None) == "Pair ADB"] + + async def test_no_pair_adb_button_while_adb_is_off(self, modules): + """THE MITIGATION: a disabled entry has no button to press.""" + added = await self._setup(modules, {}) + + assert self._pair_adb(added) == [] + assert all( + not getattr(entity, "_attr_name", "").startswith("Pair") + for entity in added + ) + + async def test_no_pair_adb_button_with_an_explicit_off_option(self, modules): + added = await self._setup(modules, {CONF_ADB_ENABLED: False}) + + assert self._pair_adb(added) == [] + + async def test_the_button_is_back_when_adb_is_on(self, modules): + """The anti-overcorrection pin: the entity is not gone for good (t25).""" + added = await self._setup(modules, {CONF_ADB_ENABLED: True}) + + pair = self._pair_adb(added) + assert len(pair) == 1 + assert pair[0]._attr_unique_id.endswith("_pair_adb") + + async def test_the_press_does_not_consult_the_option_and_that_is_pinned( + self, modules + ): + """THE MEASUREMENT, executed: identical traffic whichever way the option reads. + + `async_press` never asks the entry's ADB option, so a press costs one + constructor, one connect and one disconnect with the option OFF — the same + as option-ON (the t24 review measured 1/1/1 both ways). This is the + carve-out's substance, so it is DRIVEN rather than quoted: if a future + change adds a gate here, this fails and the carve-out entry (and the audit + text beside it) must be updated in the same change — which is the point. + """ + import importlib + + button = importlib.import_module("nsh.button") + + class RecordingPoller: + constructed = 0 + connects = 0 + disconnects = 0 + + def __init__(self, host, port, key_path): + type(self).constructed += 1 + self.host = host + + async def connect(self): + type(self).connects += 1 + return True + + async def disconnect(self): + type(self).disconnects += 1 + + class NoReload: + async def async_reload(self, entry_id): + return True + + original_poller = button.AdbMediaSessionPoller + original_key = button.ensure_adb_key + button.AdbMediaSessionPoller = RecordingPoller + button.ensure_adb_key = lambda config_dir: "/tmp/test-key.pem" + try: + measured = {} + for label, adb_enabled in (("off", False), ("on", True)): + RecordingPoller.constructed = 0 + RecordingPoller.connects = 0 + RecordingPoller.disconnects = 0 + entry = SimpleNamespace( + entry_id="e1", + options={ + CONF_ADB_ENABLED: adb_enabled, + "host": "shield.invalid", + }, + title="Shield", + runtime_data=None, + ) + entity = button.NvidiaShieldPairAdbButton(None, entry) + entity.hass = SimpleNamespace( + config=SimpleNamespace(config_dir="/tmp"), + async_add_executor_job=FakeHass.async_add_executor_job, + config_entries=NoReload(), + ) + await entity.async_press() + measured[label] = ( + RecordingPoller.constructed, + RecordingPoller.connects, + RecordingPoller.disconnects, + ) + finally: + button.AdbMediaSessionPoller = original_poller + button.ensure_adb_key = original_key + + assert measured["off"] == (1, 1, 1), measured["off"] + assert measured["on"] == (1, 1, 1), measured["on"] + assert measured["off"] == measured["on"], ( + "the option does not gate this path — that is the named carve-out" + ) + + async def test_the_reload_that_removes_it_is_the_options_flow(self): + """The second half of the mitigation, pinned where it lives.""" + source = ( + Path(__file__).resolve().parent.parent + / "custom_components" + / "nvidia_shield" + / "config_flow.py" + ).read_text(encoding="utf-8") + + assert "OptionsFlowWithReload" in source, ( + "an options save must reload the entry (which re-runs button setup)" + ) + button_source = ( + Path(__file__).resolve().parent.parent + / "custom_components" + / "nvidia_shield" + / "button.py" + ).read_text(encoding="utf-8") + assert "if entry.options.get(CONF_ADB_ENABLED, False):" in button_source, ( + "the Pair ADB entity is created only while the option is on" + )