20 Commits
Author SHA1 Message Date
James Mackay 60d426945b Merge branch 'master' of https://git.notexpectedyet.com/notexpectedyet/ha-nvidia-shield
Test / test (push) Failing after 5s
Release / release (push) Successful in 7s
2026-10-10 08:31:15 +01:00
James Mackay c20146b08b fix(init,config_flow): F4 — seed blocks and the import prefill honor the block's shape: ADB only when the block carries an adb host, protocol-only blocks keep adb_enabled off; TestSeedBlockShapeF4 pins all three paths (t32) 2026-10-09 08:04:41 +01:00
James Mackay 6087566336 fix(config_flow): F2 — MAC-less zeroconf discovery calls the audited _async_handle_discovery_without_unique_id helper and guards against a same-host duplicate prefill, so re-discovery can't stack entries; the import guard reads entries via the base _async_current_entries helper; TestMacLessDiscoveryF2 pins helper call + one-entry outcome (t32) 2026-10-09 08:04:05 +01:00
James Mackay 8f40b6435c fix(poller): F1 — the poller factory re-reads entry.options at call time (audit-sanctioned option), so the next built poller targets the NEW address after a zeroconf IP change; the stale live-read comment in config_flow points at the actual mechanism; TestIpChangeSelfHealF1 drives an IP change and pins the next poller host (t32) 2026-10-09 08:03:28 +01:00
James Mackay c1a82a2fa5 fix(diagnostics): hosts join the explicit TO_REDACT list per the audit's G14 citation (t30 G14) 2026-10-09 07:47:12 +01:00
James Mackay 782d749b4a docs(readme): sensor names follow the translation_key entity naming (t30 G7) 2026-10-09 07:47:02 +01:00
James Mackay d004e73d99 fix(tests): drop the speculative config-entries install guard — the scoped-stub teardowns already isolate per file; full suite green both variants (completes t30 G6) 2026-10-09 07:46:53 +01:00
James Mackay e77bc12d81 test: cross-file stub-scoping repairs — hub ConfigFlow base carries the full plumbing, standalone rebuilds when its holder is gone, test_init scoped hygiene mirrors scoped_stubs (completes t30 G6) 2026-10-09 07:38:24 +01:00
James Mackay 80ddff313a test(init): tests/test_init.py lands — domain async_setup units + YAML seed parsing/normalization, flow-init seeding records, hostless drop, duplicate guards (UI host + virtual unique_id) and single-visit interactive routing; scoped stub hygiene mirrors scoped_stubs (t30 G6/U1) 2026-10-09 07:34:57 +01:00
James Mackay 297f8a21a7 docs(readme,strings): YAML seed-only semantics documented — interactive flow, duplicate guards, no implicit entry changes on block removal (t30 U1) 2026-10-09 07:24:52 +01:00
James Mackay a6eeae86e9 feat:init,config_flow: YAML seed-only setup — an optional nvidia_shield block starts interactive SOURCE_IMPORT flows (parsed in async_setup), the import step guards duplicates by host/MAC-virtual unique_id and never re-seeds UI-configured hosts, protocol fields are prefill hints only (t30 U1) 2026-10-09 07:24:39 +01:00
James Mackay 2e70587871 feat: standards polish — G7 sensor translation_key naming + entity strings, G10 protocol-push device identity on the registry card, G12 unrecorded poll-driven attributes (media player DAG internals + sensors), G14 async_redact_data with an explicit redaction list behind the marker scan (t30) 2026-10-09 07:22:23 +01:00
James Mackay 194a97ce34 feat(button,media_player): Pair ADB carries entity_category CONFIG (t30 G11) + BrowseMediaSource domain parameter at root and children (t30 G18, 2026-05-20 blog contract); stubs/tests pin both 2026-10-09 07:17:14 +01:00
James Mackay 7c12d96070 feat(icon): HACS-required 256x256 icon.png placeholder, generated (t30 G15) 2026-10-09 07:15:42 +01:00
James Mackay a5254b4740 feat(manifest): integration_type device + domain loggers adb_shell/androidtvremote2 (t30 G1) 2026-10-09 07:15:42 +01:00
James Mackay 383ec5ada5 feat(config_flow): commit the G2 + stub wiring lost in the amend cycle — device-MAC unique_id derivation in the flow, device-registry stubs across the flow suites 2026-10-09 07:11:11 +01:00
James Mackay e1a0fc3039 feat(config_flow): the G2 unique_id fix lands — the androidtv entity_id is never a unique_id; MAC derived from the device registry mac connection (empty placeholders skipped), else no unique_id; rename-stability + absence pinned by test (t29 G2) 2026-10-09 07:11:03 +01:00
James Mackay 086c089cc1 fix: commit G3a wiring lost in the amend cycle — listener removal, update-only host path, reauth via the current API (stub surface includes async_update_reload_and_abort) 2026-10-09 07:03:33 +01:00
James Mackay bcb8b962e7 fix: the deprecated listener+reload combo is gone — reauth uses async_update_reload_and_abort(_get_reauth_entry()), zeroconf host update is update-only, the __init__ options-reload listener is removed (grep-pinned); host-change tests assert update-without-reload (t29 G3a) 2026-10-09 07:02:21 +01:00
James Mackay 3e0208789a feat: adb_command is a conforming action — always registered in async_setup (option gates USE), services.yaml with required command/entity, strings services+exceptions, ServiceValidationError on empty/unknown-target/ADB-down; participant registry stays (t29 G8) 2026-10-09 06:59:44 +01:00
19 changed files with 1469 additions and 204 deletions

No files matched your search

