14 Commits
Author SHA1 Message Date
James Mackay c588a7c66b docs(readme): record the MAC-less discovery contract (t37 D3, t39)
Release / release (push) Successful in 5s
Test / test (push) Successful in 22s
Test / ha-api-drift (push) Successful in 2m35s
t37 made MAC-less zeroconf discovery honour Home Assistant's own
`_async_handle_discovery_without_unique_id` contract — the helper is a coroutine
and its dedup (abort `already_configured` when ANY entry exists for this domain,
abort `already_in_progress` when another flow runs) only happens when it is
awaited. That is stricter than the accidental host-equality guard it replaced,
so it changes what a user can expect:

* with a MAC, discovery still aborts duplicates and updates a moved device's
  hosts;
* without a MAC, a second (or third) Shield must be added MANUALLY — discovery
  will abort as "already configured".

Someone with two Shields would otherwise read that abort as a bug. The README
now has a "Discovery (zeroconf)" section under Configuration stating the rule and
the manual route, and the drift sweep's docstring carries the same note next to
the D3 entry, so the decision is discoverable from wherever you meet it.

Verified: 289 passed, 2 skipped; ruff clean.
2026-10-10 21:29:46 +01:00
James Mackay fd62d850b7 fix(drift): F1+F2 — the sweep refuses false greens, and CI provisions the real floor (t39)
t38 found two holes in the drift gate itself.

F2 — the job could never run where it mattered. Home Assistant 2026.10.1
declares ``Requires-Python >=3.14.2`` (PyPI metadata), so the old guard let a
3.13 runner through and the ``homeassistant>=2026.10`` install then failed with
a resolver error. The job now provisions its own interpreter
(``actions/setup-python@v5`` with ``python-version: "3.14"``) and the guard runs
against the REAL floor, read from ``ha_api_drift.HOMEASSISTANT_PYTHON_FLOOR`` —
one source of truth, pinned by a test against the installed distribution's own
``Requires-Python`` metadata. Verified: on 3.13.16 the guard refuses with
"Home Assistant needs Python >= 3.14.2, but this job runs 3.13.16 / Fix: raise
actions/setup-python…"; on 3.14.8 it reports "interpreter ok"; and
``uvx --python 3.14 --with 'homeassistant>=2026.10'`` resolves 2026.10.1.

F1 — the standalone entry point was a false green. It printed counters and
exited 0 even when it had walked nothing, which is exactly how the manual
red-proof tool gets misused. The non-vacuity floors now live in
``ha_api_drift.FLOORS`` and are enforced in ONE place, ``Report.floor_violations()``,
used by the pytest wrapper and by ``__main__`` (so the constants cannot drift
apart; a test asserts the wrapper holds no floor-like constants of its own).
On top of that the sweep can no longer lie in either direction:

* an empty or mis-targeted directory reports ``FLOOR:`` lines and
  ``drift sweep FAILED`` and exits 1 (a non-existent path is a ``bad-path``
  finding, not a traceback and not a silent zero-module sweep);
* with no homeassistant installed it returns a single ``no-homeassistant``
  finding with zero counters — never a wall of phantom "broken-import"s, never
  a green exit.

Verified in the CI-equivalent environment (Python 3.14.8 + homeassistant
2026.10.1): HEAD exits 0 with "no Home-Assistant API drift (13 modules, 60
imports, 35 call arguments, 7 awaited HA calls)"; the empty directory exits 1
with 4 FLOOR lines and no findings; a missing path exits 1 with the ``bad-path``
finding; and the full pre-fix tree (``git archive 5ede4fe``) exits 1 with
exactly the three t36 drifts (media_player.py:965 BrowseMediaSource,
diagnostics.py:93 homeassistant.util.diagnostic, config_flow.py:425
missing-await) and 0 floor violations.

The fast ``test`` job and ``release.yml`` are untouched; the workflow is
unchanged in shape except the new provisioning + guard steps, and parses as
valid YAML.
2026-10-10 21:29:39 +01:00
James Mackay 824d6192a1 test(ci): D1/D2/D3 — repo-owned HA API-drift sweep with a CI job (t37)
The unit suite validates production modules against hand-written stubs, so a
call site that no longer matches real Home Assistant stays green — that is how
D1, D2 and D3 reached master. This adds the check that would have caught them.

`tests/ha_api_drift.py` (stdlib-only, documented inline) parses every module of
custom_components/nvidia_shield and resolves each Home-Assistant dependency
against an INSTALLED homeassistant:

1. every `homeassistant.*` import — module present, attribute present (or a
   submodule of that name exists, which is what `from package import submodule`
   means) — catches D1/D2 and any future move;
2. every keyword argument passed to a known HA API (HA_CALL_OWNERS: the
   ConfigFlow/ConfigEntries APIs we call) must exist in the real signature —
   catches the `options_updates` class of bug;
3. every `self.<method>()` call in a class that inherits from an HA class is
   compared with the real definition: a coroutine must be awaited and an
   awaited call must be a coroutine — catches D3.

`tests/test_ha_api_drift.py` runs it against the production tree, asserts the
sweep is not vacuous (module/import/argument/await counters), and pins the red
proof with the four pre-fix forms, plus the green proof for the fixed forms. It
skips cleanly when homeassistant is absent, so the stdlib-only jobs are
unaffected.

`.gitea/workflows/test.yml` gains the `ha-api-drift` job — the one place Home
Assistant is installed — guarded by an explicit Python >= 3.13 check (Home
Assistant 2026.x requires it) and the same PEP 668 retry the test job uses.

Red proof against the real pre-fix revisions (svn-style checkout of
`git show HEAD:…` into a temp dir, and the pre-t35 config_flow.py for the kwarg):

  media_player.py:965: broken-import: from homeassistant.components.media_player
      import BrowseMediaSource: …BrowseMediaSource does not exist
  diagnostics.py:93: broken-import: from homeassistant.util.diagnostic import
      async_redact_data: module 'homeassistant.util.diagnostic' is not importable
  config_flow.py:425: missing-await: self._async_handle_discovery_without_unique_id()
      is a coroutine on homeassistant.config_entries.ConfigFlow
  config_flow.py:700: unknown-kwarg: async_update_reload_and_abort() has no
      'options_updates' parameter (exit 1 on that tree, exit 0 on this one)

Verified: variant 1 289 passed / 2 skipped; variant 2 (androidtvremote2) 290
passed / 1 skipped; ruff clean; `UV_PYTHON=3.13 uvx --with homeassistant --with
pytest-asyncio pytest tests/test_ha_api_drift.py -q` → 4 passed (the bare form of
that command needs the pin here only because this sandbox's default Python is
3.15 and pycares has no wheel for it / no CMake to build it; CI pins the
interpreter explicitly).
2026-10-10 21:20:28 +01:00
James Mackay 60fb4f7557 fix(config_flow): D3 — await the discovery-dedup coroutine (t37)
`ConfigFlow._async_handle_discovery_without_unique_id` is an `async def` in core
(verified: `inspect.iscoroutinefunction` is True) and core awaits it. t32 called
it without `await`, so the coroutine body never ran: no `already_configured`
abort when the handler already had entries, no
`DEFAULT_DISCOVERY_UNIQUE_ID` claim, no `already_in_progress` abort — HA's
discovery dedup was dead, Python warned "coroutine ... was never awaited", and
the host-equality loop that used to sit here was the only accidental dedup.

The call is awaited and the now-unreachable host loop is gone (the real helper
is stricter: any entry for the domain, or any in-flight flow). The harness
mirrors the real contract instead of a sync no-op:

* `entity_stublib`'s hub stub makes the helper a coroutine (a sync stub would
  turn the production `await` into a TypeError);
* `test_standalone_setup`'s stub models the real semantics — a coroutine raising
  `DiscoveryDedupAbortError` (core raises AbortFlow), driven by the shared
  holder's entry list and in-progress list — and its tests now assert all three
  outcomes: awaited-and-proceeds (counter + claimed discovery unique id),
  `already_configured` when entries exist, `already_in_progress` when another
  flow runs. A fourth test collects the un-awaited-coroutine warning, and the
  prefill test in the UX suite asserts the awaited counter.

Red proof: with the pre-fix config_flow.py the repo-owned sweep reports
`config_flow.py:425: missing-await: self._async_handle_discovery_without_unique_id()`.

Verified: 289 passed, 2 skipped (D3 stage); ruff clean.
2026-10-10 21:20:13 +01:00
James Mackay bd72ae3170 fix(diagnostics): D2 — redact through homeassistant.helpers.redact (t37)
`homeassistant.util.diagnostic` does not exist — verified against a real
install: `import homeassistant.util.diagnostic` raises ModuleNotFoundError, and
the helper has always lived in `homeassistant.helpers.redact`. The import sat
inside a `try/except ImportError` that set the helper to None, so the
documented G14 pass — the one carrying `_TO_REDACT`, hosts included — silently
did nothing; only the marker scan (pairing-/key-shaped names) masked anything,
which is why the existing test's pairing-code assertion stayed green.

The helper is now imported at MODULE scope from homeassistant.helpers.redact and
applied unconditionally after the marker-name pass. A missing module is a loud
import error at integration load instead of a silent downgrade.

* the diagnostics test now asserts `adb_host` and `protocol_host` really are
  masked in the emitted payload (options and a nested data value);
* TestFaithfulHaStubs asserts the module USES the helper from that path
  (`nsh.diagnostics.async_redact_data is homeassistant.helpers.redact.async_redact_data`),
  that no production module imports the non-existent module (AST), and that the
  try/except ImportError fallback is gone;
* `entity_stublib` gains a faithful `homeassistant.helpers.redact` stub (same
  masking, same nested mapping/list recursion, same None/empty-string skip) so
  the harness meets the import the way a real install does.

Verified: 287 passed, 2 skipped (D2 stage); ruff clean. Full gate re-run on the
completed branch: 289 passed / 2 skipped and 290 passed / 1 skipped.
2026-10-10 21:20:04 +01:00
James Mackay 37b2accfda fix(media_player): D1 — browse builds real BrowseMedia nodes, not BrowseMediaSource (t37)
`homeassistant.components.media_player` has not exported `BrowseMediaSource`
for years: the class lives in `homeassistant.components.media_source.models`,
where it exists to build `media_source://<domain>` ids and is the only place a
`domain` parameter belongs (t30 G18's note was about THAT class). The local
import therefore raised ImportError, and with it every real browse failed —
while the suite stayed green because the harness stubbed a `BrowseMediaSource`
that real HA does not expose from there.

`async_browse_media` now builds the standard tree exactly as core's own
androidtv_remote/media_player.py does: `BrowseMedia` imported from
homeassistant.components.media_player at module scope, every node carrying an
explicit media_content_id/media_content_type (MediaType.APP/APPS),
`children_media_class=MediaClass.APP` on the directory, and no `domain`.
Verified against a real install: the old import raises ImportError, the old
`domain=` kwarg raises TypeError, and the new node constructs.

The harness is now faithful so this cannot hide again:
* `entity_stublib` exposes a `BrowseMedia` with the real keyword-only
  signature, `as_dict()` and `calculate_children_class()`, plus `MediaType`,
  and deliberately NOT `BrowseMediaSource` — so both the drifted import and the
  drifted kwarg fail in-suite;
* the app-directory test asserts the standard shape (classes, explicit ids and
  types, children_media_class, the serialised payload) instead of the removed
  `domain` attribute;
* TestFaithfulHaStubs pins the harness contract on the AST (no production
  module imports BrowseMediaSource) rather than on prose.

Verified: 284 passed, 2 skipped (D1 stage); ruff clean. Full gate re-run on the
completed branch: 289 passed / 2 skipped and 290 passed / 1 skipped.
2026-10-10 21:19:53 +01:00
James Mackay 5ede4fe722 test(config_flow): prove a reconfigure entry really rebuilds the protocol transport (t35 F1)
The F1 test asserted the updated options; this closes the loop on the user
outcome — after the Reconfigure pairing, __init__._build_protocol_settings
returns settings (transport ON) pointing at the paired certificate under
.storage/nvidia_shield/ and at the address that just paired. That is why no
__init__.py change was needed: the reloaded entry already rebuilds the
transport from the persisted options + shared cert paths.

Verified: uvx --with pytest-asyncio pytest tests -q → 281 passed, 1 skipped;
uvx --with pytest-asyncio --with androidtvremote2 pytest tests -q → 282 passed;
uvx ruff check custom_components tests → clean.
2026-10-10 21:05:13 +01:00
James Mackay e11dccf2c8 fix(config_flow): F1 — a discoverable later pairing path (Reconfigure) and truthful skip copy (t35 F1/F1b)
"Continue without protocol pairing" switched the transport off for good: the
options flow carries no protocol fields, so nothing in the UI could ever turn
it back on, and the skip copy promised nothing about it (t34 F1). The entry now
answers Home Assistant's Reconfigure action:

* async_step_reconfigure (audit G5) seeds the flow from the entry (title, adb
  host/port/enabled, androidtv entity for the host fallback) and re-runs the
  pairing step — the address is editable, the Shield shows a PIN, and
  _finish_reconfigure persists protocol_enabled=True plus the address that just
  paired (so the transport no longer depends on the ADB fallback), adopts the
  paired MAC when it differs, and updates + reloads the entry. HA derives the
  "reconfigure_successful" abort from the flow source. A Reconfigure visit that
  clicks skip aborts honestly with "reconfigure_skipped" (entry untouched), and
  an entry that is gone aborts with "reconfigure_entry_missing".
* the skip data_description now states the truth — the transport stays off
  until you pair it — and names the later path (the integration's entry menu >
  Reconfigure).
* a host corrected on the pairing form now DROPS the session started against the
  old address and restarts pairing there, so the typed address is authoritative
  instead of being paired against the stale one.
* the success path no longer runs _abort_if_unique_id_configured for a flow
  that already owns an entry: core looks the domain up by unique_id, so
  re-pairing an entry aborted on itself as "already_configured" (this also
  unblocked reauth). The fresh-setup path is unchanged.
* adjacent fix while wiring the same call: _finish_reauth passed
  `options_updates`, which core's async_update_reload_and_abort does not accept
  (the kwarg is `options`, REPLACE) — the reauth path would have raised
  TypeError in production. Both paths now pass `options` and the test stubs
  mirror the real signature.

Tests: tests/test_config_flow_ux.py::TestProtocolReconfigure drives reconfigure
from a protocol-disabled entry through to updated options (transport on, host
persisted, other options kept, MAC adopted, reload recorded), a corrected host
reopening the PIN step against the new address, the skip abort, and the
missing-entry abort; TestCopyContract gained the F1b copy pin. Against HEAD's
config_flow.py all four reconfigure tests fail (no async_step_reconfigure).

Verified: uvx --with pytest-asyncio pytest tests -q → 281 passed, 1 skipped;
uvx --with pytest-asyncio --with androidtvremote2 pytest tests -q → 282 passed;
uvx ruff check custom_components tests → clean.
2026-10-10 21:04:47 +01:00
James Mackay 7a0152e6b5 fix(tests): F2 — the voluptuous mirror compares fields, not marker classes (t35)
real voluptuous 0.16.0 implements ``Marker.__eq__`` as ``self.schema ==
other`` — a field-name comparison with no class check, so
``Optional("x") == Required("x")`` is True (verified against the library
source, not just the audit note). The mirror compared class + field, which
kept two dict entries for one field: a ``Required(field, default=...)`` after
``Optional(field)`` LOOKED like it applied the default, while real voluptuous
keeps the first (defaultless) key object and silently drops it — the exact
t33 F-A failure mode the mirror exists to catch, i.e. the harness could mask
the bug it was built to see.

``Marker.__eq__`` is now the library's comparison and the contract tests gained
both pins: field-only equality (with equal hashes) and the mixed-marker
re-declaration that used to be masked. Reverting only ``__eq__`` makes those
two fail, which is the red proof recorded for t35.

Verified: uvx --with pytest-asyncio pytest tests -q → 276 passed, 1 skipped;
uvx ruff check custom_components tests → clean.
2026-10-10 21:04:34 +01:00
James Mackay 186e5c1935 test(config_flow): the t33 UX suite — schema-level prefill, skip paths, copy contract, strings parity (t33)
One home for the user-reported regressions, all driven end to end against the
scoped stub harness:

- TestVoluptuousMirrorContract pins tests/voluptuous_mirror.py itself (a
  default on the declaration is visible; a later re-declaration cannot add one)
  so the harness cannot silently go blind again.
- TestDiscoveryPrefill: the discovered address is the default of BOTH host
  keys, the port default is 5555, the step announces "Discovered Shield at
  <host>", a manual/seed flow claims no discovery, and the error re-render
  keeps the prefill. Reads the rendered schema, not the stored attribute.
- TestNoDeadEndPairing: skip protocol → entry created (passive), skip ADB →
  entry created (ADB intent kept for the Pair ADB button) with the real
  async_setup_entry accepting both entries (no ConfigEntryNotReady), skip ADB
  routes on to the protocol step, both normal pair paths still create the
  entry, a host typed on a pairing form is accepted in place, and a missing
  androidtvremote2 reports protocol_unavailable instead of raising.
- TestCopyContract: no copy points at "the previous step", the no_host errors
  name the field and the discovery prefill, every error/abort key raised by
  config_flow.py exists in both copy files, and every field rendered by the
  user/adb_pair/protocol_pair/options forms has a label.
- TestStringsTranslationsParity: strings.json and translations/en.json expose
  the same key tree.
- TestPassiveSetupNeverDemandsReadiness: async_setup_entry raises nothing.

Verified: uvx --with pytest-asyncio pytest tests -q → 274 passed, 1 skipped;
uvx --with pytest-asyncio --with androidtvremote2 pytest tests -q → 275 passed;
uvx ruff check custom_components tests → clean.
2026-10-10 20:50:38 +01:00
James Mackay 08b97fde73 fix(strings): F-C — plain, self-consistent setup copy for the live-run feedback (t33)
Every user/pairing-step label and data_description rewritten in plain language,
and the fields the new pairing forms render now have labels at all (adb_host,
adb_port, skip_pairing, protocol_host, pairing_code).

No error string sends the user to "the previous step" any more: adb_no_host and
protocol_no_host name the field on the same form AND say a discovered Shield is
pre-filled automatically. The confirmation messages for both pairing steps
describe what the form actually does (including the restart-on-empty-submit
path), and protocol_unavailable covers a missing androidtvremote2.

The options-flow copy was audited in the same pass so it matches the setup
step's vocabulary ("Read state over ADB", "Show title and artist", ...), and
translations/en.json stays byte-identical to strings.json.

Verified: uvx --with pytest-asyncio pytest tests -q → 242 passed, 1 skipped;
uvx ruff check custom_components tests → clean.
2026-10-10 20:50:28 +01:00
James Mackay a6cc56feb4 fix(config_flow): F-B — neither pairing step is a dead end (t33)
protocol_pair and adb_pair now each offer "continue without pairing" (the new
CONF_SKIP_PAIRING form key), so the entry still finishes — in passive mode —
instead of stranding the user on a form with no way forward.

adb_pair: the ADB host/port live ON this form (the user step is gone), a host
typed here is authoritative, the addr the error names is therefore fixable
where the error appears, and skip routes through _async_after_pairing. ADB
stays enabled on purpose: the manager already degrades to its passive,
background-retrying state without a key, and the Pair ADB button (registered
for adb-enabled entries) then pairs later without re-adding the integration.

protocol_pair: the protocol host lives on this form too and a corrected
address restarts the pairing session against it; submitting without a code asks
the Shield for a fresh code (_async_restart_protocol_pairing) instead of
erroring; skip switches the transport off, because an unpaired protocol client
fails auth on its first connect and would only bounce the user into the reauth
flow — a fresh dead end. On reauth the skip aborts with reauth_skipped instead
of faking a repair.

_start_protocol_pairing also returns None now when build_protocol_client does
(androidtvremote2 missing) so the step reports protocol_unavailable instead of
crashing on a None client in the default CI variant.

Verified: uvx --with pytest-asyncio pytest tests -q → 242 passed, 1 skipped;
uvx ruff check custom_components tests → clean. The end-to-end skip-path and
copy tests land with the t33 UX suite commit.
2026-10-10 20:50:22 +01:00
James Mackay 73e886dafe fix(config_flow): F-A/F-A2/F-D — the discovered host and the 5555 port ride the schema KEY defaults (t33)
F-A (HIGH): _user_schema_dict re-declared vol.Optional(CONF_ADB_HOST) /
CONF_PROTOCOL_HOST inside `if "host" in defaults:` — a no-op, because
voluptuous keys compare by field name and the first (defaultless) key object
survives. The discovered address therefore never reached the form, and the
user submitted a blank host. The host keys are now declared WITH default=host
(_host_field_key), so a discovery-driven user step arrives pre-filled.

F-A2: a discovery-driven user step says the address is known
("{discovery_note}Discovered Shield at <host> — its address is already filled
in below. ") through description_placeholders, and the no_state_source
re-render keeps the prefill instead of rebuilding a blank schema.

F-D: the ADB port is declared with default=DEFAULT_ADB_PORT (5555), and a YAML
seed's port hint is carried into the form instead of being flattened to 5555.
The seed host now feeds _seed_host (same prefill, no discovery claim).

Reproduce-through-test: tests/voluptuous_mirror.py mirrors the real
voluptuous.Marker key contract (hash/eq by field name, default=UNDEFINED when
omitted), because voluptuous is not installed in either CI variant and the old
bare-key stubs could not tell a schema WITH a default from one without. The
pre-t33 discovery assertion only checked the stored attribute and stayed green
while the form rendered blank; it now reads the schema keys and fails on the
pre-fix code (8 tests reproduced red locally).

Verified: uvx --with pytest-asyncio pytest tests -q → 242 passed, 1 skipped;
uvx ruff check custom_components tests → clean.
2026-10-10 20:50:16 +01:00
James Mackay bbd98ebc3f ci: survive PEP 668 on externally managed runners — retry the test-tooling install with --break-system-packages, pin the gate tool versions
Release / release (push) Successful in 6s
Test / test (push) Successful in 21s
2026-10-10 08:34:34 +01:00
16 changed files with 2932 additions and 262 deletions

No files matched your search

+86 -2
View File
@@ -1,4 +1,4 @@
# Quality gate: lint + unit tests.
# Quality gate: lint + unit tests + the HA API-drift sweep.
#
# The test suite is stdlib-only (parsers, intents, running-apps) and runs the
# vendored ha_adb_common helpers without installing Home Assistant — see
@@ -6,6 +6,11 @@
# volume/TYPED-dict tests execute (tests/test_protocol_transport.py). Real
# firmware captures live in tests/fixtures
# (how to capture more: docs/RESEARCH_SHIELD.md §8).
#
# The `ha-api-drift` job is the ONE place Home Assistant is installed: every
# other job validates our code against hand-written stubs, which stay green
# even when a call site no longer matches the real API. That is how three
# drifts reached master (t36 D1/D2/D3) — see tests/ha_api_drift.py.
name: Test
@@ -21,10 +26,89 @@ jobs:
- uses: actions/checkout@v4
- name: Install test tooling
run: python3 -m pip install --quiet pytest pytest-asyncio ruff androidtvremote2
run: |
set -euo pipefail
# Try the plain install first: it succeeds on images whose Python is
# not externally managed (and on pip < 23, which has no PEP 668
# enforcement at all). When the runner image DOES mark its Python as
# externally managed — Debian/Ubuntu based Gitea act images do — pip
# refuses with externally-managed-environment and we retry with the
# sanctioned override. That override is safe here specifically
# because the job runs in a disposable container, and because the
# flag only exists on pip >= 23.0.1, the same version that raises
# the refusal in the first place.
#
# Tools are pinned to the versions the suite was validated against so
# a newer ruff rule set cannot redden the gate on unrelated commits;
# androidtvremote2 tracks the manifest floor so CI exercises the same
# dependency users install.
PKGS="pytest~=9.1 pytest-asyncio~=1.4 ruff~=0.17.0 'androidtvremote2>=0.2.0'"
if ! eval python3 -m pip install --quiet $PKGS; then
echo "system Python is externally managed; retrying with --break-system-packages"
eval python3 -m pip install --quiet --break-system-packages $PKGS
fi
- name: Lint
run: ruff check custom_components tests
- name: Unit tests
run: python3 -m pytest tests -q
# Resolve every Home-Assistant API use in our production modules against a
# real install: imports, keyword arguments and await/sync mismatches. The
# sweep is repo-owned (tests/ha_api_drift.py) and skips cleanly when
# Home Assistant is absent, so it is inert in the job above and live here.
ha-api-drift:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Provision the interpreter Home Assistant needs
uses: actions/setup-python@v5
with:
# Home Assistant 2026.10.1 declares Requires-Python >= 3.14.2 (PyPI
# metadata; tests/ha_api_drift.py holds our copy of that number and the
# drift suite pins it against the installed distribution), so the
# runner's own python3 cannot install the version this gate must run
# against. This resolves the newest 3.14.x patch; the guard below
# refuses anything still short of the floor rather than letting the
# install fail halfway through.
python-version: "3.14"
- name: Check the interpreter satisfies Home Assistant's floor
run: |
set -euo pipefail
# The floor comes from tests/ha_api_drift.py — one source of truth for
# this guard and the metadata pin in tests/test_ha_api_drift.py.
python3 - <<'PY'
import sys
from pathlib import Path
sys.path.insert(0, str(Path("tests").resolve()))
from ha_api_drift import HOMEASSISTANT_PYTHON_FLOOR as FLOOR # noqa: E402
needed = ".".join(str(part) for part in FLOOR)
if sys.version_info < FLOOR:
print(
f"Home Assistant needs Python >= {needed}, but this job runs "
f"{sys.version.split()[0]}."
)
print(f"Fix: raise actions/setup-python in this job to a >= {needed} build.")
raise SystemExit(1)
print(f"interpreter ok: Python {sys.version.split()[0]} satisfies >= {needed}")
PY
- name: Install Home Assistant + test tooling
run: |
set -euo pipefail
# Same PEP 668 dance as the test job (see the note there). The
# version floor tracks what we validate against; a resolver failure
# here is a real signal (Python too old), not something to swallow.
PKGS="pytest~=9.1 pytest-asyncio~=1.4 'homeassistant>=2026.10'"
if ! eval python3 -m pip install --quiet $PKGS; then
echo "system Python is externally managed; retrying with --break-system-packages"
eval python3 -m pip install --quiet --break-system-packages $PKGS
fi
- name: HA API-drift sweep
run: python3 -m pytest tests/test_ha_api_drift.py -q
+18
View File
@@ -182,6 +182,24 @@ device page in HA.
Requirements: Settings > Developer options > Network debugging enabled on the Shield.
### Discovery (zeroconf)
Home Assistant can discover a Shield through the `_androidtvremote2._tcp`
service, and the discovered address is pre-filled on the setup form. The dedup
rules are Home Assistant's own, and they differ with what the discovery carries:
- **with** the device MAC, a duplicate aborts as "already configured", and an
existing entry's hosts are updated when the address changed;
- **without** a MAC, the flow follows HA's stricter
`_async_handle_discovery_without_unique_id` contract: **any** existing entry
for this integration aborts the discovery, and only one discovery flow runs at
a time.
So a second (or third) Shield that advertises no MAC has to be added **manually**
— Settings > Devices & Services > Add Integration > "NVIDIA Shield" — rather than
by discovery. That is a deliberate, tested contract (t37 D3), not a bug: the
helper the flow calls is a coroutine and its dedup only runs when awaited.
### Options
- **Jellyfin Media Player** — auto-detected or manually selected
+302 -65
View File
@@ -32,6 +32,7 @@ from .const import (
CONF_PROTOCOL_HOST,
CONF_RECEIVER_ENTITY,
CONF_RECEIVER_INPUT,
CONF_SKIP_PAIRING,
DEFAULT_ADB_PORT,
DEFAULT_NAME,
DOMAIN,
@@ -154,12 +155,31 @@ def _app_map_from_text(
return app_map
def _host_field_key(field: str, host: str):
"""The host-field KEY, carrying the known address as its default.
t33 F-A: the default MUST ride on this key. Re-declaring a bare
``vol.Optional(field)`` afterwards is a no-op that silently drops it —
voluptuous keys compare by field name, so the first (defaultless) key
object survives and the form renders blank.
"""
if host:
return vol.Optional(field, default=host)
return vol.Optional(field)
def _user_schema_dict(sel, defaults: dict | None = None) -> dict:
"""User-step schema — the androidtv entity is optional (standalone).
``defaults`` pre-fills name/hosts from a zeroconf discovery event.
``defaults`` pre-fills name/hosts from a zeroconf discovery event (or a
YAML seed hint); the ADB port is pre-filled with the standard 5555 so it
reads as an advanced setting rather than required knowledge (t33 F-D).
"""
defaults = defaults or {}
host = str(defaults.get("host") or "")
port = defaults.get("port")
if not isinstance(port, int):
port = DEFAULT_ADB_PORT
schema_dict = {
vol.Required("name", default=defaults.get("name", DEFAULT_NAME)): str,
vol.Optional(CONF_ANDROIDTV_ENTITY): sel["androidtv"],
@@ -174,14 +194,13 @@ def _user_schema_dict(sel, defaults: dict | None = None) -> dict:
schema_dict[vol.Optional(CONF_POWER_ENTITY)] = sel["power"]
schema_dict[vol.Optional(CONF_ADB_ENABLED)] = bool
schema_dict[vol.Optional(CONF_ADB_STORE_APP_INFO)] = bool
schema_dict[vol.Optional(CONF_ADB_HOST)] = str
schema_dict[vol.Optional(CONF_ADB_PORT)] = int
# the code's own fallback for this flag is True (async_step_user) — the
# rendered checkbox must agree with it
schema_dict[vol.Optional(CONF_ADB_STORE_APP_INFO, default=True)] = bool
schema_dict[_host_field_key(CONF_ADB_HOST, host)] = str
schema_dict[vol.Optional(CONF_ADB_PORT, default=port)] = int
schema_dict[vol.Optional(CONF_PROTOCOL_ENABLED)] = bool
schema_dict[vol.Optional(CONF_PROTOCOL_HOST)] = str
if "host" in defaults:
schema_dict[vol.Optional(CONF_ADB_HOST)] = str
schema_dict[vol.Optional(CONF_PROTOCOL_HOST)] = str
schema_dict[_host_field_key(CONF_PROTOCOL_HOST, host)] = str
return schema_dict
@@ -212,9 +231,14 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._user_data = {}
self._protocol_client = None
self._reauth_entry = None
# audit G5 / t35 F1: the entry a "Reconfigure" visit targets
self._reconfigure_entry = None
# Zeroconf discovery state (P4-1)
self._discovered_host: str | None = None
self._discovered_name: str | None = None
# YAML-seed prefill hint (t33 F-A): feeds the same user-step prefill as
# a discovery, without claiming the address was discovered.
self._seed_host: str | None = None
async def async_step_user(self, user_input=None):
"""Step 1: Basic setup.
@@ -236,12 +260,7 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
if not has_source:
errors["base"] = "no_state_source"
sel = _build_selectors()
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(_user_schema_dict(sel)),
errors=errors,
)
return self._show_user_form(errors)
# t29 G2: the androidtv entity_id is a renameable reference —
# never a unique_id. A sqlite/registry MAC is preferred (the
@@ -301,21 +320,48 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self._create_entry()
return self._show_user_form(errors)
sel = _build_selectors()
def _user_defaults(self) -> dict:
"""Defaults for the user step: whatever setup already knows.
defaults = {}
if self._discovered_host:
A zeroconf discovery (or a YAML seed hint) supplies the address; the
port keeps a seed's hint instead of being flattened to the standard
value. This is the ONLY place the known address enters the form — the
schema keys carry these values as their defaults (t33 F-A).
"""
defaults = {
"name": self._discovered_name or DEFAULT_NAME,
"host": self._discovered_host,
"name": self._discovered_name
or self._user_data.get("name")
or DEFAULT_NAME,
}
schema_dict = _user_schema_dict(sel, defaults)
host = self._discovered_host or self._seed_host
if host:
defaults["host"] = host
port = self._user_data.get(CONF_ADB_PORT)
if isinstance(port, int):
defaults["port"] = port
return defaults
def _show_user_form(self, errors=None):
"""Render the user step with the known address actually applied.
t33 F-A2: a discovery-driven form says so in its description, so a
pre-filled host can never be misread as "unknown".
"""
note = ""
if self._discovered_host:
note = (
f"Discovered Shield at {self._discovered_host} — its address "
"is already filled in below. "
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(schema_dict),
errors=errors,
data_schema=vol.Schema(
_user_schema_dict(_build_selectors(), self._user_defaults())
),
errors=errors or {},
description_placeholders={"discovery_note": note},
)
async def async_step_zeroconf(self, discovery_info):
@@ -371,18 +417,17 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_abort(reason="already_configured")
self._abort_if_unique_id_configured()
else:
# t32 F2 (review rider): MAC-less discovery cannot carry a
# unique_id — HA's helper records the uniqueness-less discovery
# (the same discovery context can't stack duplicates) and the
# host guard keeps a second discovery of an already-configured
# host from stacking another prefill.
self._async_handle_discovery_without_unique_id()
for entry in self._async_current_entries():
existing_host = entry.options.get(
CONF_PROTOCOL_HOST
) or entry.options.get(CONF_ADB_HOST)
if existing_host and existing_host == host:
return self.async_abort(reason="already_configured")
# t32 F2 / t37 D3: MAC-less discovery cannot carry a unique_id, so
# HA's own helper owns the dedup — and it MUST be awaited. It is a
# coroutine that aborts `already_configured` when the handler
# already has entries, claims DEFAULT_DISCOVERY_UNIQUE_ID, and
# aborts `already_in_progress` when another flow for this handler
# is running. t32 shipped the call without the await, so none of
# that ran ("coroutine was never awaited") and the host-equality
# guard that used to sit here was the accidental dedup; the real
# helper is stricter (any entry for the domain, any in-flight
# flow), so nothing is left to do here.
await self._async_handle_discovery_without_unique_id()
# First-time setup: pre-filled user step
return await self.async_step_user()
@@ -420,11 +465,21 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Protocol host: explicit override, else the ADB host fallback."""
return self._user_data.get(CONF_PROTOCOL_HOST) or self._resolve_adb_host()
def _discard_protocol_client(self) -> None:
"""Drop a pairing client whose session became void (host corrected)."""
client = self._protocol_client
self._protocol_client = None
if client is not None:
with contextlib.suppress(Exception):
client.disconnect()
async def _start_protocol_pairing(self):
"""Build a throwaway protocol client and make the Shield show its PIN.
Certificates live under ``.storage/<domain>/`` (core androidtv_remote
helpers.py pattern) and are shared between pairing and runtime.
helpers.py pattern) and are shared between pairing and runtime. Returns
None when no host is resolvable or ``androidtvremote2`` is missing —
the caller turns that into an error key, never a crash.
"""
host = self._protocol_host()
if not host:
@@ -438,32 +493,116 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
certfile=certfile,
keyfile=keyfile,
)
if client is None:
# build_protocol_client already logged the missing library;
# pairing cannot start, and the form still offers the skip path
return None
await client.async_generate_cert_if_missing()
await client.async_start_pairing()
return client
async def _async_begin_protocol_pairing(self):
"""Start a PIN pairing session: ``(client, None)`` or ``(None, error)``."""
if not self._protocol_host():
return None, "protocol_no_host"
client = await self._start_protocol_pairing()
if client is None:
return None, "protocol_unavailable"
return client, None
def _show_protocol_pair_form(self, errors=None):
"""protocol_pair form: the PIN, the address, and the skip escape.
t33 F-B/F-C: the host field lives on THIS form, so an unresolvable
address is fixable where the error is shown (the user step is gone),
and the skip option keeps the step from being a dead end.
"""
return self.async_show_form(
step_id="protocol_pair",
data_schema=vol.Schema(
{
vol.Required(CONF_PAIRING_CODE): str,
_host_field_key(CONF_PROTOCOL_HOST, self._protocol_host() or ""): str,
vol.Optional(CONF_SKIP_PAIRING, default=False): bool,
}
),
errors=errors or {},
)
async def _async_skip_protocol_pairing(self):
"""Continue without the protocol transport (t33 F-B, passive mode).
The transport is switched OFF rather than left half-paired: an
unpaired protocol client fails auth on its first connect and would
only bounce the user into the reauth flow — a fresh dead end.
"""
self._discard_protocol_client()
self._user_data[CONF_PROTOCOL_ENABLED] = False
if self._reauth_entry is not None:
# an existing entry cannot be re-created: leave it untouched and
# say so instead of pretending the re-pair worked
self._reauth_entry = None
return self.async_abort(reason="reauth_skipped")
if self._reconfigure_entry is not None:
# same for a Reconfigure visit: nothing changed, the entry keeps
# its options and can be reconfigured again at any time
self._reconfigure_entry = None
return self.async_abort(reason="reconfigure_skipped")
return self._create_entry()
async def _async_restart_protocol_pairing(self, errors: dict):
"""Start a fresh pairing session and render the form for its code.
Called on the first visit, after a host correction, and when the user
submits without a code: each is a request for the Shield to show a new
code, and none of them may leave the step stuck.
"""
self._discard_protocol_client()
client, error = await self._async_begin_protocol_pairing()
if error:
errors["base"] = error
else:
self._protocol_client = client
return self._show_protocol_pair_form(errors)
async def async_step_protocol_pair(self, user_input=None):
"""Pair with the remote protocol — the PIN is displayed on the Shield."""
"""Pair with the remote protocol — the PIN is displayed on the Shield.
Never a dead end (t33 F-B): a corrected address restarts pairing right
here, submitting without a code asks for a fresh one, and
``skip_pairing`` finishes the entry in passive mode.
"""
errors = {}
if user_input is None:
# First visit: proactively start pairing so the PIN shows on the
# Shield while the form is open.
client = await self._start_protocol_pairing()
if client is None:
errors["base"] = "protocol_no_host"
else:
self._protocol_client = client
else:
return await self._async_restart_protocol_pairing(errors)
if user_input.get(CONF_SKIP_PAIRING):
return await self._async_skip_protocol_pairing()
typed_host = str(user_input.get(CONF_PROTOCOL_HOST) or "").strip()
if typed_host and typed_host != (self._protocol_host() or ""):
# A host typed HERE is authoritative — the user step is gone, so
# the session started against the old address is dropped and the
# Shield is asked for a NEW code at the corrected address (t35 F1).
self._user_data[CONF_PROTOCOL_HOST] = typed_host
self._discard_protocol_client()
code = str(user_input.get(CONF_PAIRING_CODE, "") or "").strip()
if self._protocol_client is None or not code:
# No live session, or no code typed: (re)start pairing so the
# Shield shows a code for the address currently on the form.
return await self._async_restart_protocol_pairing(errors)
client = self._protocol_client
code = user_input.get(CONF_PAIRING_CODE, "")
if client is None or not code:
errors["base"] = "protocol_invalid_code"
else:
try:
await client.async_finish_pairing(code)
except Exception:
errors["base"] = "protocol_pair_failed"
else:
return self._show_protocol_pair_form(errors)
try:
_name, mac = await client.async_get_name_and_mac()
except Exception:
@@ -471,19 +610,20 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
with contextlib.suppress(Exception):
client.disconnect()
self._protocol_client = None
if mac:
target = self._reauth_entry or self._reconfigure_entry
if mac and mac != getattr(target, "unique_id", None):
# A MAC that is already configured must never be adopted silently.
# A reauth/reconfigure flow already OWNS an entry, and
# _abort_if_unique_id_configured looks the domain up by unique_id —
# it would abort on that very entry when the MAC is unchanged
# (t35 F1: re-pairing an entry used to abort as "already_configured").
await self.async_set_unique_id(mac)
self._abort_if_unique_id_configured()
if self._reauth_entry is not None:
return self._finish_reauth()
if self._reconfigure_entry is not None:
return self._finish_reconfigure(mac)
return self._create_entry()
return self.async_show_form(
step_id="protocol_pair",
data_schema=vol.Schema({
vol.Required(CONF_PAIRING_CODE): str,
}),
errors=errors,
)
async def async_step_import(self, data):
"""Seed an interactive flow from YAML (t30 U1, seed-only ruling).
@@ -530,7 +670,10 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
# protocol prefill hints ONLY (pairing stays interactive)
CONF_PROTOCOL_HOST: block.get(CONF_PROTOCOL_HOST, ""),
}
self._discovered_host = self._user_data[CONF_PROTOCOL_HOST] or self._user_data[
# t33 F-A: the seed host feeds the same user-step prefill as a
# zeroconf discovery, but WITHOUT the discovery announcement — the
# address came from configuration.yaml, not from the network.
self._seed_host = self._user_data[CONF_PROTOCOL_HOST] or self._user_data[
CONF_ADB_HOST
]
return await self.async_step_user(None)
@@ -560,17 +703,68 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
step_id="reauth_confirm", data_schema=vol.Schema({})
)
async def async_step_reconfigure(self, user_input=None):
"""Re-run protocol pairing for an existing entry (audit G5, t35 F1).
Home Assistant offers this as the entry's "Reconfigure" action — the
discoverable way back for an entry whose protocol transport is off
(typically the "continue without protocol pairing" outcome): the
address is editable, the Shield shows a PIN, and the entry's options
are updated + reloaded.
"""
entry = self._reconfigure_entry or self._get_reconfigure_entry()
if entry is None:
# HA always links the entry; a missing one gets a plain abort
# instead of an exception
return self.async_abort(reason="reconfigure_entry_missing")
self._reconfigure_entry = entry
# Seed the flow from the entry so the pairing step offers the address
# the entry already knows (and _create_entry is never reached here).
self._user_data = {
"name": entry.title,
CONF_ANDROIDTV_ENTITY: entry.data.get(CONF_ANDROIDTV_ENTITY, ""),
CONF_ADB_ENABLED: entry.options.get(CONF_ADB_ENABLED, False),
CONF_ADB_HOST: entry.options.get(CONF_ADB_HOST, ""),
CONF_ADB_PORT: entry.options.get(CONF_ADB_PORT, DEFAULT_ADB_PORT),
CONF_PROTOCOL_ENABLED: True,
CONF_PROTOCOL_HOST: entry.options.get(CONF_PROTOCOL_HOST, ""),
}
return await self.async_step_protocol_pair()
def _finish_reconfigure(self, mac: str | None):
"""Reconfigure pairing done: update + reload + abort (G3a contract).
The protocol host persists as the address that just paired, so the
entry no longer depends on the ADB-host fallback for its transport.
HA derives the abort reason ("reconfigure_successful") from the flow
source, so this needs no integration string.
"""
entry = self._reconfigure_entry or self._get_reconfigure_entry()
updates: dict = {
"options": {
**entry.options,
CONF_PROTOCOL_ENABLED: True,
CONF_PROTOCOL_HOST: self._protocol_host() or "",
},
}
if mac:
# a swapped device keeps the entry's identity honest
updates["unique_id"] = mac
self._reconfigure_entry = None
return self.async_update_reload_and_abort(entry, **updates)
def _finish_reauth(self):
"""Reauth pairing done: update + reload + abort via the current API (G3a).
The deprecated update-entry + schedule-reload boomerang listener is
gone (2026.12 treats it as an error); the current config-entries
contract is async_update_reload_and_abort with _get_reauth_entry.
Core's kwarg is `options` (REPLACE) — there is no `options_updates`.
"""
entry = self._reauth_entry or self._get_reauth_entry()
self.async_update_reload_and_abort(
entry,
options_updates={
options={
**entry.options,
CONF_PROTOCOL_ENABLED: True,
CONF_PROTOCOL_HOST: self._user_data.get(CONF_PROTOCOL_HOST, ""),
@@ -585,14 +779,61 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.hass, self._user_data.get(CONF_ANDROIDTV_ENTITY)
)
def _show_adb_pair_form(self, errors=None):
"""adb_pair form: the address/port (fixable here) and the skip escape.
t33 F-B/F-C/F-D: the fields live on THIS form — the user step is gone
— so a missing address is fixable where the error appears, the
standard 5555 is pre-filled, and the skip option ends the dead end.
"""
return self.async_show_form(
step_id="adb_pair",
data_schema=vol.Schema(
{
_host_field_key(CONF_ADB_HOST, self._resolve_adb_host() or ""): str,
vol.Optional(
CONF_ADB_PORT,
default=self._user_data.get(
CONF_ADB_PORT, DEFAULT_ADB_PORT
),
): int,
vol.Optional(CONF_SKIP_PAIRING, default=False): bool,
}
),
errors=errors or {},
)
async def async_step_adb_pair(self, user_input=None):
"""Step 2: ADB pairing. User clicks Submit after approving on the Shield."""
"""Step 2: ADB pairing. Approve the prompt on the Shield, then Submit.
Never a dead end (t33 F-B): the host/port are editable on the form, a
missing address is reported against the field above it, and the skip
option finishes the entry in passive mode.
"""
errors = {}
if user_input is not None:
host = self._resolve_adb_host()
port = self._user_data.get(CONF_ADB_PORT, DEFAULT_ADB_PORT)
if user_input.get(CONF_SKIP_PAIRING):
# Continue WITHOUT pairing. ADB stays enabled on purpose: the
# manager already degrades to its passive, background-retrying
# state when no key exists, and the Pair ADB button (only
# registered for adb-enabled entries) pairs later without
# re-adding the integration.
return await self._async_after_pairing()
typed_host = str(user_input.get(CONF_ADB_HOST) or "").strip()
if typed_host:
# a host typed HERE is authoritative (the user step is gone)
self._user_data[CONF_ADB_HOST] = typed_host
port = user_input.get(CONF_ADB_PORT, DEFAULT_ADB_PORT)
try:
port = int(port)
except (TypeError, ValueError):
port = DEFAULT_ADB_PORT
self._user_data[CONF_ADB_PORT] = port
host = self._resolve_adb_host()
if not host:
errors["base"] = "adb_no_host"
else:
@@ -608,11 +849,7 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
else:
errors["base"] = "adb_failed"
return self.async_show_form(
step_id="adb_pair",
data_schema=vol.Schema({}),
errors=errors,
)
return self._show_adb_pair_form(errors)
@staticmethod
@callback
+4
View File
@@ -26,6 +26,10 @@ CONF_ADB_PORT = "adb_port"
CONF_PROTOCOL_ENABLED = "protocol_enabled"
CONF_PROTOCOL_HOST = "protocol_host"
CONF_PAIRING_CODE = "pairing_code"
# t33 F-B: the "continue without pairing" escape offered on BOTH pairing
# steps — the entry still finishes (passive mode) instead of stranding the
# user on a form with no way forward.
CONF_SKIP_PAIRING = "skip_pairing"
DEFAULT_PROTOCOL_API_PORT = 6466
DEFAULT_PROTOCOL_PAIR_PORT = 6467
@@ -16,6 +16,7 @@ from dataclasses import asdict, is_dataclass
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.redact import async_redact_data
from .connection import ShieldConnectionManager
@@ -87,15 +88,13 @@ async def async_get_config_entry_diagnostics(
"is_on": manager.is_on,
},
}
# G14 (t30): the documented helper is the primary redaction path with
# an explicit field list; the marker scan above runs defensively after.
try:
from homeassistant.util.diagnostic import async_redact_data
except ImportError:
async_redact_data = None
if async_redact_data is not None:
payload = async_redact_data(payload, _TO_REDACT)
return _redact(payload)
# G14 (t30) + t37 D2: the documented helper is the primary redaction path
# with an explicit field list (the hosts included), and the marker scan
# above runs defensively after it. It is imported at MODULE scope from
# homeassistant.helpers.redact — the real module; the old
# `homeassistant.util.diagnostic` never existed, and the swallowed
# ImportError turned the whole pass into a silent no-op.
return _redact(async_redact_data(payload, _TO_REDACT))
__all__ = ["async_get_config_entry_diagnostics"]
+20 -13
View File
@@ -26,10 +26,13 @@ import time
import voluptuous as vol
from homeassistant.components.media_player import (
BrowseMedia,
MediaClass,
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaPlayerState,
MediaType,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_PAUSED, STATE_PLAYING
@@ -961,36 +964,40 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
# ── BrowseMedia app directory (P4-2c) ──
async def async_browse_media(self, media_content_type=None, media_content_id=None):
"""Root lists the apps map (or the curated default) → play_media(app)."""
from homeassistant.components.media_player import (
BrowseMediaSource,
MediaClass,
)
"""Root lists the apps map (or the curated default) → play_media(app).
t37 D1: the standard media_player tree, exactly as core's own
androidtv_remote/media_player.py builds it — ``BrowseMedia`` from
homeassistant.components.media_player, every node carrying an explicit
media_content_id/media_content_type, and the directory naming the
class of its children. ``BrowseMediaSource`` only ever existed in
homeassistant.components.media_source.models (t30 G18's domain
parameter is ITS contract: it builds ``media_source://<domain>`` ids),
so an app-directory node must never claim one — and the import above
removed a name media_player has not exported for a long time, which
made every real browse raise.
"""
app_map = self._entry.options.get(CONF_APP_NAME_MAP) or DEFAULT_APP_DIRECTORY
# G18 (2026-05-20 blog): BrowseMediaSource must carry the domain
# parameter so the frontend can attribute the media tree.
children = [
BrowseMediaSource(
BrowseMedia(
title=name,
media_class=MediaClass.APP,
media_content_type="app",
media_content_type=MediaType.APP,
media_content_id=pkg,
can_play=True,
can_expand=False,
domain=DOMAIN,
)
for pkg, name in app_map.items()
]
return BrowseMediaSource(
return BrowseMedia(
title="Apps",
media_class=MediaClass.DIRECTORY,
media_content_type="apps",
media_content_type=MediaType.APPS,
media_content_id="apps",
children_media_class=MediaClass.APP,
can_play=False,
can_expand=True,
children=children,
domain=DOMAIN,
)
# ── play_media (§7): channel digits, app/url launch ──
+82 -62
View File
@@ -3,65 +3,85 @@
"step": {
"user": {
"title": "NVIDIA Shield",
"description": "Set up your NVIDIA Shield. Link the Android TV remote, Jellyfin, receiver, and power monitoring.",
"description": "{discovery_note}Set up your NVIDIA Shield. Everything here is optional except one state source \u2014 an Android TV Remote entity, the Nvidia Remote Protocol, or ADB. Link the devices and services you want to use.",
"data": {
"name": "Device Name",
"androidtv_entity": "The media_player entity from the Android TV Remote integration. Optional for standalone setups that pair with the Nvidia Remote Protocol instead.",
"jellyfin_entity": "Jellyfin Media Player",
"name": "Name",
"androidtv_entity": "Android TV Remote entity",
"jellyfin_entity": "Jellyfin media player",
"receiver_entity": "Receiver",
"receiver_input": "Receiver Input",
"power_entity": "Power Switch",
"adb_enabled": "Enable ADB Media Session",
"adb_store_app_info": "Show App Media Info",
"adb_host": "ADB Host",
"adb_port": "ADB Port",
"protocol_enabled": "Pair with Nvidia Remote Protocol",
"protocol_host": "Protocol Host"
"receiver_input": "Receiver input",
"power_entity": "Power switch",
"adb_enabled": "Read state over ADB",
"adb_store_app_info": "Show title and artist",
"adb_host": "ADB host (IP address)",
"adb_port": "ADB port",
"protocol_enabled": "Add the Nvidia Remote Protocol",
"protocol_host": "Protocol host (IP address)"
},
"data_description": {
"androidtv_entity": "The media_player entity from the Android TV Remote integration.",
"jellyfin_entity": "The Jellyfin media_player for this device. Leave blank to auto-detect.",
"receiver_entity": "The receiver this device is connected to.",
"receiver_input": "Which input on the receiver (e.g. 'Shield').",
"power_entity": "Smart plug switch for reliable on/off detection.",
"adb_enabled": "Use ADB to read what's playing from any app (NewPipe, Spotify, etc.). Requires ADB debugging enabled on the Shield.",
"adb_store_app_info": "Show title and artist from non-Jellyfin apps. Disable to only show the app name (privacy mode).",
"adb_host": "Override the ADB host IP. Leave blank to auto-detect from the Android TV entity \u2014 pre-filled automatically when the Shield was discovered.",
"adb_port": "ADB port number. Default is 5555.",
"protocol_enabled": "Adds the all-in-one remote-protocol transport (power/availability/app push and volume readout). Adds a PIN pairing step; the Shield shows the code.",
"protocol_host": "Override the protocol host IP. Leave blank to auto-detect (same fallback as the ADB host) \u2014 pre-filled automatically on discovery."
"androidtv_entity": "The media_player entity from the Android TV Remote integration. Leave blank if you use the Nvidia Remote Protocol or ADB instead.",
"jellyfin_entity": "The Jellyfin media_player for this Shield. Leave blank to detect it automatically.",
"receiver_entity": "The receiver or soundbar this Shield plays through.",
"receiver_input": "The receiver input the Shield uses, for example 'Shield'. Leave blank if you are not sure.",
"power_entity": "A smart plug switch, if you want reliable on/off detection.",
"adb_enabled": "Reads now-playing information over ADB from any app. Requires Developer options > Network debugging on the Shield.",
"adb_store_app_info": "Shows the title and artist from non-Jellyfin apps. Turn off to show only the app name (privacy mode).",
"adb_host": "The Shield's IP address. Pre-filled when the Shield was discovered, or when an Android TV Remote entity is set \u2014 change it only if you need a different address.",
"adb_port": "The ADB port. The standard value is 5555 \u2014 change it only if ADB uses a custom port on your network.",
"protocol_enabled": "Adds the all-in-one remote-protocol transport (power, availability, app push and volume readout). It needs a one-time PIN pairing right after this step.",
"protocol_host": "The Shield's IP address for the remote protocol. Pre-filled when the Shield was discovered \u2014 leave blank to use the ADB host."
}
},
"adb_pair": {
"title": "Pair ADB",
"description": "On the Shield, go to Settings > Developer options and enable Network debugging. Then click Submit. You will see a prompt on the Shield to allow the connection -- approve it. If the connection fails, click Submit to try again."
"title": "Allow ADB access",
"description": "On the Shield, open Settings > System > Developer options and turn on Network debugging (it may be called ADB debugging). Then select Submit \u2014 the Shield shows an Allow prompt; approve it. Nothing happened? Check the address below and select Submit again. Not now? Turn on 'Continue without ADB pairing' and select Submit.",
"data": {
"adb_host": "ADB host (IP address)",
"adb_port": "ADB port",
"skip_pairing": "Continue without ADB pairing"
},
"data_description": {
"adb_host": "The Shield's IP address. Pre-filled when the Shield was discovered \u2014 change it only if the Shield moved or you gave it a different address.",
"adb_port": "The standard ADB port is 5555.",
"skip_pairing": "Finishes setup now and leaves ADB unpaired. The Pair ADB button stays available on the device page, so you can pair later."
}
},
"protocol_pair": {
"title": "Pair Remote Protocol",
"description": "The Shield is now showing a pairing code. Enter it here. Click Submit to confirm.",
"title": "Pair the remote protocol",
"description": "While this step is open the Shield shows a pairing code. Type it below and select Submit. If no code appears on the Shield, check the IP address below and select Submit again \u2014 pairing restarts and the Shield shows a new code. To set up without the remote protocol, turn on 'Continue without protocol pairing'.",
"data": {
"pairing_code": "Pairing code shown on the Shield"
"pairing_code": "Pairing code shown on the Shield",
"protocol_host": "Protocol host (IP address)",
"skip_pairing": "Continue without protocol pairing"
},
"data_description": {
"pairing_code": "The code the Shield displays while this step is open.",
"protocol_host": "The Shield's IP address. Pre-filled when the Shield was discovered \u2014 change it if the Shield moved.",
"skip_pairing": "Finishes setup with the remote protocol off \u2014 it stays off until you pair it. Pair later from this integration's entry menu: choose Reconfigure."
}
},
"reauth_confirm": {
"title": "Re-pair Nvidia Shield",
"description": "The Shield reset to factory settings, so the saved certificate is no longer trusted. A pairing code will be shown on the Shield."
"description": "The Shield no longer trusts the saved pairing certificate, usually after a factory reset. Select Submit to pair again \u2014 the Shield then shows a code."
}
},
"error": {
"adb_failed": "Could not connect via ADB. Make sure Network debugging is enabled on the Shield and you approved the connection prompt. Click Submit to try again.",
"adb_no_host": "Could not determine the Shield's IP address. Set the ADB Host field on the previous step.",
"protocol_no_host": "Could not determine the Shield's IP address. Set the Protocol Host or ADB Host field.",
"protocol_invalid_code": "Pairing has not started or the code is empty. Click Submit to retry.",
"protocol_pair_failed": "The Shield rejected that pairing code. Click Submit to try again.",
"no_state_source": "No state source selected. Provide at least one of: the Android TV Remote entity, the Nvidia Remote Protocol pairing, or ADB media-session mode (enable ADB and set the ADB Host).",
"invalid_app_map": "Each line must be package=name with a non-empty package (no spaces) and a non-empty name \u2014 e.g. ar.tvplayer.tv=TiviMate.",
"adb_failed": "The Shield did not accept the ADB connection. Make sure Network debugging is on, approve the Allow prompt on the Shield, then select Submit to try again.",
"adb_no_host": "No IP address for the Shield yet. Fill in the ADB host field above (a discovered Shield, or one linked to an Android TV Remote entity, is pre-filled automatically).",
"protocol_no_host": "No IP address for the Shield yet. Fill in the protocol host field above (a discovered Shield, or one linked to an Android TV Remote entity, is pre-filled automatically).",
"protocol_pair_failed": "The Shield rejected that pairing code. Check the code on the Shield and select Submit to try again.",
"protocol_unavailable": "The androidtvremote2 library is not installed, so the remote protocol cannot pair. Install it and restart Home Assistant, or turn on 'Continue without protocol pairing' to finish setup anyway.",
"no_state_source": "No state source selected. Choose at least one: an Android TV Remote entity, 'Add the Nvidia Remote Protocol', or 'Read state over ADB'.",
"invalid_app_map": "Each line must be package=name with a non-empty package (no spaces) and a non-empty name \u2014 for example ar.tvplayer.tv=TiviMate.",
"invalid_digit_delay": "The digit delay must be between 0.05 and 1.0 seconds."
},
"abort": {
"already_configured": "This Shield entity is already configured.",
"already_configured": "This Shield is already set up.",
"reauth_successful": "Re-paired successfully.",
"reauth_skipped": "Pairing skipped. The saved certificate is still not trusted, so the remote protocol stays unavailable until you pair again.",
"reauth_entry_missing": "This entry no longer exists.",
"reconfigure_skipped": "Nothing changed \u2014 the remote protocol stays off until you pair it. Choose Reconfigure from this integration's entry menu when you are ready.",
"reconfigure_entry_missing": "This entry no longer exists, so there is nothing to reconfigure.",
"zeroconf_no_host": "The discovery event carried no usable address.",
"yaml_seed_only": {},
"yaml_seed": {
@@ -73,37 +93,37 @@
"options": {
"step": {
"init": {
"title": "NVIDIA Shield Settings",
"description": "Configure linked devices and monitoring.",
"title": "NVIDIA Shield settings",
"description": "Settings for this Shield. Saving reloads the integration so the changes take effect.",
"data": {
"jellyfin_entity": "Jellyfin Media Player",
"jellyfin_entity": "Jellyfin media player",
"receiver_entity": "Receiver",
"receiver_input": "Receiver Input",
"power_entity": "Power Switch",
"adb_enabled": "Enable ADB Media Session",
"adb_store_app_info": "Show App Media Info",
"adb_host": "ADB Host",
"adb_port": "ADB Port",
"app_name_map": "Apps name/icon map",
"adb_command_enabled": "Enable raw ADB shell service",
"receiver_input": "Receiver input",
"power_entity": "Power switch",
"adb_enabled": "Read state over ADB",
"adb_store_app_info": "Show title and artist",
"adb_host": "ADB host (IP address)",
"adb_port": "ADB port",
"app_name_map": "App names and icons",
"adb_command_enabled": "Enable the adb_command service",
"channel_digit_delay": "Channel digit delay (seconds)",
"channel_send_enter": "Send ENTER after channel digits",
"logcat_events": "Live events push (logcat)"
"logcat_events": "Instant app detection (logcat stream)"
},
"data_description": {
"jellyfin_entity": "The Jellyfin media_player for this device. Session stream info is read from the HA Jellyfin integration's existing coordinator \u2014 no extra config needed.",
"receiver_entity": "The receiver this device is connected to.",
"receiver_input": "Which input on the receiver.",
"power_entity": "Smart plug switch for reliable on/off detection.",
"adb_enabled": "Use ADB to read now-playing from any app. Requires ADB debugging on the Shield.",
"adb_store_app_info": "Show title and artist from non-Jellyfin apps. Disable for privacy.",
"adb_host": "IP address for ADB connection. Leave blank to auto-detect from the Android TV Remote integration.",
"adb_port": "ADB port (default 5555).",
"app_name_map": "One entry per line, package=name (e.g. ar.tvplayer.tv=TiviMate). The friendly name resolves after the Android TV Remote entity and the protocol push; icons follow the media image lookup by name. Leave blank to clear.",
"adb_command_enabled": "Adds the nvidia_shield.adb_command service (raw ADB shell over the persistent connection). Security-relevant: leave off unless you need it \u2014 it is the same opt-in posture as the core androidtv integration.",
"channel_digit_delay": "Pause between the channel digits entered via play_media(channel) \u2014 0.05 to 1.0 seconds (default 0.1, matching the validated behavior).",
"channel_send_enter": "Optionally presses ENTER after the last digit (protocol key, then the ADB keyevent fallback). Off by default \u2014 matches TiviMate's validated behavior.",
"logcat_events": "Instant app-open detection via a dedicated logcat stream (AdbDevice.streaming_shell). Default off \u2014 watchdog polling keeps the state honest either way; if the stream fails repeatedly the integration disables it for the session and falls back to polling. Flip on and report stability on your device."
"jellyfin_entity": "The Jellyfin media_player for this Shield. Stream details come from the HA Jellyfin integration \u2014 no extra setup needed.",
"receiver_entity": "The receiver or soundbar this Shield plays through.",
"receiver_input": "The receiver input the Shield uses, for example 'Shield'.",
"power_entity": "A smart plug switch, if you want reliable on/off detection.",
"adb_enabled": "Reads now-playing information over ADB from any app. Requires Developer options > Network debugging on the Shield.",
"adb_store_app_info": "Shows the title and artist from non-Jellyfin apps. Turn off to show only the app name (privacy mode).",
"adb_host": "The Shield's IP address. Leave blank to use the Android TV Remote entity's address.",
"adb_port": "The ADB port. The standard value is 5555.",
"app_name_map": "One entry per line, package=name (for example ar.tvplayer.tv=TiviMate), with an optional |icon after the name. Leave blank to clear.",
"adb_command_enabled": "Adds the nvidia_shield.adb_command service (a raw ADB shell command over the persistent connection). Security-relevant: leave it off unless you need it \u2014 the same opt-in posture as the core androidtv integration.",
"channel_digit_delay": "Pause between the channel digits entered via play_media(channel) \u2014 0.05 to 1.0 seconds (default 0.1, the tested value).",
"channel_send_enter": "Presses ENTER after the last digit (protocol key first, then the ADB keyevent fallback). Off by default, matching TiviMate's tested behavior.",
"logcat_events": "Detects app changes instantly through a dedicated logcat stream (AdbDevice.streaming_shell). Off by default \u2014 polling keeps the state correct either way, and the integration falls back to polling if the stream keeps failing."
}
}
}
@@ -3,65 +3,85 @@
"step": {
"user": {
"title": "NVIDIA Shield",
"description": "Set up your NVIDIA Shield. Link the Android TV remote, Jellyfin, receiver, and power monitoring.",
"description": "{discovery_note}Set up your NVIDIA Shield. Everything here is optional except one state source \u2014 an Android TV Remote entity, the Nvidia Remote Protocol, or ADB. Link the devices and services you want to use.",
"data": {
"name": "Device Name",
"androidtv_entity": "The media_player entity from the Android TV Remote integration. Optional for standalone setups that pair with the Nvidia Remote Protocol instead.",
"jellyfin_entity": "Jellyfin Media Player",
"name": "Name",
"androidtv_entity": "Android TV Remote entity",
"jellyfin_entity": "Jellyfin media player",
"receiver_entity": "Receiver",
"receiver_input": "Receiver Input",
"power_entity": "Power Switch",
"adb_enabled": "Enable ADB Media Session",
"adb_store_app_info": "Show App Media Info",
"adb_host": "ADB Host",
"adb_port": "ADB Port",
"protocol_enabled": "Pair with Nvidia Remote Protocol",
"protocol_host": "Protocol Host"
"receiver_input": "Receiver input",
"power_entity": "Power switch",
"adb_enabled": "Read state over ADB",
"adb_store_app_info": "Show title and artist",
"adb_host": "ADB host (IP address)",
"adb_port": "ADB port",
"protocol_enabled": "Add the Nvidia Remote Protocol",
"protocol_host": "Protocol host (IP address)"
},
"data_description": {
"androidtv_entity": "The media_player entity from the Android TV Remote integration.",
"jellyfin_entity": "The Jellyfin media_player for this device. Leave blank to auto-detect.",
"receiver_entity": "The receiver this device is connected to.",
"receiver_input": "Which input on the receiver (e.g. 'Shield').",
"power_entity": "Smart plug switch for reliable on/off detection.",
"adb_enabled": "Use ADB to read what's playing from any app (NewPipe, Spotify, etc.). Requires ADB debugging enabled on the Shield.",
"adb_store_app_info": "Show title and artist from non-Jellyfin apps. Disable to only show the app name (privacy mode).",
"adb_host": "Override the ADB host IP. Leave blank to auto-detect from the Android TV entity \u2014 pre-filled automatically when the Shield was discovered.",
"adb_port": "ADB port number. Default is 5555.",
"protocol_enabled": "Adds the all-in-one remote-protocol transport (power/availability/app push and volume readout). Adds a PIN pairing step; the Shield shows the code.",
"protocol_host": "Override the protocol host IP. Leave blank to auto-detect (same fallback as the ADB host) \u2014 pre-filled automatically on discovery."
"androidtv_entity": "The media_player entity from the Android TV Remote integration. Leave blank if you use the Nvidia Remote Protocol or ADB instead.",
"jellyfin_entity": "The Jellyfin media_player for this Shield. Leave blank to detect it automatically.",
"receiver_entity": "The receiver or soundbar this Shield plays through.",
"receiver_input": "The receiver input the Shield uses, for example 'Shield'. Leave blank if you are not sure.",
"power_entity": "A smart plug switch, if you want reliable on/off detection.",
"adb_enabled": "Reads now-playing information over ADB from any app. Requires Developer options > Network debugging on the Shield.",
"adb_store_app_info": "Shows the title and artist from non-Jellyfin apps. Turn off to show only the app name (privacy mode).",
"adb_host": "The Shield's IP address. Pre-filled when the Shield was discovered, or when an Android TV Remote entity is set \u2014 change it only if you need a different address.",
"adb_port": "The ADB port. The standard value is 5555 \u2014 change it only if ADB uses a custom port on your network.",
"protocol_enabled": "Adds the all-in-one remote-protocol transport (power, availability, app push and volume readout). It needs a one-time PIN pairing right after this step.",
"protocol_host": "The Shield's IP address for the remote protocol. Pre-filled when the Shield was discovered \u2014 leave blank to use the ADB host."
}
},
"adb_pair": {
"title": "Pair ADB",
"description": "On the Shield, go to Settings > Developer options and enable Network debugging. Then click Submit. You will see a prompt on the Shield to allow the connection -- approve it. If the connection fails, click Submit to try again."
"title": "Allow ADB access",
"description": "On the Shield, open Settings > System > Developer options and turn on Network debugging (it may be called ADB debugging). Then select Submit \u2014 the Shield shows an Allow prompt; approve it. Nothing happened? Check the address below and select Submit again. Not now? Turn on 'Continue without ADB pairing' and select Submit.",
"data": {
"adb_host": "ADB host (IP address)",
"adb_port": "ADB port",
"skip_pairing": "Continue without ADB pairing"
},
"data_description": {
"adb_host": "The Shield's IP address. Pre-filled when the Shield was discovered \u2014 change it only if the Shield moved or you gave it a different address.",
"adb_port": "The standard ADB port is 5555.",
"skip_pairing": "Finishes setup now and leaves ADB unpaired. The Pair ADB button stays available on the device page, so you can pair later."
}
},
"protocol_pair": {
"title": "Pair Remote Protocol",
"description": "The Shield is now showing a pairing code. Enter it here. Click Submit to confirm.",
"title": "Pair the remote protocol",
"description": "While this step is open the Shield shows a pairing code. Type it below and select Submit. If no code appears on the Shield, check the IP address below and select Submit again \u2014 pairing restarts and the Shield shows a new code. To set up without the remote protocol, turn on 'Continue without protocol pairing'.",
"data": {
"pairing_code": "Pairing code shown on the Shield"
"pairing_code": "Pairing code shown on the Shield",
"protocol_host": "Protocol host (IP address)",
"skip_pairing": "Continue without protocol pairing"
},
"data_description": {
"pairing_code": "The code the Shield displays while this step is open.",
"protocol_host": "The Shield's IP address. Pre-filled when the Shield was discovered \u2014 change it if the Shield moved.",
"skip_pairing": "Finishes setup with the remote protocol off \u2014 it stays off until you pair it. Pair later from this integration's entry menu: choose Reconfigure."
}
},
"reauth_confirm": {
"title": "Re-pair Nvidia Shield",
"description": "The Shield reset to factory settings, so the saved certificate is no longer trusted. A pairing code will be shown on the Shield."
"description": "The Shield no longer trusts the saved pairing certificate, usually after a factory reset. Select Submit to pair again \u2014 the Shield then shows a code."
}
},
"error": {
"adb_failed": "Could not connect via ADB. Make sure Network debugging is enabled on the Shield and you approved the connection prompt. Click Submit to try again.",
"adb_no_host": "Could not determine the Shield's IP address. Set the ADB Host field on the previous step.",
"protocol_no_host": "Could not determine the Shield's IP address. Set the Protocol Host or ADB Host field.",
"protocol_invalid_code": "Pairing has not started or the code is empty. Click Submit to retry.",
"protocol_pair_failed": "The Shield rejected that pairing code. Click Submit to try again.",
"no_state_source": "No state source selected. Provide at least one of: the Android TV Remote entity, the Nvidia Remote Protocol pairing, or ADB media-session mode (enable ADB and set the ADB Host).",
"invalid_app_map": "Each line must be package=name with a non-empty package (no spaces) and a non-empty name \u2014 e.g. ar.tvplayer.tv=TiviMate.",
"adb_failed": "The Shield did not accept the ADB connection. Make sure Network debugging is on, approve the Allow prompt on the Shield, then select Submit to try again.",
"adb_no_host": "No IP address for the Shield yet. Fill in the ADB host field above (a discovered Shield, or one linked to an Android TV Remote entity, is pre-filled automatically).",
"protocol_no_host": "No IP address for the Shield yet. Fill in the protocol host field above (a discovered Shield, or one linked to an Android TV Remote entity, is pre-filled automatically).",
"protocol_pair_failed": "The Shield rejected that pairing code. Check the code on the Shield and select Submit to try again.",
"protocol_unavailable": "The androidtvremote2 library is not installed, so the remote protocol cannot pair. Install it and restart Home Assistant, or turn on 'Continue without protocol pairing' to finish setup anyway.",
"no_state_source": "No state source selected. Choose at least one: an Android TV Remote entity, 'Add the Nvidia Remote Protocol', or 'Read state over ADB'.",
"invalid_app_map": "Each line must be package=name with a non-empty package (no spaces) and a non-empty name \u2014 for example ar.tvplayer.tv=TiviMate.",
"invalid_digit_delay": "The digit delay must be between 0.05 and 1.0 seconds."
},
"abort": {
"already_configured": "This Shield entity is already configured.",
"already_configured": "This Shield is already set up.",
"reauth_successful": "Re-paired successfully.",
"reauth_skipped": "Pairing skipped. The saved certificate is still not trusted, so the remote protocol stays unavailable until you pair again.",
"reauth_entry_missing": "This entry no longer exists.",
"reconfigure_skipped": "Nothing changed \u2014 the remote protocol stays off until you pair it. Choose Reconfigure from this integration's entry menu when you are ready.",
"reconfigure_entry_missing": "This entry no longer exists, so there is nothing to reconfigure.",
"zeroconf_no_host": "The discovery event carried no usable address.",
"yaml_seed_only": {},
"yaml_seed": {
@@ -73,37 +93,37 @@
"options": {
"step": {
"init": {
"title": "NVIDIA Shield Settings",
"description": "Configure linked devices and monitoring.",
"title": "NVIDIA Shield settings",
"description": "Settings for this Shield. Saving reloads the integration so the changes take effect.",
"data": {
"jellyfin_entity": "Jellyfin Media Player",
"jellyfin_entity": "Jellyfin media player",
"receiver_entity": "Receiver",
"receiver_input": "Receiver Input",
"power_entity": "Power Switch",
"adb_enabled": "Enable ADB Media Session",
"adb_store_app_info": "Show App Media Info",
"adb_host": "ADB Host",
"adb_port": "ADB Port",
"app_name_map": "Apps name/icon map",
"adb_command_enabled": "Enable raw ADB shell service",
"receiver_input": "Receiver input",
"power_entity": "Power switch",
"adb_enabled": "Read state over ADB",
"adb_store_app_info": "Show title and artist",
"adb_host": "ADB host (IP address)",
"adb_port": "ADB port",
"app_name_map": "App names and icons",
"adb_command_enabled": "Enable the adb_command service",
"channel_digit_delay": "Channel digit delay (seconds)",
"channel_send_enter": "Send ENTER after channel digits",
"logcat_events": "Live events push (logcat)"
"logcat_events": "Instant app detection (logcat stream)"
},
"data_description": {
"jellyfin_entity": "The Jellyfin media_player for this device. Session stream info is read from the HA Jellyfin integration's existing coordinator \u2014 no extra config needed.",
"receiver_entity": "The receiver this device is connected to.",
"receiver_input": "Which input on the receiver.",
"power_entity": "Smart plug switch for reliable on/off detection.",
"adb_enabled": "Use ADB to read now-playing from any app. Requires ADB debugging on the Shield.",
"adb_store_app_info": "Show title and artist from non-Jellyfin apps. Disable for privacy.",
"adb_host": "IP address for ADB connection. Leave blank to auto-detect from the Android TV Remote integration.",
"adb_port": "ADB port (default 5555).",
"app_name_map": "One entry per line, package=name (e.g. ar.tvplayer.tv=TiviMate). The friendly name resolves after the Android TV Remote entity and the protocol push; icons follow the media image lookup by name. Leave blank to clear.",
"adb_command_enabled": "Adds the nvidia_shield.adb_command service (raw ADB shell over the persistent connection). Security-relevant: leave off unless you need it \u2014 it is the same opt-in posture as the core androidtv integration.",
"channel_digit_delay": "Pause between the channel digits entered via play_media(channel) \u2014 0.05 to 1.0 seconds (default 0.1, matching the validated behavior).",
"channel_send_enter": "Optionally presses ENTER after the last digit (protocol key, then the ADB keyevent fallback). Off by default \u2014 matches TiviMate's validated behavior.",
"logcat_events": "Instant app-open detection via a dedicated logcat stream (AdbDevice.streaming_shell). Default off \u2014 watchdog polling keeps the state honest either way; if the stream fails repeatedly the integration disables it for the session and falls back to polling. Flip on and report stability on your device."
"jellyfin_entity": "The Jellyfin media_player for this Shield. Stream details come from the HA Jellyfin integration \u2014 no extra setup needed.",
"receiver_entity": "The receiver or soundbar this Shield plays through.",
"receiver_input": "The receiver input the Shield uses, for example 'Shield'.",
"power_entity": "A smart plug switch, if you want reliable on/off detection.",
"adb_enabled": "Reads now-playing information over ADB from any app. Requires Developer options > Network debugging on the Shield.",
"adb_store_app_info": "Shows the title and artist from non-Jellyfin apps. Turn off to show only the app name (privacy mode).",
"adb_host": "The Shield's IP address. Leave blank to use the Android TV Remote entity's address.",
"adb_port": "The ADB port. The standard value is 5555.",
"app_name_map": "One entry per line, package=name (for example ar.tvplayer.tv=TiviMate), with an optional |icon after the name. Leave blank to clear.",
"adb_command_enabled": "Adds the nvidia_shield.adb_command service (a raw ADB shell command over the persistent connection). Security-relevant: leave it off unless you need it \u2014 the same opt-in posture as the core androidtv integration.",
"channel_digit_delay": "Pause between the channel digits entered via play_media(channel) \u2014 0.05 to 1.0 seconds (default 0.1, the tested value).",
"channel_send_enter": "Presses ENTER after the last digit (protocol key first, then the ADB keyevent fallback). Off by default, matching TiviMate's tested behavior.",
"logcat_events": "Detects app changes instantly through a dedicated logcat stream (AdbDevice.streaming_shell). Off by default \u2014 polling keeps the state correct either way, and the integration falls back to polling if the stream keeps failing."
}
}
}
+134 -18
View File
@@ -10,6 +10,9 @@ import asyncio
import enum
import sys
import types
from collections.abc import Mapping
import voluptuous_mirror
ENTITIES_STUB_NAMES = (
"voluptuous",
@@ -27,6 +30,7 @@ ENTITIES_STUB_NAMES = (
"homeassistant.helpers.device_registry",
"homeassistant.helpers.selector",
"homeassistant.helpers.event",
"homeassistant.helpers.redact",
"homeassistant.helpers.entity",
"homeassistant.helpers.entity_platform",
"nsh.helpers",
@@ -62,35 +66,94 @@ class MediaPlayerEntityFeature(enum.IntFlag):
class MediaClass:
"""homeassistant.components.media_player.MediaClass (values we use)."""
APP = "app"
DIRECTORY = "directory"
class BrowseMediaSource:
"""HA BrowseMedia shape used by the app directory tests."""
class MediaType:
"""homeassistant.components.media_player.MediaType (values we use)."""
APP = "app"
APPS = "apps"
class BrowseMedia:
"""The REAL media_player.BrowseMedia shape (t37 D1).
Keyword-only with the six arguments real HA requires, the optional ones it
accepts, a faithful ``as_dict`` and — crucially — NO ``domain`` parameter:
``BrowseMediaSource`` lives in
``homeassistant.components.media_source.models`` (its contract is the
``media_source://<domain>`` id) and is NOT importable from
``homeassistant.components.media_player`` at all. The old import and the
old ``domain=`` kwarg therefore fail IN-SUITE instead of hiding behind a
permissive stub.
"""
def __init__(
self,
*,
title,
media_class,
media_content_type,
media_content_id,
media_content_type,
title,
can_play,
can_expand,
domain=None, # t30 G18 (2026-05-20 blog contract)
children=None,
children_media_class=None,
thumbnail=None,
not_shown=0,
can_search=False,
search_media_classes=None,
):
self.title = title
self.media_class = media_class
self.media_content_type = media_content_type
self.media_content_id = media_content_id
self.media_content_type = media_content_type
self.title = title
self.can_play = can_play
self.can_expand = can_expand
self.domain = domain
self.children = children or []
self.children = children
self.children_media_class = children_media_class
self.thumbnail = thumbnail
self.not_shown = not_shown
self.can_search = can_search
self.search_media_classes = search_media_classes
def as_dict(self, *, parent=True):
"""Real payload conversion — what the frontend actually consumes."""
if self.children_media_class is None and self.children:
self.calculate_children_class()
response = {
"title": self.title,
"media_class": self.media_class,
"media_content_type": self.media_content_type,
"media_content_id": self.media_content_id,
"children_media_class": self.children_media_class,
"can_play": self.can_play,
"can_expand": self.can_expand,
"can_search": self.can_search,
"search_media_classes": self.search_media_classes,
"thumbnail": self.thumbnail,
}
if not parent:
return response
response["not_shown"] = self.not_shown
response["children"] = (
[child.as_dict(parent=False) for child in self.children]
if self.children
else []
)
return response
def calculate_children_class(self):
"""Real helper: the children's common class, else DIRECTORY."""
self.children_media_class = MediaClass.DIRECTORY
assert self.children is not None
proposed_class = self.children[0].media_class
if all(child.media_class == proposed_class for child in self.children):
self.children_media_class = proposed_class
class MediaEntityManager:
@@ -115,9 +178,9 @@ def install() -> None:
return
vol = types.ModuleType("voluptuous")
vol.Required = lambda key, default=None: key
vol.Optional = lambda key, default=None: key
vol.Schema = lambda schema: schema
# t33 F-A: the faithful mirror keeps schema defaults visible — a bare-key
# stub cannot tell a schema WITH a default from one without.
voluptuous_mirror.install(vol)
sys.modules["voluptuous"] = vol
ha = types.ModuleType("homeassistant")
@@ -144,16 +207,24 @@ def install() -> None:
async_set_unique_id = _stub_set_unique_id
_abort_if_unique_id_configured = _noop_abort
def _async_handle_discovery_without_unique_id(self):
# t32 F2 (audited helper): recording no-op at hub scope
pass
async def _async_handle_discovery_without_unique_id(self):
# t37 D3: the real helper is a COROUTINE and core awaits it — a
# sync stub here would turn the production `await` into a
# TypeError. The abort semantics live in
# tests/test_standalone_setup.py, whose stub owns entry/in-progress
# state; at hub scope there is nothing to abort on.
return None
def async_show_form(self, *, step_id, data_schema=None, errors=None):
def async_show_form(
self, *, step_id, data_schema=None, errors=None,
description_placeholders=None,
):
return {
"type": "form",
"step_id": step_id,
"errors": dict(errors or {}),
"data_schema": data_schema,
"description_placeholders": dict(description_placeholders or {}),
}
def async_create_entry(self, *, title, data, options=None):
@@ -167,6 +238,11 @@ def install() -> None:
return entry
return None # the caller's cached entry is used
def _get_reconfigure_entry(self, entry=None):
if entry is not None:
return entry
return None # the caller's cached entry is used
# NOTE: the real HA signature is async_update_reload_and_abort(entry, ...)
# which updates, schedules the reload and aborts; the stub records both.
@@ -267,9 +343,12 @@ def install() -> None:
mp.MediaPlayerEntity = MediaPlayerEntity
mp.MediaPlayerDeviceClass = types.SimpleNamespace(TV="tv")
mp.MediaPlayerEntityFeature = MediaPlayerEntityFeature
mp.BrowseMedia = BrowseMediaSource # factory used as a constructor
mp.BrowseMediaSource = BrowseMediaSource
mp.BrowseMedia = BrowseMedia
mp.MediaClass = MediaClass
mp.MediaType = MediaType
# NOTE (t37 D1): homeassistant.components.media_player exposes NO
# BrowseMediaSource — deliberately not stubbed, so the drifted import
# fails here exactly as it does in a real install.
mp.MediaPlayerState = MediaPlayerState
sys.modules["homeassistant.components.media_player"] = mp
@@ -336,6 +415,43 @@ def install() -> None:
entity_platform = types.ModuleType("homeassistant.helpers.entity_platform")
entity_platform.AddEntitiesCallback = object
helpers.entity_platform = entity_platform
# t37 D2: homeassistant.helpers.redact.async_redact_data — the REAL helper
# (diagnostics.py imports it at module scope; `homeassistant.util.diagnostic`
# never existed). Modelled faithfully: the same masking of keys found in
# ``to_redact``, the same nested mapping/list recursion, the same skip of
# None/empty-string values.
redact = types.ModuleType("homeassistant.helpers.redact")
redact.REDACTED = "**REDACTED**"
def _stub_async_redact_data(data, to_redact):
if not isinstance(data, (Mapping, list)):
return data
if isinstance(data, list):
return [_stub_async_redact_data(val, to_redact) for val in data]
redacted = {**data}
for key, value in redacted.items():
if value is None:
continue
if isinstance(value, str) and not value:
continue
if key in to_redact:
if isinstance(to_redact, Mapping):
redacted[key] = to_redact[key](value)
else:
redacted[key] = redact.REDACTED
elif isinstance(value, Mapping):
redacted[key] = _stub_async_redact_data(value, to_redact)
elif isinstance(value, list):
redacted[key] = [
_stub_async_redact_data(item, to_redact) for item in value
]
return redacted
redact.async_redact_data = _stub_async_redact_data
helpers.redact = redact
sys.modules["homeassistant.helpers.redact"] = redact
sys.modules["homeassistant.helpers.event"] = event
sys.modules["homeassistant.helpers.entity"] = entity
sys.modules["homeassistant.helpers.entity_platform"] = entity_platform
+514
View File
@@ -0,0 +1,514 @@
"""Home-Assistant API-drift sweep (t37).
Why this exists
---------------
The unit suite runs our production modules against hand-written stubs
(``tests/entity_stublib.py`` plus the per-file stub fixtures), so a call site
that no longer matches the REAL Home Assistant API stays green. t36's sweep
found exactly that three times — two of them silent in production:
* **D1** ``from homeassistant.components.media_player import BrowseMediaSource``
— media_player has not exported that name for years (it lives in
``homeassistant.components.media_source.models``), so every real browse
raised ImportError;
* **D2** ``from homeassistant.util.diagnostic import async_redact_data`` — that
module does not exist at all, and the swallowed ImportError turned the
documented redaction pass into a no-op;
* **D3** ``self._async_handle_discovery_without_unique_id()`` — a coroutine
``ConfigFlow`` awaits in core; called without ``await`` its body never ran,
so HA's discovery dedup was dead and Python warned about the coroutine.
What it does
------------
``scan()`` parses every production module and resolves each Home-Assistant
dependency against an INSTALLED homeassistant:
1. every ``homeassistant.*`` import — module present, attribute present (or a
submodule of that name exists, which is what ``from x import submodule``
really means);
2. every keyword argument passed to a known HA API (see ``HA_CALL_OWNERS``)
must exist in the real signature — the ``options_updates`` class of bug;
3. every ``self.<method>()`` call inside a class that inherits from an HA class
is compared against the real definition: a coroutine must be awaited, and an
awaited call must actually be a coroutine.
Known limits (deliberate, and why the sweep still catches the real classes):
``**kwargs`` splats cannot be checked statically, calls on objects other than
``self`` (``self.hass.config_entries.…``) are only keyword-checked, and a
method our own class overrides is not resolved against HA. Those gaps are
visible in the ``Report`` counters, which the pytest wrapper asserts on so the
sweep cannot silently become vacuous.
The standalone entry point (``python tests/ha_api_drift.py [path …]``) exits
non-zero on findings *and* on counters below ``FLOORS`` — an empty or
mis-targeted directory is a failure, never a green sweep. ``FLOORS`` is the one
source of truth for both that entry point and the pytest wrapper, and the
interpreter floor Home Assistant itself declares lives here too
(``HOMEASSISTANT_PYTHON_FLOOR``), where CI's guard reads it instead of repeating
the number.
Behaviour note for users (t37 D3): MAC-less zeroconf discovery now follows Home
Assistant's stricter contract — the awaited
``_async_handle_discovery_without_unique_id`` aborts ``already_configured`` as
soon as ANY entry exists for this domain, so a second Shield that advertises no
MAC is added manually. See README "Discovery (zeroconf)".
Without homeassistant installed the sweep cannot resolve anything: ``scan()``
returns a single ``no-homeassistant`` finding, the counters stay at zero (so the
floors fail on purpose) and nothing is ever reported green. The pytest wrapper
(``tests/test_ha_api_drift.py``) skips cleanly in that case, and CI runs one job
with homeassistant installed for exactly this check.
"""
from __future__ import annotations
import ast
import importlib
import importlib.util
import inspect
import re
from dataclasses import dataclass, field
from pathlib import Path
SHIELD_DIR = (
Path(__file__).resolve().parent.parent / "custom_components" / "nvidia_shield"
)
#: Attribute name -> (module, class) that owns the REAL signature. Resolved with
#: getattr at scan time, so a renamed or dropped API is a finding, not a crash.
HA_CALL_OWNERS: dict[str, tuple[str, str]] = {
"async_update_reload_and_abort": ("homeassistant.config_entries", "ConfigFlow"),
"async_show_form": ("homeassistant.config_entries", "ConfigFlow"),
"async_create_entry": ("homeassistant.config_entries", "ConfigFlow"),
"async_abort": ("homeassistant.config_entries", "ConfigFlow"),
"async_set_unique_id": ("homeassistant.config_entries", "ConfigFlow"),
"_abort_if_unique_id_configured": (
"homeassistant.config_entries",
"ConfigFlow",
),
"async_update_entry": ("homeassistant.config_entries", "ConfigEntries"),
"async_schedule_reload": ("homeassistant.config_entries", "ConfigEntries"),
"async_get_entry": ("homeassistant.config_entries", "ConfigEntries"),
"async_forward_entry_setups": ("homeassistant.config_entries", "ConfigEntries"),
"async_unload_platforms": ("homeassistant.config_entries", "ConfigEntries"),
"async_add_executor_job": ("homeassistant.core", "HomeAssistant"),
}
#: Non-vacuity floors — the ONE source of truth for the standalone entry point
#: and the pytest wrapper (t39 F1). They sit below the current walk (13 modules,
#: 60 imports, 35 checked arguments, 7 awaited HA calls) so a broken or
#: mis-targeted walk fails loudly instead of printing zeros and exiting 0.
FLOORS: dict[str, int] = {
"modules": 10,
"imports": 40,
"call_args": 20,
"awaited_calls": 5,
}
#: The interpreter floor Home Assistant itself declares: PyPI metadata for
#: 2026.10.1 is ``Requires-Python >=3.14.2``, so a 3.13 runner cannot even
#: install the version this sweep must run against. CI's ``ha-api-drift`` job
#: provisions a matching interpreter and guards with THIS constant, so the
#: number is written once; tests/test_ha_api_drift.py pins it against the
#: installed distribution's own metadata.
HOMEASSISTANT_PYTHON_FLOOR: tuple[int, int, int] = (3, 14, 2)
_MISSING = object()
@dataclass(frozen=True)
class Finding:
"""One drifted API use."""
path: str
lineno: int
kind: str
detail: str
def __str__(self) -> str:
return f"{self.path}:{self.lineno}: {self.kind}: {self.detail}"
@dataclass
class Report:
"""Findings plus the counters that prove the sweep actually walked things."""
findings: list[Finding] = field(default_factory=list)
modules: int = 0
imports: int = 0
call_args: int = 0
awaited_calls: int = 0
def floor_violations(self, floors: dict[str, int] | None = None) -> list[str]:
"""Counters below the non-vacuity floors — a sweep nobody should trust.
The floors default to module-level ``FLOORS`` so the standalone entry
point and the pytest wrapper cannot drift apart (t39 F1).
"""
floors = FLOORS if floors is None else floors
return [
f"only {getattr(self, name)} {name} (floor: {floor})"
for name, floor in floors.items()
if getattr(self, name) < floor
]
def __str__(self) -> str:
if not self.findings:
return (
"no Home-Assistant API drift "
f"({self.modules} modules, {self.imports} imports, "
f"{self.call_args} call arguments, "
f"{self.awaited_calls} awaited HA calls)"
)
return "\n".join(str(finding) for finding in self.findings)
# --------------------------------------------------------------------- imports
def _check_import(module_name: str, attr: str | None) -> str | None:
"""Return a reason string when the import target is missing, else None."""
try:
module = importlib.import_module(module_name)
except Exception as err: # noqa: BLE001 - the error IS the finding
return (
f"module {module_name!r} is not importable "
f"({type(err).__name__}: {err})"
)
if attr is None or hasattr(module, attr):
return None
# `from package import submodule` resolves through the import system even
# when the attribute is not visible on the package yet.
try:
importlib.import_module(f"{module_name}.{attr}")
except Exception: # noqa: BLE001 - the error IS the finding
return f"{module_name}.{attr} does not exist"
return None
def _scan_imports(tree: ast.AST, path: str, report: Report) -> None:
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom):
module_name = node.module or ""
if not module_name.startswith("homeassistant"):
continue
for alias in node.names:
report.imports += 1
reason = _check_import(module_name, alias.name)
if reason:
report.findings.append(
Finding(
path,
node.lineno,
"broken-import",
f"from {module_name} import {alias.name}: {reason}",
)
)
elif isinstance(node, ast.Import):
for alias in node.names:
if not alias.name.startswith("homeassistant"):
continue
report.imports += 1
reason = _check_import(alias.name, None)
if reason:
report.findings.append(
Finding(
path,
node.lineno,
"broken-import",
f"import {alias.name}: {reason}",
)
)
# -------------------------------------------------------------------- keywords
def _real_signature(module_name: str, owner: str, attr: str):
"""(signature, reason) for an HA API; both None when uninspectable."""
try:
module = importlib.import_module(module_name)
except Exception as err: # noqa: BLE001 - the error IS the finding
return None, f"{module_name} is not importable ({type(err).__name__})"
owner_obj = getattr(module, owner, None)
if owner_obj is None:
return None, f"{module_name}.{owner} does not exist"
api = getattr(owner_obj, attr, None)
if api is None:
return None, f"{module_name}.{owner}.{attr} does not exist"
try:
return inspect.signature(api), None
except (TypeError, ValueError): # pragma: no cover - C callables
return None, None
def _scan_call_keywords(tree: ast.AST, path: str, report: Report) -> None:
for node in ast.walk(tree):
if not isinstance(node, ast.Call) or not node.keywords:
continue
func = node.func
if not isinstance(func, ast.Attribute):
continue
owner = HA_CALL_OWNERS.get(func.attr)
if owner is None:
continue
signature, reason = _real_signature(*owner, func.attr)
if reason:
report.findings.append(
Finding(path, node.lineno, "unknown-ha-api", reason)
)
continue
if signature is None: # pragma: no cover - uninspectable
continue
if any(
param.kind is inspect.Parameter.VAR_KEYWORD
for param in signature.parameters.values()
):
continue
for keyword in node.keywords:
if keyword.arg is None: # **splat: not statically checkable
continue
report.call_args += 1
if keyword.arg not in signature.parameters:
report.findings.append(
Finding(
path,
keyword.lineno,
"unknown-kwarg",
f"{func.attr}() has no {keyword.arg!r} parameter; "
f"real signature: {signature}",
)
)
# ---------------------------------------------------------------- await/sync
def _module_imports(tree: ast.AST) -> dict[str, tuple[str, str | None]]:
"""Local name -> (module, attribute-or-None) for the module's HA imports."""
imports: dict[str, tuple[str, str | None]] = {}
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and (node.module or "").startswith(
"homeassistant"
):
for alias in node.names:
imports[alias.asname or alias.name] = (node.module, alias.name)
elif isinstance(node, ast.Import):
for alias in node.names:
if alias.name.startswith("homeassistant"):
imports[alias.asname or alias.name.split(".")[0]] = (
alias.name,
None,
)
return imports
def _resolve_base(expr: ast.AST, imports: dict[str, tuple[str, str | None]]):
"""Resolve a class base expression to (module, attribute) when it is HA's."""
if isinstance(expr, ast.Attribute) and isinstance(expr.value, ast.Name):
target = imports.get(expr.value.id)
if target is None:
return None
module_name, attr = target
if attr is None:
# `import homeassistant.config_entries as config_entries`
return module_name, expr.attr
if module_name == "homeassistant":
# `from homeassistant import config_entries` binds the SUBMODULE,
# so `config_entries.ConfigFlow` is that module's attribute.
return f"{module_name}.{attr}", expr.attr
return None
if isinstance(expr, ast.Name):
target = imports.get(expr.id)
if target is not None and target[1] is not None:
return target
return None
def _ha_bases(
class_node: ast.ClassDef, imports: dict[str, tuple[str, str | None]]
) -> list[type]:
bases: list[type] = []
for expr in class_node.bases:
target = _resolve_base(expr, imports)
if target is None:
continue
try:
candidate = getattr(importlib.import_module(target[0]), target[1])
except Exception: # noqa: BLE001 - unresolvable base, skip it
continue
if isinstance(candidate, type):
bases.append(candidate)
return bases
def _method_on_bases(bases: list[type], attr: str):
for base in bases:
found = getattr(base, attr, _MISSING)
if found is not _MISSING:
return found, base
return None, None
class _AwaitVisitor(ast.NodeVisitor):
"""Flags awaited/not-awaited mismatches for `self.<ha method>()` calls."""
def __init__(self, bases: list[type], path: str, report: Report) -> None:
self._bases = bases
self._path = path
self._report = report
self._in_await = 0
def visit_Await(self, node: ast.Await) -> None:
self._in_await += 1
self.generic_visit(node)
self._in_await -= 1
def visit_Call(self, node: ast.Call) -> None:
func = node.func
if (
isinstance(func, ast.Attribute)
and isinstance(func.value, ast.Name)
and func.value.id == "self"
):
api, owner = _method_on_bases(self._bases, func.attr)
if api is not None:
is_coro = inspect.iscoroutinefunction(api)
awaited = self._in_await > 0
where = f"{owner.__module__}.{owner.__qualname__}"
if is_coro and not awaited:
self._report.findings.append(
Finding(
self._path,
node.lineno,
"missing-await",
f"self.{func.attr}() is a coroutine on {where} — "
f"core awaits it; without `await` its body never runs",
)
)
elif awaited and not is_coro:
self._report.findings.append(
Finding(
self._path,
node.lineno,
"awaits-sync-api",
f"self.{func.attr}() is not a coroutine on {where}, "
f"so awaiting it raises TypeError",
)
)
if awaited:
self._report.awaited_calls += 1
self.generic_visit(node)
def _scan_awaits(tree: ast.AST, path: str, report: Report) -> None:
imports = _module_imports(tree)
for node in ast.walk(tree):
if not isinstance(node, ast.ClassDef):
continue
bases = _ha_bases(node, imports)
if not bases:
continue
for statement in node.body:
# visit() walks the statement itself: a flat ast.walk would lose the
# `await` nesting and report every awaited call as un-awaited.
_AwaitVisitor(bases, path, report).visit(statement)
# ------------------------------------------------------ floors and versioning
def parse_requires_python(spec: str | None) -> tuple[int, ...] | None:
"""Parse a ``>=X.Y[.Z]`` Requires-Python string into a version tuple.
The CI interpreter guard and the metadata pin in the pytest wrapper both
need the same reading of Home Assistant's own declaration.
"""
if not spec:
return None
match = re.match(r"\s*>=\s*(\d+(?:\.\d+)*)", spec)
if match is None:
return None
return tuple(int(part) for part in match.group(1).split("."))
# ----------------------------------------------------------------------- scan
def homeassistant_available() -> bool:
"""True when a real homeassistant is importable in this interpreter."""
return importlib.util.find_spec("homeassistant") is not None
def scan(paths=None) -> Report:
"""Resolve every Home-Assistant API use in the production tree."""
if paths is None:
paths = [SHIELD_DIR]
elif isinstance(paths, (str, Path)):
paths = [paths]
report = Report()
if not homeassistant_available():
# Neither a false green nor a wall of phantom findings: with no package
# to resolve against, every import would "fail", so the sweep reports
# that it cannot judge and leaves the counters at zero — which trips the
# floors below on purpose.
report.findings.append(
Finding(
"<environment>",
0,
"no-homeassistant",
"homeassistant is not installed in this interpreter, so nothing "
"can be resolved; install it (uvx --with homeassistant --python "
"3.14 ...) or run tests/test_ha_api_drift.py, which skips cleanly",
)
)
return report
for root in paths:
root = Path(root)
if not root.exists():
report.findings.append(
Finding(
str(root),
0,
"bad-path",
"path does not exist — nothing was scanned",
)
)
continue
files = sorted(root.glob("*.py")) if root.is_dir() else [root]
for file in files:
report.modules += 1
tree = ast.parse(file.read_text(encoding="utf-8"), filename=str(file))
name = file.name
_scan_imports(tree, name, report)
_scan_call_keywords(tree, name, report)
_scan_awaits(tree, name, report)
return report
if __name__ == "__main__": # pragma: no cover - manual red-proof entry point
import sys
targets = sys.argv[1:] or None
result = scan(targets)
print(result)
violations = result.floor_violations()
if result.findings or violations:
for line in violations:
print(f"FLOOR: {line}", file=sys.stderr)
print(
f"drift sweep FAILED ({len(result.findings)} finding(s), "
f"{len(violations)} floor violation(s))",
file=sys.stderr,
)
raise SystemExit(1)
raise SystemExit(0)
+118 -5
View File
@@ -6,11 +6,13 @@ legacy external entity, plus the §7 channel digit sequence (protocol digits,
layer marks which tier served each command.
"""
import ast
import asyncio
import sys
import entity_stublib
import pytest
from conftest import _SHIELD_DIR
from entity_stublib import ENTITIES_STUB_NAMES, install, load_entity_modules
from nsh.connection import ShieldConnectionManager
from nsh.const import TIVIMATE_PACKAGE
@@ -992,10 +994,16 @@ class TestEntryDiagnostics:
from types import SimpleNamespace as DiagEntry
entry = DiagEntry(
data={},
data={
"name": "Shield",
"androidtv_entity": "media_player.shield",
"nested": {"adb_host": "10.0.0.5"},
},
options={
"adb_command_enabled": True,
"protocol_pairing_code": "4821",
"adb_host": "192.168.1.50",
"protocol_host": "192.168.1.50",
},
title="Shield",
)
@@ -1008,12 +1016,103 @@ class TestEntryDiagnostics:
payload = await async_get_config_entry_diagnostics(None, entry)
assert payload["entry"]["options"]["protocol_pairing_code"] == "**REDACTED**"
# t37 D2: the G14 field list (hosts included) really applies now — the
# old import pointed at homeassistant.util.diagnostic, which does not
# exist, so this pass silently did nothing and only the marker scan
# (pairing-/key-shaped names) masked anything.
assert payload["entry"]["options"]["adb_host"] == "**REDACTED**"
assert payload["entry"]["options"]["protocol_host"] == "**REDACTED**"
assert payload["entry"]["data"]["nested"]["adb_host"] == "**REDACTED**"
assert payload["entry"]["data"]["androidtv_entity"] == "media_player.shield"
assert payload["entry"]["options"]["adb_command_enabled"] is True
assert "awake" in payload["store"]
assert payload["transport"]["adb"]["connected"] is not None
assert payload["transport"]["protocol"]["active"] is not None
class TestFaithfulHaStubs:
"""t37: the harness mirrors WHERE real HA exposes its API.
A regression of the exact drifts this task repairs must fail IN-SUITE, not
hide behind a stub that exposes anything anywhere.
"""
def test_browse_media_source_is_not_importable_from_media_player(self):
import homeassistant.components.media_player as mp
assert isinstance(mp.BrowseMedia, type) # the real node class
assert not hasattr(mp, "BrowseMediaSource")
with pytest.raises(ImportError):
# the drifted import must fail here exactly as in a real install
from homeassistant.components.media_player import BrowseMediaSource # noqa: F401
def test_no_production_module_imports_browse_media_source(self):
"""t37 D1: the drift is an IMPORT — checked on the AST, not on prose."""
offenders = []
for path in sorted(_SHIELD_DIR.glob("*.py")):
tree = ast.parse(path.read_text(encoding="utf-8"))
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and any(
alias.name == "BrowseMediaSource" for alias in node.names
):
offenders.append(f"{path.name}:{node.lineno} imports it")
if isinstance(node, ast.Import) and any(
alias.name.endswith("BrowseMediaSource") for alias in node.names
):
offenders.append(f"{path.name}:{node.lineno} imports it")
assert offenders == []
def test_the_browse_nodes_refuse_the_media_source_domain_kwarg(self):
import homeassistant.components.media_player as mp
with pytest.raises(TypeError):
mp.BrowseMedia(
title="x",
media_class=mp.MediaClass.APP,
media_content_type=mp.MediaType.APP,
media_content_id="pkg",
can_play=True,
can_expand=False,
domain="nvidia_shield", # G18 keyword of BrowseMediaSource only
)
def test_redaction_uses_the_real_helper_module(self):
import homeassistant.helpers.redact as redact
import nsh.diagnostics as diagnostics
assert redact.REDACTED == "**REDACTED**"
# the module actually USES the helper from the real module path
assert diagnostics.async_redact_data is redact.async_redact_data
assert {"adb_host", "protocol_host"} <= diagnostics._TO_REDACT
assert diagnostics._REDACTED == redact.REDACTED
def test_no_production_module_imports_the_nonexistent_diagnostic_module(self):
"""t37 D2: `homeassistant.util.diagnostic` never existed."""
offenders = [
f"{path.name}:{node.lineno}"
for path in sorted(_SHIELD_DIR.glob("*.py"))
for node in ast.walk(ast.parse(path.read_text(encoding="utf-8")))
if isinstance(node, ast.ImportFrom)
and node.module == "homeassistant.util.diagnostic"
]
assert offenders == []
def test_diagnostics_has_no_silent_import_fallback(self):
"""The redaction import is at module scope — loud, not swallowed."""
tree = ast.parse(
(_SHIELD_DIR / "diagnostics.py").read_text(encoding="utf-8")
)
caught = [
handler
for node in ast.walk(tree)
if isinstance(node, ast.Try)
for handler in node.handlers
if isinstance(handler.type, ast.Name)
and handler.type.id == "ImportError"
]
assert caught == []
class TestBrowseMediaDirectory:
"""P4-2c: the app directory root and its launch path."""
@@ -1034,10 +1133,24 @@ class TestBrowseMediaDirectory:
("com.netflix.ninja", "Netflix"),
]
assert all(c.media_content_type == "app" for c in root.children)
# t30 G18 (2026-05-20 blog): the domain parameter is present at
# every level of the tree
assert root.domain == "nvidia_shield"
assert all(c.domain == "nvidia_shield" for c in root.children)
# t37 D1: a standard media_player.BrowseMedia tree — explicit
# media_content_id/-type at every level, the directory naming the class
# of its children, and NO media_source `domain` (that parameter belongs
# to media_source.models.BrowseMediaSource, which also builds the
# media_source:// ids).
assert root.media_class == "directory"
assert root.media_content_id == "apps"
assert root.media_content_type == "apps"
assert root.children_media_class == "app"
assert not hasattr(root, "domain")
assert [c.media_class for c in root.children] == ["app", "app"]
# ...and it serialises into the payload the frontend consumes
payload = root.as_dict()
assert payload["media_content_id"] == "apps"
assert payload["children_media_class"] == "app"
assert payload["children"][0]["media_content_id"] == "ar.tvplayer.tv"
assert payload["children"][0]["media_content_type"] == "app"
assert payload["children"][0]["can_play"] is True
async def test_root_falls_back_to_the_curated_default(self, modules):
player, _s, _i = make_player(modules) # no apps map configured
+14 -2
View File
@@ -158,12 +158,24 @@ def _install_stubs() -> None:
ConfigFlow._get_reauth_entry = lambda self, entry=None: entry
def _update_reload_and_abort(self, entry, **kwargs):
updates = kwargs.get("options_updates") or {}
# `options` REPLACES the entry options (the real signature); the merge
# form is still accepted for older callers.
updates = dict(kwargs.get("options_updates") or {})
updates.update(kwargs.get("options") or {})
options = entry.options
for key, value in updates.items():
options[key] = value
if kwargs.get("unique_id"):
entry.unique_id = kwargs["unique_id"]
self.hass.config_entries.async_schedule_reload(entry.entry_id)
return self.async_abort(reason="reauth_successful")
reason = kwargs.get("reason")
if reason is None:
reason = (
"reconfigure_successful"
if getattr(self, "source", None) == "reconfigure"
else "reauth_successful"
)
return self.async_abort(reason=reason)
ConfigFlow.async_update_reload_and_abort = _update_reload_and_abort
File diff suppressed because it is too large. Load diff
+208
View File
@@ -0,0 +1,208 @@
"""Run the Home-Assistant API-drift sweep (t37).
Skips cleanly when homeassistant is not installed — every other test file runs
against hand-written stubs, and this is the one check that must see the real
package. CI's `ha-api-drift` job installs it:
uvx --with homeassistant --with pytest-asyncio pytest tests/test_ha_api_drift.py -q
The sweep itself lives in ``tests/ha_api_drift.py`` (documented there).
"""
from __future__ import annotations
import ast
import subprocess
import sys
from pathlib import Path
import ha_api_drift
import pytest
pytest.importorskip(
"homeassistant",
reason="the API-drift sweep resolves our call sites against a real install",
)
#: The three t36 drifts plus the `options_updates` kwarg t35 removed — the
#: sweep must flag every one of them, or it cannot protect us from the next.
PRE_FIX_SOURCES = {
"d1_browse_media_source.py": (
"from homeassistant.components.media_player import BrowseMediaSource\n"
),
"d2_util_diagnostic.py": (
"from homeassistant.util.diagnostic import async_redact_data\n"
),
"d3_unawaited_discovery.py": (
"from homeassistant import config_entries\n"
"\n"
"\n"
"class PreFixFlow(config_entries.ConfigFlow, domain='nvidia_shield'):\n"
" async def async_step_zeroconf(self, discovery_info):\n"
" self._async_handle_discovery_without_unique_id()\n"
),
"t35_options_updates.py": (
"from homeassistant import config_entries\n"
"\n"
"\n"
"class PreFixReauthFlow(config_entries.ConfigFlow, domain='nvidia_shield'):\n"
" async def async_step_reauth_confirm(self, user_input=None):\n"
" self.async_update_reload_and_abort(\n"
" self.entry, options_updates={'protocol_enabled': True}\n"
" )\n"
" return self.async_abort(reason='reauth_successful')\n"
),
}
def _require_installed_homeassistant():
"""Skip when sys.modules holds a test stub rather than an install."""
import sys
module = sys.modules["homeassistant"]
path = getattr(module, "__file__", None)
if path is None or not Path(path).exists():
pytest.skip("homeassistant in sys.modules is a test stub, not an install")
def test_no_api_drift_in_the_production_tree():
"""The gate: every HA import, kwarg and await matches the real package."""
_require_installed_homeassistant()
report = ha_api_drift.scan()
assert report.findings == [], "\n" + str(report)
def test_the_sweep_is_not_vacuous():
"""The counters prove the walk really covered the production tree."""
_require_installed_homeassistant()
report = ha_api_drift.scan()
# Floors, not targets: the numbers live in ha_api_drift.FLOORS — one source
# of truth for this wrapper and the standalone entry point (t39 F1) — so a
# broken or mis-targeted walk fails here instead of looking green.
violations = report.floor_violations()
assert violations == [], "\n".join(violations)
def test_the_sweep_flags_the_pre_fix_defects(tmp_path):
"""Red proof: the t36 drifts and the options_updates kwarg are all caught."""
_require_installed_homeassistant()
for name, source in PRE_FIX_SOURCES.items():
(tmp_path / name).write_text(source, encoding="utf-8")
report = ha_api_drift.scan(tmp_path)
found = {(finding.path, finding.kind) for finding in report.findings}
assert found == {
("d1_browse_media_source.py", "broken-import"),
("d2_util_diagnostic.py", "broken-import"),
("d3_unawaited_discovery.py", "missing-await"),
("t35_options_updates.py", "unknown-kwarg"),
}, "\n" + str(report)
def test_the_sweep_accepts_the_fixed_forms(tmp_path):
"""The same call sites written the way production writes them are clean."""
_require_installed_homeassistant()
(tmp_path / "fixed.py").write_text(
"from homeassistant.components.media_player import BrowseMedia\n"
"from homeassistant.helpers.redact import async_redact_data\n"
"from homeassistant import config_entries\n"
"\n"
"\n"
"class FixedFlow(config_entries.ConfigFlow, domain='nvidia_shield'):\n"
" async def async_step_zeroconf(self, discovery_info):\n"
" await self._async_handle_discovery_without_unique_id()\n"
"\n"
" def finish(self):\n"
" return self.async_update_reload_and_abort(\n"
" self.entry, options={'protocol_enabled': True}\n"
" )\n",
encoding="utf-8",
)
report = ha_api_drift.scan(tmp_path)
assert report.findings == [], "\n" + str(report)
def _run_drift_module(*paths):
"""Run the standalone entry point in this (HA-bearing) interpreter."""
return subprocess.run(
[
sys.executable,
str(Path(ha_api_drift.__file__).resolve()),
*[str(path) for path in paths],
],
capture_output=True,
text=True,
check=False,
)
def test_the_standalone_entry_point_passes_on_the_production_tree():
"""F1: the CLI is the sweep, with the same verdict on the real tree."""
result = _run_drift_module()
assert result.returncode == 0, result.stderr
assert "no Home-Assistant API drift" in result.stdout
def test_the_standalone_entry_point_fails_on_an_empty_directory(tmp_path):
"""t39 F1: zeros must not exit 0 — the floors apply to the CLI too."""
result = _run_drift_module(tmp_path)
assert result.returncode != 0
assert "0 modules" in result.stdout # the counter that used to look green
assert "FLOOR:" in result.stderr
assert "drift sweep FAILED" in result.stderr
def test_the_standalone_entry_point_fails_on_a_missing_path(tmp_path):
"""A mis-specified path is a failure, not a silent zero-module sweep."""
result = _run_drift_module(tmp_path / "not-a-directory")
assert result.returncode != 0
assert "bad-path" in result.stdout
def test_no_floor_constants_are_duplicated_in_this_wrapper():
"""F1: the floors come from ha_api_drift.FLOORS — nowhere else."""
source = Path(__file__).read_text(encoding="utf-8")
tree = ast.parse(source)
local_ints = [
target.id
for node in tree.body
if isinstance(node, ast.Assign)
for target in node.targets
if isinstance(target, ast.Name)
and isinstance(node.value, ast.Constant)
and isinstance(node.value.value, int)
]
assert local_ints == [], f"floor-like constants belong in ha_api_drift: {local_ints}"
assert "floor_violations" in source
def test_the_declared_python_floor_never_lags_homeassistant():
"""F2: the CI guard's floor is HA's own Requires-Python, not a guess."""
_require_installed_homeassistant()
import importlib.metadata as metadata
declared = metadata.metadata("homeassistant")["Requires-Python"]
parsed = ha_api_drift.parse_requires_python(declared)
assert parsed is not None, f"unparsable Requires-Python: {declared!r}"
assert tuple(parsed) <= ha_api_drift.HOMEASSISTANT_PYTHON_FLOOR, (
f"tests/ha_api_drift.py declares {ha_api_drift.HOMEASSISTANT_PYTHON_FLOOR} "
f"but the installed homeassistant requires {declared} — raise the constant "
f"and the setup-python version in .gitea/workflows/test.yml"
)
+107 -24
View File
@@ -8,13 +8,16 @@ migration validity for v2 entries without the field.
"""
import asyncio # noqa: F401 (async steps need the event loop)
import gc
import sys
import types
import warnings
from pathlib import Path # noqa: E402 (F1 test key staging)
from types import SimpleNamespace
from unittest import mock
import pytest
import voluptuous_mirror as vol_mirror
from conftest import _SHIELD_DIR, _load_module
from nsh.const import (
CONF_ADB_ENABLED,
@@ -41,6 +44,19 @@ _STUB_NAMES = (
)
class DiscoveryDedupAbortError(Exception):
"""Stands in for data_entry_flow.AbortFlow inside the discovery stub.
HA's `ConfigFlow._async_handle_discovery_without_unique_id` RAISES on a
duplicate (the flow manager turns that into an abort result); modelling it
as a raise is what makes an un-awaited call fail loudly here.
"""
def __init__(self, reason):
super().__init__(reason)
self.reason = reason
def _install_stubs() -> None:
"""Import-time stand-ins (idempotent; same shape as the pairing tests).
@@ -54,9 +70,10 @@ def _install_stubs() -> None:
return
vol = types.ModuleType("voluptuous")
vol.Required = lambda key, default=None: key
vol.Optional = lambda key, default=None: key
vol.Schema = lambda schema: schema
# t33 F-A: a faithful mirror — the default stays ON the key, so a schema
# whose host default was dropped FAILS here instead of looking identical
# to one that carries it (the old bare-key stub hid exactly that bug).
vol_mirror.install(vol)
sys.modules["voluptuous"] = vol
ha = types.ModuleType("homeassistant")
@@ -71,12 +88,16 @@ def _install_stubs() -> None:
kwargs.pop("domain", None)
super().__init_subclass__(**kwargs)
def async_show_form(self, *, step_id, data_schema=None, errors=None):
def async_show_form(
self, *, step_id, data_schema=None, errors=None,
description_placeholders=None,
):
return {
"type": "form",
"step_id": step_id,
"errors": dict(errors or {}),
"data_schema": data_schema,
"description_placeholders": dict(description_placeholders or {}),
}
def async_create_entry(self, *, title, data, options=None):
@@ -95,11 +116,23 @@ def _install_stubs() -> None:
def _async_current_entries(self):
return _zc_holder.entries
# t32 F2: the audited helper for uniqueness-less discovery — a
# recording no-op here; the flow's own host guard does the stacking
# prevention the acceptance pins.
def _async_handle_discovery_without_unique_id(self):
def _async_in_progress(self, include_uninitialized=False, **kwargs):
return _zc_holder.in_progress
# t37 D3: HA's own helper, modelled faithfully — a COROUTINE that
# dedups uniqueness-less discovery by raising. An un-awaited call
# therefore does nothing at all, which is exactly the production bug
# this replaces.
async def _async_handle_discovery_without_unique_id(self):
_zc_holder.discovery_without_unique_id_calls += 1
if getattr(self, "_unique_id", None) is not None:
return
if self._async_current_entries():
raise DiscoveryDedupAbortError("already_configured")
await self.async_set_unique_id(_zc_holder.discovery_unique_id)
self._abort_if_unique_id_configured()
if self._async_in_progress(include_uninitialized=True):
raise DiscoveryDedupAbortError("already_in_progress")
class OptionsFlow:
def async_show_form(self, *, step_id, data_schema=None, errors=None):
@@ -134,7 +167,12 @@ def _install_stubs() -> None:
# Shared holder so tests can swap the current-entry list through the
# module object (the stub closure reads this same object).
_zc_holder = types.SimpleNamespace(entries=[], discovery_without_unique_id_calls=0)
_zc_holder = types.SimpleNamespace(
entries=[],
in_progress=[],
discovery_unique_id="default_discovery_unique_id",
discovery_without_unique_id_calls=0,
)
sys.modules[__name__]._zc_holder = _zc_holder
class AbortIfConfiguredError(Exception):
@@ -850,23 +888,27 @@ class TestMacLessDiscoveryF2:
ip_address=host, host=host, properties=props or {}
)
async def test_mac_less_records_the_helper(self, tmp_path):
def _holder(self):
return sys.modules[__name__]._zc_holder
async def test_mac_less_discovery_awaits_the_helper(self, tmp_path):
module = _form_module()
flow = module.NvidiaShieldConfigFlow()
flow.hass = FakeHass(str(tmp_path))
sys.modules[__name__]._zc_holder.discovery_without_unique_id_calls = 0
self._holder().discovery_without_unique_id_calls = 0
result = await flow.async_step_zeroconf(self._discovery())
assert result["type"] == "form" # prefill still offered once
assert (
sys.modules[__name__]._zc_holder.discovery_without_unique_id_calls
== 1
)
# t37 D3: only an AWAITED coroutine body runs — before the fix this
# counter stayed at 0 and Python warned "coroutine ... was never
# awaited", while HA's dedup never happened
assert self._holder().discovery_without_unique_id_calls == 1
# the helper claims HA's discovery sentinel for the flow
assert flow._unique_id == self._holder().discovery_unique_id
async def test_mac_less_rediscovery_does_not_stack(self, tmp_path):
"""A second MAC-less discovery of an already-configured host
aborts — one-entry outcome."""
async def test_mac_less_discovery_with_entries_aborts(self, tmp_path):
"""Real contract: any entry for the handler → already_configured."""
module = _form_module()
flow = module.NvidiaShieldConfigFlow()
flow.hass = FakeHass(str(tmp_path))
@@ -876,13 +918,45 @@ class TestMacLessDiscoveryF2:
options={CONF_PROTOCOL_HOST: "192.168.1.50"},
title="Shield",
)
sys.modules[__name__]._zc_holder.entries[:] = [entry]
self._holder().entries[:] = [entry]
result = await flow.async_step_zeroconf(self._discovery())
with pytest.raises(DiscoveryDedupAbortError) as raised:
await flow.async_step_zeroconf(self._discovery())
assert result == {"type": "abort", "reason": "already_configured"}
assert raised.value.reason == "already_configured"
# the entries list is untouched: one-entry outcome
assert len(sys.modules[__name__]._zc_holder.entries) == 1
assert len(self._holder().entries) == 1
async def test_mac_less_discovery_while_another_flow_runs_aborts(self, tmp_path):
"""The other half of the real contract: one flow at a time."""
module = _form_module()
flow = module.NvidiaShieldConfigFlow()
flow.hass = FakeHass(str(tmp_path))
self._holder().in_progress[:] = [{"flow_id": "other"}]
with pytest.raises(DiscoveryDedupAbortError) as raised:
await flow.async_step_zeroconf(self._discovery())
assert raised.value.reason == "already_in_progress"
async def test_mac_less_discovery_leaves_no_un_awaited_coroutine(
self, tmp_path
):
"""t37 D3: the helper must never be left un-awaited."""
module = _form_module()
flow = module.NvidiaShieldConfigFlow()
flow.hass = FakeHass(str(tmp_path))
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
await flow.async_step_zeroconf(self._discovery())
del flow
gc.collect()
never_awaited = [
str(w.message) for w in caught if "never awaited" in str(w.message)
]
assert never_awaited == []
class TestSeedBlockShapeF4:
@@ -942,6 +1016,11 @@ class TestZeroconfDiscovery:
)
async def test_discovery_prefills_the_user_step(self, tmp_path):
"""t33 F-A: the discovered host is the DEFAULT OF THE SCHEMA KEYS.
The old check (the attribute was stored) stayed green while the
rendered form was blank — this one reads the schema itself.
"""
module = _form_module()
flow = module.NvidiaShieldConfigFlow()
flow.hass = FakeHass(str(tmp_path))
@@ -950,9 +1029,13 @@ class TestZeroconfDiscovery:
assert result["type"] == "form"
assert result["step_id"] == "user"
# pre-fill landed: the discovered host is recorded for the schema
assert flow._discovered_host == "192.168.1.50"
assert result["data_schema"] is not None
defaults = vol_mirror.defaults_by_key(result["data_schema"])
assert defaults[CONF_ADB_HOST] == "192.168.1.50"
assert defaults[CONF_PROTOCOL_HOST] == "192.168.1.50"
# ...and a discovery-driven form says the address is already known
note = result["description_placeholders"]["discovery_note"]
assert "Discovered Shield at 192.168.1.50" in note
async def test_discovery_without_address_aborts(self, tmp_path):
module = _form_module()
+94
View File
@@ -0,0 +1,94 @@
"""A voluptuous mirror faithful enough to SEE a dropped schema default.
The test environment never installs voluptuous (the config-flow tests import
``config_flow`` against import-only stubs), so the harness has to model the two
behaviours the flow schemas depend on:
* ``Optional(field, default=value)`` / ``Required(field, default=value)`` carry
that default on the KEY object;
* keys compare by FIELD NAME only — real ``voluptuous.Marker.__eq__`` is
``self.schema == other`` (no class check at all), so
``Optional("x") == Required("x")`` is True, and ``__hash__`` is
``hash(self.schema)``. A later re-declaration of the same field therefore
collides with the FIRST key object, so a default added by a second
declaration is silently dropped.
That second property is exactly how the t33 F-A discovery-prefill bug hid from
the suite: the old stubs reduced every key to its bare string, so a schema
whose host key carried no default looked identical to one that carried the
discovered address. ``TestVoluptuousMirrorContract`` in
``tests/test_config_flow_ux.py`` pins both properties so the harness itself
cannot regress into blindness.
"""
from __future__ import annotations
UNDEFINED = object()
"""Sentinel for "this key declares no default" (real ``voluptuous.UNDEFINED``)."""
class Marker:
"""Stand-in for ``voluptuous.Marker`` (the ``Required``/``Optional`` base)."""
def __init__(self, schema, default=UNDEFINED):
self.schema = schema
self.default = default
def __hash__(self):
# real Marker hashes the field name only
return hash(self.schema)
def __eq__(self, other):
# real Marker.__eq__ is `self.schema == other`: the FIELD is the whole
# comparison — no class check, so Optional("x") == Required("x") (and
# even Marker("x") == "x") is True. That is what makes a re-declaration
# of a field collide with the first key object and drop its default.
return self.schema == other
def __repr__(self):
return f"<{type(self).__name__} {self.schema!r} default={self.default!r}>"
class Required(Marker):
"""Stand-in for ``voluptuous.Required``."""
class Optional(Marker):
"""Stand-in for ``voluptuous.Optional``."""
def install(vol) -> None:
"""Wire the mirror onto a stub ``voluptuous`` module object."""
vol.UNDEFINED = UNDEFINED
vol.Marker = Marker
vol.Required = Required
vol.Optional = Optional
vol.Schema = lambda schema: schema
def defaults_by_key(schema_dict) -> dict:
"""``{field: default}`` for a schema declared with mirror keys."""
return {key.schema: key.default for key in declared_keys(schema_dict)}
def declared_keys(schema_dict) -> list:
"""The schema's marker keys, in declaration order.
Raises on a non-marker key: a bare string key means the harness (or the
code under test) lost the KEY structure this mirror exists to check.
Test-harness bookkeeping keys (``__suggested__`` and friends) are skipped.
"""
keys = []
for key in schema_dict:
if isinstance(key, Marker):
keys.append(key)
continue
if isinstance(key, str) and key.startswith("__"):
continue
raise AssertionError(f"schema key {key!r} is not a voluptuous marker")
return keys
def declared_fields(schema_dict) -> list:
"""The field names of a schema, in declaration order."""
return [key.schema for key in declared_keys(schema_dict)]