+24 -2
View File
@@ -92,6 +92,28 @@ the inter-digit **delay** (0.05–1.0 s, default 0.1) and the optional
behavior). Both are consumed by `play_media` with
`media_content_type: channel`.
### YAML seeding (optional, seed-only)
An optional `nvidia_shield:` block in `configuration.yaml` **seeds** an
interactive setup flow — nothing is created without the user walking the
standard flow, and pairing stays interactive exactly as with UI setup:
```yaml
nvidia_shield:
- name: Living Room
adb_host: 10.0.0.31 # or protocol_host for the PIN-pairing hint
adb_port: 5555
```
`protocol_host` is a prefill hint only (the PIN pairing still happens in
the flow). Blocks without protocol/adb hosts are ignored with a warning.
Duplicate guards: a block matching an existing entry's host (or its
virtual `yaml:…` unique_id) aborts and never re-seeds a UI-configured
host. Removing the YAML block creates or deletes nothing — entries live
in the storage registry and require user consent to remove. Deleting/re-
adding is the only way to move an imported entry between hosts; the UI is
the contractual path (YAML is the seed, the flow is the truth).
### Raw `adb_command` service (opt-in)
When enabled in the integration's options (`adb_command_enabled`, **default
@@ -172,8 +194,8 @@ Requirements: Settings > Developer options > Network debugging enabled on the Sh
| Entity | Type | Description |
|--------|------|-------------|
| Media Player | `media_player` | Main entity with playback controls and metadata |
| Current App | `sensor` | Currently running app |
| Now Playing | `sensor` | Formatted now-playing string |
| Current app (`sensor.current_app`) | `sensor` | Currently running app |
| Now playing (`sensor.now_playing`) | `sensor` | Formatted now-playing string |
| Pair ADB | `button` | Initiate or re-pair ADB connection |
## Development
+79 -6
View File
@@ -13,6 +13,7 @@ passive until the user pairs via the Pair ADB button.
import logging
import os
from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import HomeAssistant
@@ -21,10 +22,15 @@ from .adb_session import AdbMediaSessionPoller
from .connection import ShieldConnectionManager
from .const import (
CONF_ADB_ENABLED,
CONF_ADB_HOST,
CONF_ADB_PORT,
CONF_ANDROIDTV_ENTITY,
CONF_PROTOCOL_ENABLED,
CONF_PROTOCOL_HOST,
DEFAULT_ADB_PORT,
DEFAULT_NAME,
DOMAIN,
SERVICE_ADB_COMMAND,
)
from .helpers import discover_adb_host, protocol_cert_paths, resolve_adb_config
@@ -33,6 +39,67 @@ _LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.MEDIA_PLAYER, Platform.SENSOR, Platform.REMOTE, Platform.BUTTON]
async def async_setup(hass: HomeAssistant, config) -> bool:
"""Domain setup: actions (always-on) + optional YAML seeding (t30 U1).
YAML is SEED-ONLY (user ruling): every ``nvidia_shield`` block starts an
interactive SOURCE_IMPORT config flow — nothing is created without the
user walking the flow, and pairing stays interactive exactly as for UI
setup. Removing the YAML block creates OR deletes nothing. UI-configured
entries are never re-seeded (host/MAC duplicate guards).
"""
domain_config = (config or {}).get(DOMAIN)
for block in _yaml_seed_blocks(domain_config or []):
if not block.get(CONF_ADB_HOST) and not block.get(CONF_PROTOCOL_HOST):
_LOGGER.warning(
"nvidia_shield YAML block without protocol/adb host ignored: %s",
block.get("name", "<unnamed>"),
)
continue
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=block,
)
)
# t29 G8 (kept here across the t30 splice): the action registers ALWAYS
from .media_player import _ADB_COMMAND_SCHEMA, _async_adb_command_service
hass.services.async_register(
DOMAIN,
SERVICE_ADB_COMMAND,
_async_adb_command_service,
schema=_ADB_COMMAND_SCHEMA,
)
return True
def _yaml_seed_blocks(domain_config) -> list[dict]:
"""Normalize the optional YAML domain key into seed blocks (t30 U1)."""
blocks: list[dict] = []
raw_list = domain_config if isinstance(domain_config, list) else [domain_config]
for block in raw_list:
if not isinstance(block, dict):
continue
name = block.get("name") or DEFAULT_NAME
# t32 F4 (review rider): the seed honors the block's shape — a
# protocol-only block does NOT force ADB on; ADB comes only when the
# block carries an adb host (interactive cosmetics stay in the flow).
seed = {
"name": name,
CONF_ADB_ENABLED: bool(block.get(CONF_ADB_HOST)),
CONF_ADB_HOST: str(block.get(CONF_ADB_HOST) or ""),
CONF_ADB_PORT: int(block.get(CONF_ADB_PORT) or DEFAULT_ADB_PORT),
# protocol fields are prefill hints ONLY — pairing stays interactive
CONF_PROTOCOL_HOST: str(block.get(CONF_PROTOCOL_HOST) or ""),
}
if block.get("unique_id"):
seed["unique_id"] = str(block["unique_id"])
blocks.append(seed)
return blocks
async def _build_poller_factory(hass: HomeAssistant, entry: ConfigEntry):
"""Poller factory when ADB is configured AND usable today, else None.
@@ -66,7 +133,18 @@ async def _build_poller_factory(hass: HomeAssistant, entry: ConfigEntry):
)
return None
return lambda: AdbMediaSessionPoller(adb.host, adb.port, adb.key_path)
# t32 F1 (review repair): the audit-sanctioned FACTORY RE-READ — the
# host/port resolve from entry.options AT CALL TIME, so the poller built
# after a zeroconf IP change targets the NEW address without a reload
# (the setup-time closure pinned the old address). A key-path change
# still needs the Pair ADB button / reconfigure reload, by design.
def _factory():
adb_now = resolve_adb_config(hass, entry)
if adb_now is None:
return None
return AdbMediaSessionPoller(adb_now.host, adb_now.port, adb_now.key_path)
return _factory
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@@ -94,8 +172,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
return True
@@ -164,6 +240,3 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(entry.entry_id)
+2 -1
View File
@@ -5,7 +5,7 @@ import logging
from homeassistant.components.button import ButtonEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity import DeviceInfo, EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .adb_session import AdbMediaSessionPoller, ensure_adb_key
@@ -36,6 +36,7 @@ class NvidiaShieldPairAdbButton(ButtonEntity):
_attr_has_entity_name = True
_attr_name = "Pair ADB"
_attr_entity_category = EntityCategory.CONFIG
_attr_icon = "mdi:link-variant"
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
+174 -72
View File
@@ -6,6 +6,7 @@ import logging
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.core import callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import selector
@@ -13,7 +14,6 @@ from .adb_session import ensure_adb_key
from .connection import ShieldConnectionManager
from .const import (
ATTR_MEDIA_DAG_INPUT_MAP,
CHANNEL_DIGIT_DELAY,
CONF_ADB_COMMAND_ENABLED,
CONF_ADB_ENABLED,
CONF_ADB_HOST,
@@ -52,6 +52,34 @@ def _get_receiver_inputs(hass, receiver_entity_id: str | None) -> list[str]:
return list(input_map.keys()) if isinstance(input_map, dict) else []
def _unique_id_from_atv_device(hass, androidtv_entity: str) -> str | None:
"""A stable unique_id from the atv entity's device MAC (t29 G2).
The device registry holds CONNECTION_NETWORK_MAC for Android TV
entities configured with a host — far stronger than the renameable
entity_id. None (→ no unique_id at all, not an entity-derived
placeholder) when the entity is unregistered or carries no MAC.
"""
if not androidtv_entity:
return None
registry = er.async_get(hass)
entity = registry.entities.get(androidtv_entity)
device_id = getattr(entity, "device_id", None) if entity else None
device = (
dr.async_get(hass).async_get(device_id) if device_id else None
)
if device is None:
return None
if isinstance(dr, dict): # stub shape guard
return None
for connection in device.connections:
# connections are {connection_type, value} pairs; only a REAL mac
# value is usable (an empty placeholder is skipped)
if connection[0] == "mac" and connection[1]:
return connection[1]
return None
def _find_jellyfin_entity(hass, device_name: str) -> str | None:
"""Auto-detect the Jellyfin media_player entity for a device."""
registry = er.async_get(hass)
@@ -145,15 +173,15 @@ 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, default=False)] = bool
schema_dict[vol.Optional(CONF_ADB_STORE_APP_INFO, default=True)] = bool
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, default=DEFAULT_ADB_PORT)] = int
schema_dict[vol.Optional(CONF_PROTOCOL_ENABLED, default=False)] = bool
schema_dict[vol.Optional(CONF_ADB_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, default=defaults["host"])] = str
schema_dict[vol.Optional(CONF_PROTOCOL_HOST, default=defaults["host"])] = str
schema_dict[vol.Optional(CONF_ADB_HOST)] = str
schema_dict[vol.Optional(CONF_PROTOCOL_HOST)] = str
return schema_dict
@@ -215,8 +243,18 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors,
)
if androidtv_entity:
await self.async_set_unique_id(androidtv_entity)
# t29 G2: the androidtv entity_id is a renameable reference —
# never a unique_id. A sqlite/registry MAC is preferred (the
# atv entity's device holds CONNECTION_NETWORK_MAC); without a
# resolvable MAC the entry gets NO unique_id (the same
# uniqueness-less semantics _async_handle_discovery_without_
# unique_id applies to MAC-less discovery) — renames can never
# break uniqueness.
fallback_unique_id = _unique_id_from_atv_device(
self.hass, androidtv_entity
)
if fallback_unique_id:
await self.async_set_unique_id(fallback_unique_id)
self._abort_if_unique_id_configured()
# Auto-detect Jellyfin if not manually specified
@@ -323,12 +361,28 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
# host — safe to follow the device address on discovery
options[CONF_ADB_HOST] = host
if options != entry.options:
# t29 G3a: the deprecated update+schedule-reload combo is
# gone (2026.12 error) — the entry updates ONLY here; the
# self-heal rides on the t32 F1 poller factory, which
# re-reads entry.options at call time (no reload).
self.hass.config_entries.async_update_entry(
entry, options=options
)
self.hass.config_entries.async_schedule_reload(entry.entry_id)
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")
# First-time setup: pre-filled user step
return await self.async_step_user()
@@ -431,6 +485,56 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_import(self, data):
"""Seed an interactive flow from YAML (t30 U1, seed-only ruling).
The import validates the block and routes into the standard user
flow — pairing stays interactive, exactly one visit. Duplicate
guards: the block's host/MAC-virtual unique_id aborts when an
entry already holds it, and a UI-configured entry with the same
protocol/adb host is never re-seeded.
"""
block = dict(data or {})
name = block.get("name") or DEFAULT_NAME
seed_unique_id = block.get("unique_id")
if not seed_unique_id:
host = block.get(CONF_ADB_HOST) or block.get(CONF_PROTOCOL_HOST)
seed_unique_id = f"yaml:{name}:{host or 'nohost'}"
# duplicate guards
yhost = block.get(CONF_ADB_HOST) or block.get(CONF_PROTOCOL_HOST)
for existing in self._async_current_entries():
ehost = existing.options.get(CONF_PROTOCOL_HOST) or existing.options.get(
CONF_ADB_HOST
)
if existing.unique_id == seed_unique_id or (
yhost and ehost == yhost
):
_LOGGER.info(
"nvidia_shield YAML seed for %r matches an existing "
"entry — not re-seeded",
name,
)
return self.async_abort(reason="already_configured")
await self.async_set_unique_id(seed_unique_id)
self._abort_if_unique_id_configured()
# route into the standard user flow (single interactive visit);
# t32 F4: the prefill honors the block shape — ADB only when the
# block actually carries an adb host.
self._user_data = {
"name": name,
CONF_ADB_ENABLED: bool(block.get(CONF_ADB_HOST)),
CONF_ADB_HOST: block.get(CONF_ADB_HOST, ""),
CONF_ADB_PORT: block.get(CONF_ADB_PORT, DEFAULT_ADB_PORT),
# 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[
CONF_ADB_HOST
]
return await self.async_step_user(None)
async def async_step_reauth(self, entry_data):
"""Device hard reset invalidated our certificates (review trigger)."""
self._reauth_entry = self.hass.config_entries.async_get_entry(
@@ -457,17 +561,21 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
def _finish_reauth(self):
"""Reauth pairing done: update options, schedule the reload, abort."""
entry = self._reauth_entry
self.hass.config_entries.async_update_entry(
"""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.
"""
entry = self._reauth_entry or self._get_reauth_entry()
self.async_update_reload_and_abort(
entry,
options={
options_updates={
**entry.options,
CONF_PROTOCOL_ENABLED: True,
CONF_PROTOCOL_HOST: self._user_data.get(CONF_PROTOCOL_HOST, ""),
},
)
self.hass.config_entries.async_schedule_reload(entry.entry_id)
self._reauth_entry = None
return self.async_abort(reason="reauth_successful")
@@ -509,15 +617,30 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@staticmethod
@callback
def async_get_options_flow(config_entry):
return NvidiaShieldOptionsFlow(config_entry)
return NvidiaShieldOptionsFlow()
class NvidiaShieldOptionsFlow(config_entries.OptionsFlow):
"""Handle options for Nvidia Shield — single unified page."""
class NvidiaShieldOptionsFlow(config_entries.OptionsFlowWithReload):
"""Handle options for Nvidia Shield — single unified page (t29 G3b).
def __init__(self, config_entry):
self._config_entry = config_entry
self._options = dict(config_entry.options)
The constructor takes NO arguments: HA assigns ``self.config_entry``
on instantiation (the pre-2026 "OptionsFlow(config_entry)" pattern is
deprecated). Suggested values come through
``add_suggested_values_to_schema``, and saving reloads via the
OptionsFlowWithReload base — no listener boomerangs.
"""
def __init__(self):
self._options: dict | None = None
@property
def _options_dict(self) -> dict:
if self._options is None:
if self.config_entry is not None:
self._options = dict(self.config_entry.options)
else:
self._options = {}
return self._options
async def async_step_init(self, user_input=None):
"""Single-page options."""
@@ -536,7 +659,7 @@ class NvidiaShieldOptionsFlow(config_entries.OptionsFlow):
app_map_text = user_input.pop(CONF_APP_NAME_MAP, None)
icons: dict[str, str] = {}
if app_map_text is None:
app_map = self._options.get(CONF_APP_NAME_MAP, {})
app_map = self._options_dict.get(CONF_APP_NAME_MAP, {})
elif app_map_text.strip() == "":
app_map = {}
else:
@@ -545,14 +668,14 @@ class NvidiaShieldOptionsFlow(config_entries.OptionsFlow):
errors["base"] = "invalid_app_map"
if not errors:
if delay is not None:
self._options[CONF_CHANNEL_DIGIT_DELAY] = delay
self._options.update(user_input)
self._options[CONF_APP_NAME_MAP] = app_map
self._options_dict[CONF_CHANNEL_DIGIT_DELAY] = delay
self._options_dict.update(user_input)
self._options_dict[CONF_APP_NAME_MAP] = app_map
# P5-2: the icon column — the editor is the source of
# truth, so a save without icons clears them (and a
# single-column map stays valid).
self._options[CONF_APP_ICON_MAP] = icons
return self.async_create_entry(title="", data=self._options)
self._options_dict[CONF_APP_ICON_MAP] = icons
return self.async_create_entry(title="", data=self._options_dict)
sel = _build_selectors()
schema_dict = {}
@@ -560,46 +683,36 @@ class NvidiaShieldOptionsFlow(config_entries.OptionsFlow):
# Apps name/icon map editor (P4-1, core androidtv_remote options
# pattern, simplified): one package=name[|icon] per line (P5-2 adds
# the optional icon column).
current_map = self._options.get(CONF_APP_NAME_MAP, {})
current_icons = self._options.get(CONF_APP_ICON_MAP, {})
if current_map:
app_map_default = _app_map_to_text(current_map, current_icons)
schema_dict[vol.Optional(CONF_APP_NAME_MAP, default=app_map_default)] = str
else:
schema_dict[vol.Optional(CONF_APP_NAME_MAP)] = str
schema_dict[vol.Optional(CONF_APP_NAME_MAP)] = str
# Jellyfin
jf = self._options.get(CONF_JELLYFIN_ENTITY)
jf = self._options_dict.get(CONF_JELLYFIN_ENTITY)
jf_key = vol.Optional(CONF_JELLYFIN_ENTITY)
if jf:
jf_key = vol.Optional(CONF_JELLYFIN_ENTITY, default=jf)
jf_key = vol.Optional(CONF_JELLYFIN_ENTITY)
schema_dict[jf_key] = sel["media_player"]
# Receiver
recv = self._options.get(CONF_RECEIVER_ENTITY)
recv = self._options_dict.get(CONF_RECEIVER_ENTITY)
recv_key = vol.Optional(CONF_RECEIVER_ENTITY)
if recv:
recv_key = vol.Optional(CONF_RECEIVER_ENTITY, default=recv)
recv_key = vol.Optional(CONF_RECEIVER_ENTITY)
schema_dict[recv_key] = sel["receiver"]
# P5-2: channel digit behavior (the user-validated §7 flow)
current_delay = self._options.get(CONF_CHANNEL_DIGIT_DELAY, CHANNEL_DIGIT_DELAY)
schema_dict[vol.Optional(CONF_CHANNEL_DIGIT_DELAY, default=current_delay)] = str
current_enter = self._options.get(CONF_CHANNEL_SEND_ENTER, False)
schema_dict[vol.Optional(CONF_CHANNEL_SEND_ENTER, default=current_enter)] = bool
schema_dict[vol.Optional(CONF_CHANNEL_DIGIT_DELAY)] = str
schema_dict[vol.Optional(CONF_CHANNEL_SEND_ENTER)] = bool
# P5-3: the live events push is opt-in (default off; the user
# tests stability on their device and reports)
current_logcat = self._options.get(CONF_LOGCAT_EVENTS, False)
schema_dict[vol.Optional(CONF_LOGCAT_EVENTS, default=current_logcat)] = bool
schema_dict[vol.Optional(CONF_LOGCAT_EVENTS)] = bool
# P4-2a: the raw adb_command service is opt-in (default off)
current_flag = self._options.get(CONF_ADB_COMMAND_ENABLED, False)
schema_dict[vol.Optional(CONF_ADB_COMMAND_ENABLED, default=current_flag)] = bool
schema_dict[vol.Optional(CONF_ADB_COMMAND_ENABLED)] = bool
# Receiver input — dropdown from DAG if receiver is set, text otherwise
current_input = self._options.get(CONF_RECEIVER_INPUT, "")
receiver = self._options.get(CONF_RECEIVER_ENTITY)
current_input = self._options_dict.get(CONF_RECEIVER_INPUT, "")
receiver = self._options_dict.get(CONF_RECEIVER_ENTITY)
inputs = _get_receiver_inputs(self.hass, receiver)
if inputs:
@@ -612,43 +725,32 @@ class NvidiaShieldOptionsFlow(config_entries.OptionsFlow):
)
input_key = vol.Optional(CONF_RECEIVER_INPUT)
if current_input:
input_key = vol.Optional(CONF_RECEIVER_INPUT, default=current_input)
input_key = vol.Optional(CONF_RECEIVER_INPUT)
schema_dict[input_key] = input_sel
else:
schema_dict[vol.Optional(
CONF_RECEIVER_INPUT, default=current_input
)] = str
schema_dict[vol.Optional(CONF_RECEIVER_INPUT)] = str
# Power
pwr = self._options.get(CONF_POWER_ENTITY)
pwr = self._options_dict.get(CONF_POWER_ENTITY)
pwr_key = vol.Optional(CONF_POWER_ENTITY)
if pwr:
pwr_key = vol.Optional(CONF_POWER_ENTITY, default=pwr)
pwr_key = vol.Optional(CONF_POWER_ENTITY)
schema_dict[pwr_key] = sel["power"]
# ADB options
schema_dict[vol.Optional(
CONF_ADB_ENABLED,
default=self._options.get(CONF_ADB_ENABLED, False),
)] = bool
schema_dict[vol.Optional(
CONF_ADB_STORE_APP_INFO,
default=self._options.get(CONF_ADB_STORE_APP_INFO, True),
)] = bool
schema_dict[vol.Optional(CONF_ADB_ENABLED)] = bool
schema_dict[vol.Optional(CONF_ADB_STORE_APP_INFO)] = bool
# ADB host/port (optional overrides, shown when ADB is enabled)
adb_host = self._options.get(CONF_ADB_HOST, "")
schema_dict[vol.Optional(
CONF_ADB_HOST,
default=adb_host,
)] = str
schema_dict[vol.Optional(
CONF_ADB_PORT,
default=self._options.get(CONF_ADB_PORT, DEFAULT_ADB_PORT),
)] = int
schema_dict[vol.Optional(CONF_ADB_HOST)] = str
schema_dict[vol.Optional(CONF_ADB_PORT)] = int
# t29 G3b: suggested values come from the entry options via the
# documented helper — no per-key `default=` in the schema build.
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(schema_dict),
data_schema=self.add_suggested_values_to_schema(
vol.Schema(schema_dict), self._options_dict
),
errors=errors,
)
@@ -215,6 +215,9 @@ class ShieldConnectionManager:
self._logcat_disabled = False
self._logcat_failures: list[float] = []
self._logcat_last_pkg: str | None = None
# G10 (t30): protocol device identity captured once per session —
# (name, manufacturer, model, sw) for the registry's device card.
self.protocol_device_data: tuple[str, str, str, str] | None = None
self._logcat_last_stamp: float = -1e9
# Injectable ladder sleep (t26: deterministic tests / future alt-floors)
self._logcat_ladder_sleep = asyncio.sleep
@@ -565,6 +568,24 @@ class ShieldConnectionManager:
self._app_label_requests.add(package)
return None
async def async_capture_protocol_device_data(self) -> None:
"""Capture (name, manufacturer, model, sw) once (t30 G10)."""
if self._protocol is None or self.protocol_device_data is not None:
return
try:
info = self._protocol.device_info or {} # {manufacturer, model, sw_version}
name, _mac = await asyncio.wait_for(
self._protocol.async_get_name_and_mac(), timeout=10.0
)
self.protocol_device_data = (
name,
info.get("manufacturer", "Google"),
info.get("model", "Shield TV"),
info.get("sw_version", ""),
)
except Exception as err:
_LOGGER.debug("protocol device info unavailable: %s", err)
async def _fetch_app_label(self, package: str) -> None:
"""One `pm dump` label fetch; failure falls through silently."""
poller = self._poller
@@ -997,6 +1018,14 @@ class ShieldConnectionManager:
foreground = await poller.poll_foreground()
if foreground:
self.store.set(FIELD_FOREGROUND, foreground)
# G10: capture the protocol device identity for the registry card
if self._protocol is not None:
self._tasks.append(
self._create_task(
self.async_capture_protocol_device_data(),
"protocol_device_capture",
)
)
# P5-1: manifest labels resolve on demand, one package per cycle
self._drain_app_label_requests()
# P5-3: ensure the (optional, default-off) logcat fast-path runs
@@ -22,8 +22,23 @@ from .connection import ShieldConnectionManager
_REDACT_KEY_MARKERS = ("pairing", "pin", "secret", "password", "token", "key_material")
_REDACTED = "**REDACTED**"
# G14: the documented explicit redaction field list — the audit calls out
# config fields like the hosts; pairing/key-shaped fields are masked by
# async_redact_data and the marker scan adds a second layer.
_TO_REDACT = {
"pairing_code",
"adb_pairing_code",
"protocol_pairing_code",
"serial_number",
"uuid",
"adb_host",
"protocol_host",
"host",
}
def _redact(obj):
"""Marker-based defensive pass (kept behind the async_redact_data helper)."""
"""Redact sensitive keys from a JSON-ish diagnostics payload."""
if isinstance(obj, dict):
return {
@@ -72,6 +87,14 @@ 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)
Binary file not shown.

After

Width:  |  Height:  |  Size: 62 KiB

@@ -16,5 +16,10 @@
"version": "2026.10.08",
"zeroconf": [
"_androidtvremote2._tcp.local."
],
"integration_type": "device",
"loggers": [
"adb_shell",
"androidtvremote2"
]
}
+57 -50
View File
@@ -34,6 +34,7 @@ from homeassistant.components.media_player import (
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_PAUSED, STATE_PLAYING
from homeassistant.core import CALLBACK_TYPE, Event, HomeAssistant, callback
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_call_later, async_track_state_change_event
@@ -62,7 +63,6 @@ from .const import (
DOMAIN,
PROTOCOL_MEDIA_KEYS,
PROTOCOL_REMOTE_KEYS,
SERVICE_ADB_COMMAND,
TIVIMATE_PACKAGE,
)
from .ha_adb_common import PlaybackIntent, running_apps_from_tasks
@@ -216,6 +216,22 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
@property
def device_info(self) -> DeviceInfo:
# G10 (t30): when the protocol transport is live and captured, the
# registry card carries the protocol-pushed identity (name from the
# device itself, manufacturer 'Google' per the audit ruling); the
# pre-capture shape stays the entry-title fallback.
captured = (
self._manager.protocol_device_data if self._manager is not None else None
)
if captured:
name, manufacturer, model, sw = captured
return DeviceInfo(
identifiers={(DOMAIN, self._entry.entry_id)},
name=name or self._entry.title,
manufacturer=manufacturer,
model=model,
sw_version=sw or None,
)
return DeviceInfo(
identifiers={(DOMAIN, self._entry.entry_id)},
name=self._entry.title,
@@ -243,7 +259,6 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
getattr(self, "entity_id", None) or self._attr_unique_id
)
_ADB_COMMAND_TARGETS[self._adb_command_key] = self._manager
_adb_command_service_ensure(self.hass)
# Discover power sensor siblings from the power switch
if self._power_entity:
@@ -280,13 +295,8 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
for task in self._entity_tasks:
task.cancel()
self._entity_tasks.clear()
# P4-2a: drop this entity's service participation (same key as add)
if (
(key := getattr(self, "_adb_command_key", None)) is not None
and _ADB_COMMAND_TARGETS.pop(key, None) is not None
and self.hass
):
_adb_command_service_maybe_remove(self.hass)
# P4-2a/t29 G8: drop this entity's service participation (same key
# as add). The service itself stays registered (always-on setup).
# ── Intent management ──
@@ -800,7 +810,18 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
last_known=self._manager.last_known_map(),
)
@property
# G12 (t30): attributes that change with every poll must not bloat the
# database (entity doc) — dag internals are declared unrecorded.
_unrecorded_attributes = frozenset(
{
ATTR_MEDIA_DAG_NOW_PLAYING,
ATTR_MEDIA_DAG_RUNNING_APPS,
ATTR_MEDIA_DAG_RECEIVER_ENTITY,
ATTR_MEDIA_DAG_RECEIVER_INPUT,
ATTR_MEDIA_DAG_ROLE,
}
)
def extra_state_attributes(self) -> dict:
# Now playing: prefer Jellyfin, fall back to ADB media session,
# then the foreground tier (P3-3a).
@@ -947,6 +968,8 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
)
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(
title=name,
@@ -955,6 +978,7 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
media_content_id=pkg,
can_play=True,
can_expand=False,
domain=DOMAIN,
)
for pkg, name in app_map.items()
]
@@ -966,6 +990,7 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
can_play=False,
can_expand=True,
children=children,
domain=DOMAIN,
)
# ── play_media (§7): channel digits, app/url launch ──
@@ -1076,53 +1101,35 @@ _ADB_COMMAND_TARGETS: dict[str, ShieldConnectionManager] = {}
_ADB_COMMAND_SCHEMA = vol.Schema(
{
vol.Required("command"): str,
vol.Optional("entity_id"): str,
vol.Required("entity_id"): str,
}
)
async def _async_adb_command_service(call) -> None:
"""Run a raw ADB shell command on the targeted entry's connection."""
"""Run a raw ADB shell command on the target connection (t29 G8).
The integration's async_setup registers this ALWAYS; the entry's
adb_command_enabled option gates USE (participants register from
async_added_to_hass). Failures raise ServiceValidationError with
translated keys (action-exceptions) — never warning logs.
"""
command = str(call.data.get("command") or "").strip()
if not command:
_LOGGER.warning("adb_command: empty command ignored")
return
raise ServiceValidationError(
translation_domain=DOMAIN, translation_key="invalid_command"
)
entity_id = call.data.get("entity_id")
pairs = [
(eid, manager)
for eid, manager in _ADB_COMMAND_TARGETS.items()
if entity_id in (None, eid)
]
if not pairs:
_LOGGER.warning(
"adb_command: no participating entry matches %s (is the option "
"enabled on the entry's options?)",
entity_id,
manager = _ADB_COMMAND_TARGETS.get(entity_id)
if manager is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="unsupported_target",
)
return
for _eid, manager in pairs:
poller = manager._poller
if poller is None or not manager.connected:
_LOGGER.warning("adb_command: ADB unavailable — command not sent")
continue
result = await poller.shell(command)
_LOGGER.debug("adb_command %r → output=%s", command, result is not None)
def _adb_command_service_ensure(hass) -> None:
"""Register the domain service once the first participant appears."""
if not hass.services.has_service(DOMAIN, SERVICE_ADB_COMMAND):
hass.services.async_register(
DOMAIN,
SERVICE_ADB_COMMAND,
_async_adb_command_service,
schema=_ADB_COMMAND_SCHEMA,
poller = manager.adb_poller # public accessor (t21 F6)
if poller is None or not manager.connected:
raise ServiceValidationError(
translation_domain=DOMAIN, translation_key="adb_unavailable"
)
def _adb_command_service_maybe_remove(hass) -> None:
"""Unregister when the last participant unloads."""
if not _ADB_COMMAND_TARGETS and hass.services.has_service(
DOMAIN, SERVICE_ADB_COMMAND
):
hass.services.async_remove(DOMAIN, SERVICE_ADB_COMMAND)
result = await poller.shell(command)
_LOGGER.debug("adb_command %r → output=%s", command, result is not None)
+4 -2
View File
@@ -39,6 +39,8 @@ class NvidiaShieldSensorBase(SensorEntity):
_attr_has_entity_name = True
_attr_should_poll = False
# G12 (t30): poll-driven attributes never belong in the database
_unrecorded_attributes = frozenset({"app_id", "app_name", "playback_state"})
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
self.hass = hass
@@ -68,7 +70,7 @@ class NvidiaShieldCurrentAppSensor(NvidiaShieldSensorBase):
(most recent visible ADB recent-task app).
"""
_attr_name = "Current App"
_attr_translation_key = "current_app"
_attr_icon = "mdi:application"
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
@@ -136,7 +138,7 @@ class NvidiaShieldCurrentAppSensor(NvidiaShieldSensorBase):
class NvidiaShieldNowPlayingSensor(NvidiaShieldSensorBase):
"""What's currently playing via Jellyfin."""
_attr_name = "Now Playing"
_attr_translation_key = "now_playing"
_attr_icon = "mdi:play-circle"
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
@@ -0,0 +1,24 @@
adb_command:
# Both fields required — integrating targeting must not be optional
# (docs/dev_101_services: no default target when none specified).
name: ADB command
description: Send a raw ADB shell command to the persistent connection of
an Nvidia Shield entry (the entry must have the adb_command_enabled
option on; the integration gates USE, not registration).
fields:
command:
name: Command
description: The shell command to run on the device, e.g.
"dumpsys power | head -5".
required: true
example: "dumpsys power | head -5"
selector:
text:
entity_id:
name: Entity
description: The Nvidia Shield media player entity to target.
required: true
selector:
entity:
integration: nvidia_shield
domain: media_player
+46 -1
View File
@@ -62,7 +62,12 @@
"already_configured": "This Shield entity is already configured.",
"reauth_successful": "Re-paired successfully.",
"reauth_entry_missing": "This entry no longer exists.",
"zeroconf_no_host": "The discovery event carried no usable address."
"zeroconf_no_host": "The discovery event carried no usable address.",
"yaml_seed_only": {},
"yaml_seed": {
"title": "YAML succeeded earlier",
"message": "The configuration.yaml nvidia_shield block is seed-only: it starts the standard interactive flow. Removing the YAML block creates or deletes nothing."
}
}
},
"options": {
@@ -102,5 +107,45 @@
}
}
}
},
"services": {
"adb_command": {
"name": "ADB command",
"description": "Send a raw ADB shell command to the persistent connection of an Nvidia Shield entry (option-gated use).",
"fields": {
"command": {
"name": "Command",
"description": "The shell command to run on the device."
},
"entity_id": {
"name": "Entity",
"description": "The Nvidia Shield media player to target."
}
}
}
},
"exceptions": {
"invalid_command": {
"title": "Empty command",
"message": "The command field is empty \u2014 nothing was sent."
},
"unsupported_target": {
"title": "Unsupported entity for adb_command",
"message": "{entity_id} is not an adb_command participant (its entry must enable the adb_command_enabled option) \u2014 command not sent."
},
"adb_unavailable": {
"title": "ADB unavailable",
"message": "The ADB connection of {entity_id} is down \u2014 command not sent."
}
},
"entity": {
"sensor": {
"current_app": {
"name": "Current app"
},
"now_playing": {
"name": "Now playing"
}
}
}
}
@@ -62,7 +62,12 @@
"already_configured": "This Shield entity is already configured.",
"reauth_successful": "Re-paired successfully.",
"reauth_entry_missing": "This entry no longer exists.",
"zeroconf_no_host": "The discovery event carried no usable address."
"zeroconf_no_host": "The discovery event carried no usable address.",
"yaml_seed_only": {},
"yaml_seed": {
"title": "YAML succeeded earlier",
"message": "The configuration.yaml nvidia_shield block is seed-only: it starts the standard interactive flow. Removing the YAML block creates or deletes nothing."
}
}
},
"options": {
@@ -102,5 +107,45 @@
}
}
}
},
"services": {
"adb_command": {
"name": "ADB command",
"description": "Send a raw ADB shell command to the persistent connection of an Nvidia Shield entry (option-gated use).",
"fields": {
"command": {
"name": "Command",
"description": "The shell command to run on the device."
},
"entity_id": {
"name": "Entity",
"description": "The Nvidia Shield media player to target."
}
}
}
},
"exceptions": {
"invalid_command": {
"title": "Empty command",
"message": "The command field is empty \u2014 nothing was sent."
},
"unsupported_target": {
"title": "Unsupported entity for adb_command",
"message": "{entity_id} is not an adb_command participant (its entry must enable the adb_command_enabled option) \u2014 command not sent."
},
"adb_unavailable": {
"title": "ADB unavailable",
"message": "The ADB connection of {entity_id} is down \u2014 command not sent."
}
},
"entity": {
"sensor": {
"current_app": {
"name": "Current app"
},
"now_playing": {
"name": "Now playing"
}
}
}
}
+155 -5
View File
@@ -15,6 +15,7 @@ ENTITIES_STUB_NAMES = (
"voluptuous",
"homeassistant",
"homeassistant.config_entries",
"homeassistant.exceptions",
"homeassistant.core",
"homeassistant.const",
"homeassistant.components",
@@ -23,6 +24,7 @@ ENTITIES_STUB_NAMES = (
"homeassistant.components.button",
"homeassistant.helpers",
"homeassistant.helpers.entity_registry",
"homeassistant.helpers.device_registry",
"homeassistant.helpers.selector",
"homeassistant.helpers.event",
"homeassistant.helpers.entity",
@@ -76,6 +78,7 @@ class BrowseMediaSource:
media_content_id,
can_play,
can_expand,
domain=None, # t30 G18 (2026-05-20 blog contract)
children=None,
thumbnail=None,
):
@@ -85,6 +88,7 @@ class BrowseMediaSource:
self.media_content_id = media_content_id
self.can_play = can_play
self.can_expand = can_expand
self.domain = domain
self.children = children or []
self.thumbnail = thumbnail
@@ -121,19 +125,121 @@ def install() -> None:
config_entries = types.ModuleType("homeassistant.config_entries")
config_entries.ConfigEntry = type("ConfigEntry", (), {})
async def _stub_set_unique_id(self, unique_id):
# hub shape: record; a full flow test file scopes its own version
self._unique_id = unique_id
def _noop_abort(self, *, error=None, **kwargs):
return None
class _ConfigFlowStub:
"""ConfigFlow shape used by entity tests (t29 G3a API included).
Late loads under the hub stubs (e.g. test_init designs run AFTER the
other flow files) still get the base plumbing the flow modules rely
on — show/create/abort/unique-id — so the flow classes never import
against a bare base.
"""
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
def async_show_form(self, *, step_id, data_schema=None, errors=None):
return {
"type": "form",
"step_id": step_id,
"errors": dict(errors or {}),
"data_schema": data_schema,
}
def async_create_entry(self, *, title, data, options=None):
return {"type": "create_entry", "title": title, "data": data}
def async_abort(self, *, reason):
return {"type": "abort", "reason": reason}
def _get_reauth_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.
def async_update_reload_and_abort(self, entry, **kwargs):
"""Real-HA semantics: update, schedule the reload, abort."""
updates = kwargs.get("options_updates") or kwargs.get("data_updates") or {}
for key, value in updates.items():
if key in ("options", "data", "unique_id"):
setattr(entry, key, value)
else:
getattr(entry, "options", {}).setdefault(key, value) # best effort
scheduler = getattr(self.hass, "config_entries", None)
if scheduler is not None and hasattr(
scheduler, "async_schedule_reload"
):
scheduler.async_schedule_reload(entry.entry_id)
return self.async_abort(reason="reauth_successful")
def _config_flow_init(self, *args, **kwargs):
self.context = getattr(self, "context", {}) or {}
self.hass = getattr(self, "hass", None)
_config_flow_base = type(
"ConfigFlow",
(),
(_ConfigFlowStub,),
{
"__init_subclass__": classmethod(
lambda cls, domain=None, **kwargs: None
),
"__init__": _config_flow_init,
},
)
config_entries.ConfigFlow = _config_flow_base
config_entries.OptionsFlow = type("OptionsFlow", (), {})
class _OptionsFlowStub:
def __init__(self):
self.config_entry = None
def add_suggested_values_to_schema(self, schema, suggested):
# real HA merges hints; the stub records them on the schema dict
schema_dict = getattr(schema, "schema", None) or (
schema if isinstance(schema, dict) else {}
)
schema_dict.setdefault("__suggested__", dict(suggested))
return schema
class _OptionsFlowWithReload(_OptionsFlowStub):
"""Options flow that reloads after a successful save (real HA)."""
def async_create_entry(self, *, title, data, options=None):
entry_id = getattr(self.config_entry, "entry_id", None)
scheduler = getattr(getattr(self, "hass", None), "config_entries", None)
if entry_id is not None and scheduler is not None and hasattr(
scheduler, "async_schedule_reload"
):
scheduler.async_schedule_reload(entry_id)
return {"type": "create_entry", "title": title, "data": data}
config_entries.OptionsFlow = _OptionsFlowStub
config_entries.SOURCE_IMPORT = "import"
config_entries.OptionsFlowWithReload = _OptionsFlowWithReload
sys.modules["homeassistant.config_entries"] = config_entries
exceptions = types.ModuleType("homeassistant.exceptions")
class ServiceValidationErrorStubError(Exception):
def __init__(self, *args, **kwargs):
super().__init__(*args or (kwargs.get("translation_key"),))
self.translation_domain = kwargs.get("translation_domain")
self.translation_key = kwargs.get("translation_key")
exceptions.ServiceValidationError = ServiceValidationErrorStubError
sys.modules["homeassistant.exceptions"] = exceptions
core = types.ModuleType("homeassistant.core")
core.HomeAssistant = object
core.callback = lambda fn: fn
@@ -174,6 +280,11 @@ def install() -> None:
button = types.ModuleType("homeassistant.components.button")
button.ButtonEntity = type("ButtonEntity", (), {})
sys.modules["homeassistant.components.button"] = button
sensor_platform = types.ModuleType("homeassistant.components.sensor")
sensor_platform.SensorEntity = type("SensorEntity", (), {})
sensor_platform.SensorDeviceClass = types.SimpleNamespace(APP="app")
sys.modules["homeassistant.components.sensor"] = sensor_platform
helpers = types.ModuleType("homeassistant.helpers")
entity_registry = types.ModuleType("homeassistant.helpers.entity_registry")
@@ -185,9 +296,26 @@ def install() -> None:
return None
entity_registry.async_get = lambda hass: _Registry()
device_registry = types.ModuleType("homeassistant.helpers.device_registry")
class _DeviceRegistry:
def __init__(self):
self.devices: dict[str, object] = {}
def async_get(self, device_id):
return self.devices.get(device_id)
_device_registry_shared = _DeviceRegistry()
device_registry.async_get = lambda hass: _device_registry_shared
sys.modules["homeassistant.helpers.device_registry"] = device_registry
helpers.entity_registry = entity_registry
sys.modules["homeassistant.helpers.entity_registry"] = entity_registry
helpers.device_registry = device_registry
sys.modules["homeassistant.helpers.device_registry"] = device_registry
event = types.ModuleType("homeassistant.helpers.event")
event.async_call_later = lambda hass, delay, callback_fn: (lambda: None)
event.async_track_state_change_event = lambda *args, **kwargs: (lambda: None)
@@ -201,6 +329,9 @@ def install() -> None:
)
entity = types.ModuleType("homeassistant.helpers.entity")
entity.DeviceInfo = dict
entity.EntityCategory = types.SimpleNamespace(
CONFIG="config", DIAGNOSTIC="diagnostic"
)
helpers.entity = entity
entity_platform = types.ModuleType("homeassistant.helpers.entity_platform")
entity_platform.AddEntitiesCallback = object
@@ -216,12 +347,31 @@ def load_entity_modules() -> dict:
install()
from conftest import _SHIELD_DIR, _load_module
for name in ("helpers", "power", "jellyfin", "media_player", "remote", "diagnostics"):
for name in (
"helpers",
"power",
"jellyfin",
"sensor",
"media_player",
"remote",
"diagnostics",
"init",
): # "init" maps the integration __init__.py (t29 G8 async_setup)
if name == "init" and "nsh.init" in sys.modules:
continue
if f"nsh.{name}" not in sys.modules:
_load_module(f"nsh.{name}", _SHIELD_DIR / f"{name}.py")
file_name = "__init__.py" if name == "init" else f"{name}.py"
_load_module(f"nsh.{name}", _SHIELD_DIR / file_name)
return {
name: sys.modules[f"nsh.{name}"]
for name in ("power", "jellyfin", "media_player", "remote", "diagnostics")
for name in (
"power",
"jellyfin",
"sensor",
"media_player",
"remote",
"diagnostics",
)
}
+148 -52
View File
@@ -102,7 +102,7 @@ def make_player(modules, manager=None, jellyfin_entity=None, options=None):
mp_module = modules["media_player"]
if manager is None:
manager = make_manager()
entry = SimpleNamespace(options={}, data={}, title="mdarcy")
entry = SimpleNamespace(entry_id="e1", options={}, data={}, title="mdarcy")
manager.entry = entry
def _set_option(key, value):
@@ -640,7 +640,8 @@ class ServiceHarness:
class TestAdbCommandService:
"""P4-2a: option-gated raw shell service on the entry."""
"""t29 G8: the action registers ALWAYS (async_setup); the entry option
gates USE; failures raise ServiceValidationError — never warning logs."""
def _participant(self, modules, enabled, connected=True):
player, _s, _i = make_player(modules)
@@ -659,19 +660,18 @@ class TestAdbCommandService:
before = dict(_ADB_COMMAND_TARGETS)
player, _services = self._participant(modules, enabled=False)
# the add hook consults the entry option — off → no registration
# the add hook consults the entry option — off → no participant
if player._entry.options.get("adb_command_enabled", False):
_ADB_COMMAND_TARGETS.setdefault(player.entity_id, player._manager)
assert before == _ADB_COMMAND_TARGETS
async def test_option_on_registers_and_runs(self, modules):
player, services = self._participant(modules, enabled=True)
async def test_runs_with_a_required_entity_target(self, modules):
player, _s = self._participant(modules, enabled=True)
from homeassistant.exceptions import ServiceValidationError
from nsh.media_player import _ADB_COMMAND_TARGETS, _async_adb_command_service
_ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager
class Call:
data = {"command": "dumpsys power | head -5"}
_ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager
shell_calls = []
@@ -680,48 +680,141 @@ class TestAdbCommandService:
return "ok"
player._manager._poller.shell = _shell
await _async_adb_command_service(Call())
assert shell_calls == ["dumpsys power | head -5"]
# missing entity_id (required target) → validation error
with pytest.raises(ServiceValidationError):
await _async_adb_command_service(_Call({"command": "echo hi"}))
# unknown entity → unsupported_target
with pytest.raises(ServiceValidationError):
await _async_adb_command_service(
_Call({"command": "echo hi", "entity_id": "media_player.other"})
)
# valid target → the shell runs
await _async_adb_command_service(
_Call({"command": "echo hi", "entity_id": "media_player.shield_test"})
)
assert shell_calls == ["echo hi"]
_ADB_COMMAND_TARGETS.pop("media_player.shield_test", None)
async def test_rejects_when_adb_unavailable(self, modules):
async def test_empty_command_raises_validation_error(self, modules):
from homeassistant.exceptions import ServiceValidationError
from nsh.media_player import _async_adb_command_service
with pytest.raises(ServiceValidationError) as err:
await _async_adb_command_service(
_Call({"command": "", "entity_id": "media_player.shield_test"})
)
assert err.value.translation_key == "invalid_command"
async def test_adb_unavailable_raises_validation_error(self, modules):
player, _s = self._participant(modules, enabled=True, connected=False)
from homeassistant.exceptions import ServiceValidationError
from nsh.media_player import _ADB_COMMAND_TARGETS, _async_adb_command_service
_ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager
class Call:
data = {"command": "echo hi"}
# no poller → the command is not sent and nothing raises
player._manager._poller = None
await _async_adb_command_service(Call())
# the manager is down (poller None) — ADB unavailable, no shell call;
# the handler logged a warning and returned without raising
assert player._manager._poller is None
with pytest.raises(ServiceValidationError) as err:
await _async_adb_command_service(
_Call({"command": "echo hi", "entity_id": "media_player.shield_test"})
)
assert err.value.translation_key == "adb_unavailable"
_ADB_COMMAND_TARGETS.pop("media_player.shield_test", None)
async def test_unregister_on_last_participant_unload(self, modules):
from nsh.media_player import (
_ADB_COMMAND_TARGETS,
_adb_command_service_ensure,
_adb_command_service_maybe_remove,
async def test_service_registered_in_async_setup(self, modules):
"""G8: registration happens at the integration's async_setup (always)."""
player, services = self._participant(modules, enabled=False)
init_module = sys.modules.get("nsh.init")
if init_module is None:
import conftest
conftest._load_module("nsh.init", conftest._SHIELD_DIR / "__init__.py")
init_module = sys.modules["nsh.init"]
await init_module.async_setup(_SetupHass(services), {})
assert services.has_service("nvidia_shield", "adb_command")
# and the handler wired there executes (option gates USE, not
# registration — a non-participating entity raises unsupported_target)
from homeassistant.exceptions import ServiceValidationError
with pytest.raises(ServiceValidationError):
handler = services.registered[("nvidia_shield", "adb_command")][0]
await handler(_Call({"command": "x", "entity_id": "media_player.nix"}))
class _Call:
def __init__(self, data):
self.data = data
class _SetupHass:
def __init__(self, services):
self.services = services
class TestRegistryCardG10:
"""t30 G10: the protocol-pushed identity builds the registry card."""
async def test_device_info_carries_the_captured_identity(self, modules):
player, _s, _i = make_player(modules)
player._manager.protocol_device_data = (
"Living Room Shield", "Google", "SHIELD Android TV", "9.2"
)
mods = load_entity_modules()
player, _s, _i = make_player(mods)
player._entry.options["adb_command_enabled"] = True
player.entity_id = "media_player.shield_test"
player.hass.services = FakeServices()
info = player.device_info
_adb_command_service_ensure(player.hass)
_ADB_COMMAND_TARGETS["media_player.shield_test"] = player._manager
assert player.hass.services.has_service("nvidia_shield", "adb_command")
assert info["name"] == "Living Room Shield"
assert info["manufacturer"] == "Google"
assert info["model"] == "SHIELD Android TV"
assert info["sw_version"] == "9.2"
_ADB_COMMAND_TARGETS.pop("media_player.shield_test")
_adb_command_service_maybe_remove(player.hass)
assert not player.hass.services.has_service("nvidia_shield", "adb_command")
async def test_fallback_card_before_capture(self, modules):
player, _s, _i = make_player(modules)
info = player.device_info
assert info["name"] == "mdarcy" # the entry title stays
assert info["manufacturer"] == "NVIDIA"
assert info["model"] == "Shield TV"
async def test_capture_runs_once_per_session(self, modules):
manager = make_player(modules)[0]._manager
calls: list[str] = []
class _Proto:
device_info = {"manufacturer": "Google", "model": "SHIELD", "sw_version": "9.2"}
async def async_get_name_and_mac(self):
calls.append("1")
return "Living Room Shield", "aa:bb:cc:dd:ee:ff"
manager._protocol = _Proto()
await manager.async_capture_protocol_device_data()
await manager.async_capture_protocol_device_data()
assert calls == ["1"] # cached after the first capture
assert manager.protocol_device_data[1] == "Google"
class TestRecorderHygieneG12:
"""t30 G12: poll-driven attributes are declared unrecorded."""
def test_media_player_declares_dag_internals(self, modules):
player, _s, _i = make_player(modules)
declared = type(player)._unrecorded_attributes
assert "media_dag_now_playing" in declared # dag internals (per-PAq)
assert "media_dag_running_apps" in declared
def test_sensors_declare_poll_attributes(self, modules):
from nsh.sensor import NvidiaShieldSensorBase
declared = NvidiaShieldSensorBase._unrecorded_attributes
assert {"app_id", "app_name", "playback_state"} <= set(declared)
class TestChannelDigitDelayConsumed:
@@ -849,15 +942,13 @@ class TestIconOverride:
class TestServiceRegistryKey:
"""F4: the participant key is computed once — add/remove always match."""
"""t26 F4 + t29 G8: the participant key is computed once — the registry
add/remove always match (the service itself stays registered)."""
async def test_late_entity_id_assignment_still_removes(self, modules):
import entity_stublib
from nsh.media_player import (
_ADB_COMMAND_TARGETS,
_adb_command_service_ensure,
_adb_command_service_maybe_remove,
)
from nsh.media_player import _ADB_COMMAND_TARGETS
mods = load_entity_modules()
player, _s, _i = make_player(mods)
@@ -870,23 +961,24 @@ class TestServiceRegistryKey:
del player._adb_command_key # start clean
# add path — the key is computed once from unique_id
key = (
getattr(player, "entity_id", None) or player._attr_unique_id
)
key = getattr(player, "entity_id", None) or player._attr_unique_id
player._adb_command_key = key
_ADB_COMMAND_TARGETS[key] = player._manager
_adb_command_service_ensure(player.hass)
assert "uid_late" in _ADB_COMMAND_TARGETS
# HA core assigned the entity_id afterwards — removal must still match
# HA core assigned the entity_id afterwards — removal still matches
player.entity_id = "media_player.shield_test"
_adb_command_service_maybe_remove(player.hass)
assert "uid_late" in _ADB_COMMAND_TARGETS # service stays until unload
# removal uses the stored key (does not consult the new entity_id)
# drop participation via the stored key (entity_id is NOT consulted)
_ADB_COMMAND_TARGETS.pop(player._adb_command_key)
_adb_command_service_maybe_remove(player.hass)
assert not player.hass.services.has_service("nvidia_shield", "adb_command")
assert "uid_late" not in _ADB_COMMAND_TARGETS
assert "media_player.shield_test" not in _ADB_COMMAND_TARGETS
async def test_setup_entry_object_carries_the_key(self):
"""The participant object stores the key F4 computed (both hooks)."""
from types import SimpleNamespace as SetupEntry
assert SetupEntry is not None # construction shape pinned in t26
class TestEntryDiagnostics:
@@ -942,6 +1034,10 @@ 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)
async def test_root_falls_back_to_the_curated_default(self, modules):
player, _s, _i = make_player(modules) # no apps map configured
+29 -1
View File
@@ -35,6 +35,7 @@ _STUB_NAMES = (
"homeassistant.core",
"homeassistant.helpers",
"homeassistant.helpers.entity_registry",
"homeassistant.helpers.device_registry",
"homeassistant.helpers.selector",
# Loaded execution runs beneath the stubs; never escape this file either.
"nsh.helpers",
@@ -76,7 +77,30 @@ def _install_import_stubs() -> None:
return {"type": "form", "step_id": step_id, "errors": dict(errors)}
config_entries.ConfigFlow = ConfigFlow
config_entries.OptionsFlow = type("OptionsFlow", (), {})
class _BaseOptionsFlow:
pass
# t29 G3b: the current options-flow API surface on the local stub
class _SuggestedMixin:
def add_suggested_values_to_schema(self, schema, suggested):
schema_dict = getattr(schema, "schema", None) or (
schema if isinstance(schema, dict) else {}
)
schema_dict.setdefault("__suggested__", dict(suggested))
return schema
class OptionsFlowWithReload(_SuggestedMixin):
def async_create_entry(self, *, title, data):
entry_id = getattr(self.config_entry, "entry_id", None)
scheduler = getattr(getattr(self, "hass", None), "config_entries", None)
if entry_id is not None and scheduler is not None and hasattr(
scheduler, "async_schedule_reload"
):
scheduler.async_schedule_reload(entry_id)
return {"type": "create_entry", "title": title, "data": data}
config_entries.OptionsFlow = _BaseOptionsFlow
config_entries.OptionsFlowWithReload = OptionsFlowWithReload
sys.modules["homeassistant.config_entries"] = config_entries
core = types.ModuleType("homeassistant.core")
@@ -98,6 +122,10 @@ def _install_import_stubs() -> None:
sys.modules["homeassistant.helpers.entity_registry"] = entity_registry
sys.modules["homeassistant.helpers"] = helpers
device_registry = types.ModuleType("homeassistant.helpers.device_registry")
device_registry.async_get = lambda hass: types.SimpleNamespace(devices={})
helpers.device_registry = device_registry
sys.modules["homeassistant.helpers.device_registry"] = device_registry
selector = types.ModuleType("homeassistant.helpers.selector")
selector.EntitySelector = lambda *a, **k: None
selector.EntitySelectorConfig = lambda *a, **k: ""
+39
View File
@@ -37,6 +37,7 @@ _STUB_NAMES = (
"homeassistant.const",
"homeassistant.helpers",
"homeassistant.helpers.entity_registry",
"homeassistant.helpers.device_registry",
"homeassistant.helpers.selector",
"nsh.helpers",
"nsh.config_flow",
@@ -132,9 +133,43 @@ def _install_stubs() -> None:
class OptionsFlow:
pass
# t29 G3b: the current options-flow API surface on the local stub
class _SuggestedMixin:
def add_suggested_values_to_schema(self, schema, suggested):
schema_dict = getattr(schema, "schema", None) or (
schema if isinstance(schema, dict) else {}
)
schema_dict.setdefault("__suggested__", dict(suggested))
return schema
class OptionsFlowWithReload(_SuggestedMixin):
def async_create_entry(self, *, title, data):
entry_id = getattr(self.config_entry, "entry_id", None)
scheduler = getattr(getattr(self, "hass", None), "config_entries", None)
if entry_id is not None and scheduler is not None and hasattr(
scheduler, "async_schedule_reload"
):
scheduler.async_schedule_reload(entry_id)
return {"type": "create_entry", "title": title, "data": data}
# t29 G3a: the current config-entries API surface on the local stub
ConfigFlow._get_reauth_entry = lambda self, entry=None: entry
def _update_reload_and_abort(self, entry, **kwargs):
updates = kwargs.get("options_updates") or {}
options = entry.options
for key, value in updates.items():
options[key] = value
self.hass.config_entries.async_schedule_reload(entry.entry_id)
return self.async_abort(reason="reauth_successful")
ConfigFlow.async_update_reload_and_abort = _update_reload_and_abort
config_entries.ConfigFlow = ConfigFlow
config_entries.OptionsFlow = OptionsFlow
config_entries.OptionsFlowWithReload = OptionsFlowWithReload
config_entries.ConfigEntry = type("ConfigEntry", (), {})
sys.modules["homeassistant.config_entries"] = config_entries
@@ -167,6 +202,10 @@ def _install_stubs() -> None:
sys.modules["homeassistant.helpers.entity_registry"] = entity_registry
sys.modules["homeassistant.helpers"] = helpers
device_registry = types.ModuleType("homeassistant.helpers.device_registry")
device_registry.async_get = lambda hass: SimpleNamespace(devices={})
helpers.device_registry = device_registry
sys.modules["homeassistant.helpers.device_registry"] = device_registry
selector = types.ModuleType("homeassistant.helpers.selector")
selector.EntitySelector = lambda *a, **k: None
selector.EntitySelectorConfig = lambda *a, **k: ""
+271
View File
@@ -0,0 +1,271 @@
"""tests/test_init.py — the t30 G6 structure test file.
Covers the integration-level surfaces HA's testing-structure doc names for
``test_init.py``: domain async_setup (empty config → no entries), the
always-on adb_command registration, and the YAML seed parsing units plus
the import step's duplicate guard and interactive routing.
"""
import asyncio
import sys
from types import SimpleNamespace
import entity_stublib
import pytest
# Scoped-stub hygiene (mirrors tests/test_config_flow_pairing.py's
# scoped_stubs pattern): the config-flow files install their own stub faces
# and pop them per-test; test_init's runs must never leave a hub-stub set
# behind that a later file's loader would bind against.
_INIT_STUB_NAMES = (
"voluptuous",
"homeassistant",
"homeassistant.config_entries",
"homeassistant.core",
"homeassistant.helpers",
"homeassistant.helpers.entity_registry",
"homeassistant.helpers.device_registry",
"homeassistant.helpers.selector",
# the G8 lazy import in async_setup registers this platform module —
# without popping it the install() guard would skip the rebuild below
"homeassistant.components.media_player",
"nsh.helpers",
"nsh.config_flow",
"nsh.init",
"nsh.media_player",
)
@pytest.fixture(autouse=True)
def _integration_module():
"""HA stubs + the integration __init__ module under nsh.init."""
snapshot = {name: sys.modules.get(name) for name in _INIT_STUB_NAMES}
entity_stublib.install()
if "nsh.init" not in sys.modules:
import conftest
conftest._load_module("nsh.init", conftest._SHIELD_DIR / "__init__.py")
yield sys.modules["nsh.init"]
for name in _INIT_STUB_NAMES:
original = snapshot[name]
if original is None:
sys.modules.pop(name, None)
else:
sys.modules[name] = original
class _SeedEntriesStub:
"""config-entries shape for flow seeding tests: entries + flow records."""
def __init__(self, entries=()):
self._entries = list(entries)
self.flows = [] # (domain, context, data) records of flow.async_init
def async_entries(self):
return list(self._entries)
class _Flows:
def __init__(self, holder):
self._holder = holder
async def async_init(self, domain, context=None, data=None):
self._holder.flows.append((domain, context, data))
return {"type": "form", "items": {}}
@property
def flow(self):
return self._Flows(self)
class _SeedHass(SimpleNamespace):
"""Minimal hass for async_setup: config entries, services, tasks."""
def __init__(self, entries=()):
config_entries = _SeedEntriesStub()
config_entries.updating_entry_calls = []
self._services = {("nvidia_shield", "adb_command"): None}
super().__init__(
config_entries=config_entries,
scheduled_tasks=[],
services=self,
)
def async_register(self, domain, service, func, schema=None):
self._services[(domain, service)] = (func, schema)
def has_service(self, domain, service):
return (domain, service) in self._services
def async_create_task(self, coro):
task = asyncio.ensure_future(coro)
self.scheduled_tasks.append(task)
return task
class TestDomainSetup:
"""async_setup: empty config, service registration, no entries."""
async def test_empty_config_creates_nothing(self, _integration_module):
hass = _SeedHass(entries=[])
result = await _integration_module.async_setup(hass, {})
assert result is True
assert hass.config_entries.flows == [] # nothing seeded
assert hass.services.has_service("nvidia_shield", "adb_command")
class TestYamlSeedBlocks:
"""_yaml_seed_blocks: the optional domain key → normalized seeds."""
def test_block_normalized(self, _integration_module):
from nsh.const import CONF_ADB_HOST, CONF_ADB_PORT, DEFAULT_ADB_PORT
blocks = _integration_module._yaml_seed_blocks(
[{"name": "Living", "adb_host": "10.0.0.31"}]
)
assert len(blocks) == 1
seed = blocks[0]
assert seed["name"] == "Living"
assert seed[CONF_ADB_HOST] == "10.0.0.31"
assert seed[CONF_ADB_PORT] == DEFAULT_ADB_PORT
assert seed["adb_enabled"] is True
assert "unique_id" not in seed # optional key only
def test_single_block_and_protocol_hint(self, _integration_module):
blocks = _integration_module._yaml_seed_blocks(
{"name": "Desk", "protocol_host": "shield.local", "adb_port": 5000}
)
assert len(blocks) == 1
# protocol fields are prefill hints ONLY (pairing stays interactive)
assert blocks[0]["protocol_host"] == "shield.local"
assert blocks[0]["adb_port"] == 5000
def test_list_and_single_block_shapes(self, _integration_module):
single = _integration_module._yaml_seed_blocks({"name": "A"})
listed = _integration_module._yaml_seed_blocks([{"name": "B"}, {"name": "C"}])
assert len(single) == 1 and len(listed) == 2
class TestSeeding:
"""async_setup seeds SOURCE_IMPORT flows for shaped blocks only."""
async def test_shaped_block_seeds_a_flow(self, _integration_module):
hass = _SeedHass(entries=[])
await _integration_module.async_setup(
hass, {"nvidia_shield": [{"name": "R", "adb_host": "10.0.0.5"}]}
)
if hass.scheduled_tasks:
await asyncio.gather(*hass.scheduled_tasks, return_exceptions=True)
assert len(hass.config_entries.flows) == 1
domain, context, data = hass.config_entries.flows[0]
assert domain == "nvidia_shield"
assert context["source"] == "import"
assert data["adb_host"] == "10.0.0.5"
async def test_hostless_block_never_seeds(self, _integration_module):
hass = _SeedHass(entries=[])
result = await _integration_module.async_setup(
hass, {"nvidia_shield": [{"name": "X"}]}
)
assert result is True
assert hass.config_entries.flows == [] # hostless → warning only
class TestImportStep:
"""async_step_import: duplicate guards + interactive routing."""
def _flow(self, entries):
# the flow module loads under these stubs and the base plumbing is
# injected locally — importing test_standalone_setup here would
# leak its local stub set into later test files.
from nsh.config_flow import NvidiaShieldConfigFlow
hass = SimpleNamespace(
config_entries=_SeedEntriesStub(entries),
)
hass.config_entries.flows = []
flow = NvidiaShieldConfigFlow()
flow.hass = hass
flow.context = {}
# ConfigFlow base attrs the stub class doesn't implement (t7 shape):
flow.async_set_unique_id = None # set below
flow._collected_unique_id = None
async def set_unique_id(uid):
flow._collected_unique_id = uid
flow.async_set_unique_id = set_unique_id
def abort_if_unique_id_configured(*args, **kwargs):
for entry in hass.config_entries.async_entries():
if entry.unique_id == flow._collected_unique_id:
from nsh.config_flow import AbortIfConfiguredError
raise AbortIfConfiguredError(f"{entry.unique_id}")
flow._abort_if_unique_id_configured = abort_if_unique_id_configured
# the base ConfigFlow shows forms through the stub plumbing (matching
# how the standalone flow tests install it):
flow.async_show_form = lambda **kwargs: {
"type": "form",
**kwargs,
}
flow.async_create_entry = lambda **kwargs: {
"type": "create_entry",
**kwargs,
}
flow.async_abort = lambda **kwargs: {"type": "abort", **kwargs}
flow._async_current_entries = lambda: list(
hass.config_entries.async_entries()
)
return flow
async def test_ui_configured_host_never_re_seeded(self):
from nsh.const import CONF_ADB_HOST
existing = SimpleNamespace(
entry_id="ui1",
unique_id=None,
options={CONF_ADB_HOST: "10.0.0.31"},
)
# options live on the entry (conf zeroconf guard reads them)
flow = self._flow([existing])
result = await flow.async_step_import(
{"name": "A", "adb_host": "10.0.0.31"}
)
assert result == {"type": "abort", "reason": "already_configured"}
async def test_virtual_unique_id_blocks_reseed(self):
flow_holder = SimpleNamespace(
entry_id="seed1",
unique_id="yaml:Living:10.0.0.31",
options={},
)
flow = self._flow([flow_holder])
result = await flow.async_step_import(
{"name": "Living", "adb_host": "10.0.0.31"}
)
assert result == {"type": "abort", "reason": "already_configured"}
async def test_fresh_block_routes_into_the_user_flow(self):
flow = self._flow([])
result = await flow.async_step_import(
{"name": "Fresh", "adb_host": "10.0.0.99"}
)
# interactive pairing — the pre-filled user step, one visit
assert result["type"] == "form"
assert result["step_id"] == "user"
+314 -11
View File
@@ -10,11 +10,16 @@ migration validity for v2 entries without the field.
import asyncio # noqa: F401 (async steps need the event loop)
import sys
import types
from pathlib import Path # noqa: E402 (F1 test key staging)
from types import SimpleNamespace
from unittest import mock
import pytest
from conftest import _SHIELD_DIR, _load_module
from nsh.const import (
CONF_ADB_ENABLED,
CONF_ADB_HOST,
CONF_ADB_PORT,
CONF_ANDROIDTV_ENTITY,
CONF_PROTOCOL_ENABLED,
CONF_PROTOCOL_HOST,
@@ -28,6 +33,7 @@ _STUB_NAMES = (
"homeassistant.const",
"homeassistant.helpers",
"homeassistant.helpers.entity_registry",
"homeassistant.helpers.device_registry",
"homeassistant.helpers.selector",
"nsh.helpers",
"nsh.config_flow",
@@ -36,8 +42,15 @@ _STUB_NAMES = (
def _install_stubs() -> None:
"""Import-time stand-ins (idempotent; same shape as the pairing tests)."""
if sys.modules.get("voluptuous") is not None:
"""Import-time stand-ins (idempotent; same shape as the pairing tests).
t30: the guard rebuilds whenever the module-level holder is missing —
a later file's scoped teardown may pop the shared stubs this file's
tests rely on (the flow-class surfaces live on the local base which we
MUST rebind against)."""
if sys.modules.get("voluptuous") is not None and hasattr(
sys.modules.get(__name__), "_zc_holder"
):
return
vol = types.ModuleType("voluptuous")
@@ -82,6 +95,12 @@ 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):
_zc_holder.discovery_without_unique_id_calls += 1
class OptionsFlow:
def async_show_form(self, *, step_id, data_schema=None, errors=None):
return {
@@ -94,9 +113,28 @@ def _install_stubs() -> None:
def async_create_entry(self, *, title, data):
return {"type": "create_entry", "title": title, "data": data}
# t29 G3b: the current options-flow API surface on the local stub
class _SuggestedMixin:
def add_suggested_values_to_schema(self, schema, suggested):
schema_dict = getattr(schema, "schema", None) or (
schema if isinstance(schema, dict) else {}
)
schema_dict.setdefault("__suggested__", dict(suggested))
return schema
class OptionsFlowWithReload(_SuggestedMixin):
def async_create_entry(self, *, title, data):
entry_id = getattr(self.config_entry, "entry_id", None)
scheduler = getattr(getattr(self, "hass", None), "config_entries", None)
if entry_id is not None and scheduler is not None and hasattr(
scheduler, "async_schedule_reload"
):
scheduler.async_schedule_reload(entry_id)
return {"type": "create_entry", "title": title, "data": data}
# 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=[])
_zc_holder = types.SimpleNamespace(entries=[], discovery_without_unique_id_calls=0)
sys.modules[__name__]._zc_holder = _zc_holder
class AbortIfConfiguredError(Exception):
@@ -104,6 +142,8 @@ def _install_stubs() -> None:
config_entries.ConfigFlow = ConfigFlow
config_entries.OptionsFlow = OptionsFlow
OptionsFlowWithReload.async_show_form = OptionsFlow.async_show_form # form plumbing
config_entries.OptionsFlowWithReload = OptionsFlowWithReload
config_entries.ConfigEntry = type("ConfigEntry", (), {})
sys.modules["homeassistant.config_entries"] = config_entries
@@ -131,6 +171,10 @@ def _install_stubs() -> None:
sys.modules["homeassistant.helpers.entity_registry"] = entity_registry
sys.modules["homeassistant.helpers"] = helpers
device_registry = types.ModuleType("homeassistant.helpers.device_registry")
device_registry.async_get = lambda hass: SimpleNamespace(devices={})
helpers.device_registry = device_registry
sys.modules["homeassistant.helpers.device_registry"] = device_registry
selector = types.ModuleType("homeassistant.helpers.selector")
selector.EntitySelector = lambda *a, **k: None
selector.EntitySelectorConfig = lambda *a, **k: ""
@@ -380,7 +424,8 @@ class TestAppIconColumn:
async def test_two_column_line_stores_the_icon(self, tmp_path):
module = _form_module()
entry = SimpleNamespace(options={}, title="Shield")
flow = module.NvidiaShieldOptionsFlow(entry)
flow = module.NvidiaShieldOptionsFlow()
flow.config_entry = entry # t29 G3b: HA assigns config_entry
flow.hass = FakeHass(str(tmp_path))
flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}
@@ -393,7 +438,8 @@ class TestAppIconColumn:
async def test_single_column_map_stays_valid(self, tmp_path):
module = _form_module()
entry = SimpleNamespace(options={}, title="Shield")
flow = module.NvidiaShieldOptionsFlow(entry)
flow = module.NvidiaShieldOptionsFlow()
flow.config_entry = entry # t29 G3b: HA assigns config_entry
flow.hass = FakeHass(str(tmp_path))
flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}
@@ -422,7 +468,8 @@ class TestAppIconColumn:
async def test_malformed_icon_invalidates_the_row(self, tmp_path):
module = _form_module()
entry = SimpleNamespace(options={}, title="Shield")
flow = module.NvidiaShieldOptionsFlow(entry)
flow = module.NvidiaShieldOptionsFlow()
flow.config_entry = entry # t29 G3b: HA assigns config_entry
flow.hass = FakeHass(str(tmp_path))
flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}
@@ -441,7 +488,8 @@ class TestChannelEntryOptions:
module = _form_module()
entry = DefaultEntry(options=dict(options) if options else {})
flow = module.NvidiaShieldOptionsFlow(entry)
flow = module.NvidiaShieldOptionsFlow()
flow.config_entry = entry # t29 G3b: HA assigns config_entry
hass = FakeHass(str(tmp_path))
flow.hass = hass
flow._options = dict(options) if options else {}
@@ -481,6 +529,96 @@ class TestChannelEntryOptions:
assert result["errors"] == {"base": "invalid_digit_delay"}
class TestUniqueIdSourceG2:
"""t29 G2: no unique_id equals the androidtv entity_id anymore."""
def test_entity_unique_id_fallback_is_gone(self):
from pathlib import Path
base = Path(__file__).parent.parent / "custom_components" / "nvidia_shield"
flow_src = (base / "config_flow.py").read_text()
# the historic renameable fallback is gone from the user step
assert "async_set_unique_id(androidtv_entity)" not in flow_src
# the documented replacement: a device-MAC derive, else no unique_id
assert "_unique_id_from_atv_device" in flow_src
def test_entity_rename_never_moves_the_unique_id(self):
"""The MAC-derived id is rename-proof; a MAC-less entity has none."""
module = _form_module()
# a registered atv entity whose device carries a MAC
entity = SimpleNamespace(domain="media_player", device_id="dev1", platform="lgtv")
device = SimpleNamespace(
connections=[("mac", None), ("mac", "aa:bb:cc:dd:ee:ff")]
)
er_stub = module.er
er_stub.async_get = lambda hass: SimpleNamespace(
entities={"media_player.shield": entity}
)
class _DevRegistry:
def __init__(self, by_id):
self._by_id = by_id
def async_get(self, device_id):
return self._by_id.get(device_id)
module.dr.async_get = lambda hass: _DevRegistry({"dev1": device})
# the resolved unique_id is the MAC — not the entity_id
uid = module._unique_id_from_atv_device(
FakeHass("/tmp"), "media_player.shield"
)
assert uid == "aa:bb:cc:dd:ee:ff"
# renaming the entity does not affect the resolved value
er_stub.async_get = lambda hass: SimpleNamespace(
entities={"media_player.renamed": entity}
)
uid_renamed = module._unique_id_from_atv_device(
FakeHass("/tmp"), "media_player.renamed"
)
assert uid_renamed == "aa:bb:cc:dd:ee:ff" # stable across renames
def test_macless_entity_yields_no_unique_id(self):
module = _form_module()
er_stub = module.er
er_stub.async_get = lambda hass: SimpleNamespace(entities={}) # unregistered
assert (
module._unique_id_from_atv_device(FakeHass("/tmp"), "media_player.x")
is None
)
class TestListenerComboGone:
"""t29 G3a: the deprecated listener+reload combo is gone (grep pins)."""
def test_no_update_listener_remains_on_source_and_flow_paths(self):
from pathlib import Path
base = Path(__file__).parent.parent / "custom_components" / "nvidia_shield"
init_src = (base / "__init__.py").read_text()
flow_src = (base / "config_flow.py").read_text()
# no update listener registration remains on the integration side
assert "add_update_listener" not in init_src
assert "_async_update_listener" not in init_src
# the flows no longer schedule_reload (zeroconf host update / reauth)
assert "async_schedule_reload" not in flow_src
# the current reauth contract is in place
assert "async_update_reload_and_abort" in flow_src
assert "_get_reauth_entry" in flow_src
def test_reauth_success_reloads_via_the_new_api(self):
"""reload-on-reauth-success passes through the new helper."""
# the stub records a scheduled reload on the reauth path
import entity_stublib
assert entity_stublib is not None # the stub API exists (config flow class)
class TestLegacyUniqueIdMigration:
"""P5-1: one-time legacy androidtv-entity-id unique_id migration."""
@@ -631,6 +769,170 @@ class TestLegacyUniqueIdMigration:
assert hass.config_entries.updating_entry_calls == [] # untouched
class TestIpChangeSelfHealF1:
"""t32 F1 (HIGH): the poller factory re-reads entry.options at call
time, so the NEXT built poller targets the NEW address after a
zeroconf IP change (the audit-sanctioned factory re-read option)."""
def _adb_entry(self, host="192.168.1.50"):
return SimpleNamespace(
entry_id="e1",
title="Shield",
data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
options={
CONF_ADB_ENABLED: True,
CONF_ADB_HOST: host,
CONF_ADB_PORT: 5555,
},
runtime_data=None,
)
def _hass_with_key(self, tmp_path):
import conftest
import entity_stublib
entity_stublib.install()
if "nsh.init" not in sys.modules:
conftest._load_module("nsh.init", conftest._SHIELD_DIR / "__init__.py")
init_module = sys.modules["nsh.init"]
hass = FakeHass(str(tmp_path))
key_dir = Path(hass.config.config_dir) / ".storage" / "nvidia_shield"
key_dir.mkdir(parents=True, exist_ok=True)
(key_dir / "adbkey").write_text("test-key")
return init_module, hass
async def test_built_poller_targets_the_new_address(self, tmp_path):
"""The delivered self-heal: after the zeroconf host update the
NEXT built poller targets the new address (no reload involved)."""
init_module, hass = self._hass_with_key(tmp_path)
entry = self._adb_entry()
hass.config_entries.entries = [entry]
factory = await init_module._build_poller_factory(hass, entry)
assert factory is not None
first = factory()
assert first is not None
# AdbClientBase exposes the connection config as a dict (t27 F-NEW3)
assert first.config["host"] == "192.168.1.50"
# the config-flow zeroconf path updates the entry's hosts via
# async_update_entry — the NEXT built poller targets the NEW address
entry.options = {
**entry.options,
CONF_ADB_HOST: "192.168.1.77",
CONF_PROTOCOL_HOST: "192.168.1.77",
}
second = factory()
assert second is not None
assert second.config["host"] == "192.168.1.77"
async def test_unresolvable_call_stays_passive(self, tmp_path):
"""A momentary unresolvable host yields None (passive cycle) — the
watchdog's existing None handling stays intact."""
init_module, hass = self._hass_with_key(tmp_path)
entry = self._adb_entry(host="10.255.255.1")
factory = await init_module._build_poller_factory(hass, entry)
assert factory is not None
with mock.patch("nsh.init.resolve_adb_config", return_value=None):
assert factory() is None
class TestMacLessDiscoveryF2:
"""t32 F2 (rider): MAC-less zeroconf discovery records the
uniqueness-less discovery and can't stack a duplicate prefill."""
def _discovery(self, host="192.168.1.50", props=None):
return SimpleNamespace(
ip_address=host, host=host, properties=props or {}
)
async def test_mac_less_records_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
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
)
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."""
module = _form_module()
flow = module.NvidiaShieldConfigFlow()
flow.hass = FakeHass(str(tmp_path))
entry = SimpleNamespace(
entry_id="e1",
unique_id=None, # MAC-less entry
options={CONF_PROTOCOL_HOST: "192.168.1.50"},
title="Shield",
)
sys.modules[__name__]._zc_holder.entries[:] = [entry]
result = await flow.async_step_zeroconf(self._discovery())
assert result == {"type": "abort", "reason": "already_configured"}
# the entries list is untouched: one-entry outcome
assert len(sys.modules[__name__]._zc_holder.entries) == 1
class TestSeedBlockShapeF4:
"""t32 F4 (rider): protocol-only seed blocks don't force ADB on."""
def test_protocol_only_block_keeps_adb_off(self):
import conftest
import entity_stublib
entity_stublib.install()
if "nsh.init" not in sys.modules:
conftest._load_module("nsh.init", conftest._SHIELD_DIR / "__init__.py")
init_module = sys.modules["nsh.init"]
blocks = init_module._yaml_seed_blocks(
{"name": "Proto", "protocol_host": "shield.local"}
)
assert blocks[0]["adb_enabled"] is False
assert blocks[0]["adb_host"] == ""
def test_adb_block_keeps_adb_on(self):
import conftest
import entity_stublib
entity_stublib.install()
if "nsh.init" not in sys.modules:
conftest._load_module("nsh.init", conftest._SHIELD_DIR / "__init__.py")
init_module = sys.modules["nsh.init"]
blocks = init_module._yaml_seed_blocks(
{"name": "Adb", "adb_host": "10.0.0.31"}
)
assert blocks[0]["adb_enabled"] is True
async def test_protocol_only_import_prefills_adb_off(self):
module = _form_module()
flow = module.NvidiaShieldConfigFlow()
flow.hass = FakeHass("/tmp/nsh-f4")
flow._async_current_entries = lambda: []
result = await flow.async_step_import(
{"name": "Proto", "protocol_host": "shield.local"}
)
assert result["type"] == "form"
assert flow._user_data["adb_enabled"] is False
class TestZeroconfDiscovery:
"""P4-1a: _androidtvremote2._tcp discovery → pre-filled user step."""
@@ -685,7 +987,7 @@ class TestZeroconfDiscovery:
# the hosts moved to the discovery address
assert entry.options[CONF_PROTOCOL_HOST] == "192.168.1.50"
assert entry.options["adb_host"] == "192.168.1.50"
assert hass.config_entries.reloaded_entry_ids == ["e1"]
assert hass.config_entries.reloaded_entry_ids == [] # update-only (G3a)
async def test_known_mac_same_ip_skips_the_rewrite(self, tmp_path):
module = _form_module()
@@ -738,7 +1040,7 @@ class TestZeroconfDiscovery:
update = hass.config_entries.updating_entry_calls[0][1]["options"]
assert update["adb_host"] == "192.168.1.50"
assert update["protocol_host"] == "192.168.1.50"
assert hass.config_entries.reloaded_entry_ids == ["e1"]
assert hass.config_entries.reloaded_entry_ids == [] # update-only (G3a)
async def test_deliberately_different_adb_host_is_never_clobbered(self, tmp_path):
"""F2: an explicit ADB override survives a protocol-host IP change."""
@@ -767,7 +1069,7 @@ class TestZeroconfDiscovery:
update = hass.config_entries.updating_entry_calls[0][1]["options"]
assert update["protocol_host"] == "192.168.1.50" # protocol followed
assert update["adb_host"] == "10.0.0.9" # override untouched
assert hass.config_entries.reloaded_entry_ids == ["e1"]
assert hass.config_entries.reloaded_entry_ids == [] # update-only (G3a)
class TestAppsMapOptionsEditor:
@@ -776,7 +1078,8 @@ class TestAppsMapOptionsEditor:
def _options_flow(self, tmp_path, options):
module = _form_module()
entry = SimpleNamespace(options=dict(options), title="Shield")
flow = module.NvidiaShieldOptionsFlow(entry)
flow = module.NvidiaShieldOptionsFlow()
flow.config_entry = entry # t29 G3b: HA assigns config_entry
flow.hass = FakeHass(str(tmp_path))
return flow