James Mackay
75ebeb40b0
The two address fields were the same Shield spelled twice, and every bug in the zeroconf/IP-change path came from keeping them in sync. One field now: * const.CONF_HOST replaces CONF_ADB_HOST / CONF_PROTOCOL_HOST in the setup form, the options page, the pairing forms, the reconfigure/reauth seeding and the description copy. ADB port 5555 stays the only port field; * helpers.entry_host(entry) is the single resolver (options first, then data); resolve_adb_config and the protocol settings both go through it, so both transports read the same value at call time (the t32 F1 factory re-read keeps working: the NEXT built poller sees the new address); * helpers.seed_host / seed_adb_enabled own the YAML seed's shape rule. A block may use `host` or the legacy protocol_host/adb_host keys — t41 keeps the old spelling working — and t32 F4's outcome is unchanged: a protocol-only block leaves ADB off, an address (or an explicit adb_enabled) turns it on; * the zeroconf IP-change path collapses the t21-F2 per-host dance into one rule: the discovery address IS the entry's address, so the entry is updated when it differs and left alone when it matches. The blank-self-heal and deliberately-different-ADB-host cases are gone with the duplication — a second address field cannot express them anymore, and the tests now pin that (previously "an explicit override is never clobbered"); * the options page exposes the protocol toggle it never had (the page could turn ADB off but not the protocol); * button.py (granted to this task) reads the entry's own address instead of another integration's entity. The v3 entry migration that maps existing adb_host/protocol_host onto CONF_HOST and drops the androidtv key lands in the next commit; HA runs migrations before setup, so an existing entry is never read with the old shape. Verified: 311 passed / 2 skipped, 312 / 1 with androidtvremote2, ruff clean, drift engine exit 0 on HEAD. The UI-copy contract tests (every rendered field has a label in strings.json AND translations/en.json, key trees identical, host errors name the field) drove the string edits.
1207 lines
44 KiB
Python
1207 lines
44 KiB
Python
"""P3-1 tests: standalone setup — optional androidtv entity + app fallback.
|
|
|
|
The integration no longer hard-requires an external androidtv media_player:
|
|
the protocol current-app push can be the only state source. Covers the flow
|
|
validation (no state source → form error), the standalone pairing route,
|
|
entry data omitting the field, media_player app_id/app_name fallback, and
|
|
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,
|
|
CONF_ADB_PORT,
|
|
CONF_ANDROIDTV_ENTITY,
|
|
CONF_HOST,
|
|
CONF_PROTOCOL_ENABLED,
|
|
)
|
|
|
|
_STUB_NAMES = (
|
|
"voluptuous",
|
|
"homeassistant",
|
|
"homeassistant.config_entries",
|
|
"homeassistant.core",
|
|
"homeassistant.const",
|
|
"homeassistant.helpers",
|
|
"homeassistant.helpers.entity_registry",
|
|
"homeassistant.helpers.device_registry",
|
|
"homeassistant.helpers.selector",
|
|
"nsh.helpers",
|
|
"nsh.config_flow",
|
|
"nsh.entry",
|
|
)
|
|
|
|
|
|
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).
|
|
|
|
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")
|
|
# 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")
|
|
sys.modules["homeassistant"] = ha
|
|
|
|
config_entries = types.ModuleType("homeassistant.config_entries")
|
|
|
|
class ConfigFlow:
|
|
VERSION = 1
|
|
|
|
def __init_subclass__(cls, **kwargs):
|
|
kwargs.pop("domain", None)
|
|
super().__init_subclass__(**kwargs)
|
|
|
|
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):
|
|
return {"type": "create_entry", "title": title, "data": data,
|
|
"options": options}
|
|
|
|
def async_abort(self, *, reason):
|
|
return {"type": "abort", "reason": reason}
|
|
|
|
async def async_set_unique_id(self, unique_id):
|
|
self._unique_id = unique_id
|
|
|
|
def _abort_if_unique_id_configured(self, *, error=None, **kwargs):
|
|
return None
|
|
|
|
def _async_current_entries(self):
|
|
return _zc_holder.entries
|
|
|
|
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):
|
|
return {
|
|
"type": "form",
|
|
"step_id": step_id,
|
|
"errors": dict(errors or {}),
|
|
"data_schema": data_schema,
|
|
}
|
|
|
|
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=[],
|
|
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):
|
|
pass
|
|
|
|
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
|
|
|
|
core = types.ModuleType("homeassistant.core")
|
|
core.HomeAssistant = object
|
|
core.callback = lambda fn: fn
|
|
sys.modules["homeassistant.core"] = core
|
|
|
|
ha_const = types.ModuleType("homeassistant.const")
|
|
ha_const.EVENT_HOMEASSISTANT_STOP = "homeassistant_stop"
|
|
|
|
class _Platform:
|
|
MEDIA_PLAYER = "media_player"
|
|
SENSOR = "sensor"
|
|
REMOTE = "remote"
|
|
BUTTON = "button"
|
|
|
|
ha_const.Platform = _Platform
|
|
sys.modules["homeassistant.const"] = ha_const
|
|
|
|
helpers = types.ModuleType("homeassistant.helpers")
|
|
entity_registry = types.ModuleType("homeassistant.helpers.entity_registry")
|
|
entity_registry.async_get = lambda hass: SimpleNamespace(entities={})
|
|
helpers.entity_registry = entity_registry
|
|
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: ""
|
|
selector.SelectSelector = lambda *a, **k: None
|
|
selector.SelectSelectorConfig = lambda **k: ""
|
|
selector.SelectSelectorMode = types.SimpleNamespace(DROPDOWN="dropdown")
|
|
sys.modules["homeassistant.helpers.selector"] = selector
|
|
|
|
|
|
class FakeConfigEntries:
|
|
def __init__(self):
|
|
self.updating_entry_calls: list = []
|
|
self.reloaded_entry_ids: list[str] = []
|
|
|
|
def async_get_entry(self, entry_id):
|
|
return None
|
|
|
|
def async_update_entry(self, entry, **kwargs):
|
|
self.updating_entry_calls.append((entry, kwargs))
|
|
for key, value in kwargs.items():
|
|
if key in ("options", "data", "unique_id"):
|
|
setattr(entry, key, value)
|
|
|
|
def async_schedule_reload(self, entry_id):
|
|
self.reloaded_entry_ids.append(entry_id)
|
|
|
|
|
|
class FakeHass:
|
|
def __init__(self, tmp_dir: str, states: dict | None = None):
|
|
self.config = SimpleNamespace(config_dir=tmp_dir)
|
|
self.config_entries = FakeConfigEntries()
|
|
self._states = states or {}
|
|
|
|
class _States:
|
|
def get(self, entity_id):
|
|
return self._states.get(entity_id)
|
|
|
|
self._states_ref = _States()
|
|
self._states_ref._states = self._states
|
|
self.states = self._states_ref
|
|
|
|
@staticmethod
|
|
async def async_add_executor_job(fn, *args):
|
|
return fn(*args)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def scoped_stubs():
|
|
"""Keep the import stubs inside this module's tests (t7 discipline)."""
|
|
snapshot = {name: sys.modules.get(name) for name in _STUB_NAMES}
|
|
_install_stubs()
|
|
yield
|
|
for name in _STUB_NAMES:
|
|
original = snapshot[name]
|
|
if original is None:
|
|
sys.modules.pop(name, None)
|
|
else:
|
|
sys.modules[name] = original
|
|
|
|
|
|
def _form_module():
|
|
_install_stubs()
|
|
if "nsh.helpers" not in sys.modules:
|
|
_load_module("nsh.helpers", _SHIELD_DIR / "helpers.py")
|
|
if "nsh.config_flow" not in sys.modules:
|
|
_load_module("nsh.config_flow", _SHIELD_DIR / "config_flow.py")
|
|
return sys.modules["nsh.config_flow"]
|
|
|
|
|
|
def _entry_module():
|
|
_install_stubs()
|
|
if "nsh.entry" not in sys.modules:
|
|
_load_module("nsh.entry", _SHIELD_DIR / "__init__.py")
|
|
return sys.modules["nsh.entry"]
|
|
|
|
|
|
def _record(sink):
|
|
"""An async_set_unique_id stand-in that records what the flow tried."""
|
|
|
|
async def _set(unique_id):
|
|
sink.append(unique_id)
|
|
|
|
return _set
|
|
|
|
|
|
def _user_input(**overrides) -> dict:
|
|
base = {
|
|
"name": "mdarcy",
|
|
CONF_PROTOCOL_ENABLED: False,
|
|
CONF_HOST: "",
|
|
CONF_ADB_ENABLED: False,
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
class TestStandaloneValidation:
|
|
async def test_neither_source_shows_form_error(self, tmp_path):
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
|
|
result = await flow.async_step_user(_user_input())
|
|
|
|
assert result["type"] == "form"
|
|
assert result["step_id"] == "user"
|
|
assert result["errors"] == {"base": "no_state_source"}
|
|
|
|
async def test_the_androidtv_field_is_gone_from_the_form(self, tmp_path):
|
|
"""t41: we own the stack — no other integration is a state source."""
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
|
|
result = await flow.async_step_user(None)
|
|
|
|
fields = vol_mirror.declared_fields(result["data_schema"])
|
|
assert CONF_ANDROIDTV_ENTITY not in fields
|
|
# the two native transports are what is left to choose from
|
|
assert CONF_PROTOCOL_ENABLED in fields
|
|
assert CONF_ADB_ENABLED in fields
|
|
|
|
async def test_standalone_protocol_goes_to_pairing(self, tmp_path, monkeypatch):
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
flow._user_data = {}
|
|
from nsh.connection import ShieldConnectionManager
|
|
|
|
class _Client:
|
|
async def async_generate_cert_if_missing(self):
|
|
return True
|
|
|
|
async def async_start_pairing(self):
|
|
return "8421"
|
|
|
|
monkeypatch.setattr(
|
|
ShieldConnectionManager, "build_protocol_client",
|
|
lambda *a, **k: _Client(),
|
|
)
|
|
monkeypatch.setattr(
|
|
module, "protocol_cert_paths", lambda hass: ("/tmp/a.pem", "/tmp/a.key")
|
|
)
|
|
|
|
result = await flow.async_step_user(
|
|
_user_input(**{CONF_PROTOCOL_ENABLED: True})
|
|
)
|
|
|
|
assert result["type"] == "form"
|
|
assert result["step_id"] == "protocol_pair" # routed into pairing, no error
|
|
|
|
async def test_entry_data_is_empty_so_options_own_the_config(self, tmp_path, monkeypatch):
|
|
"""t41: every value lives in options; `data` carries no leftovers."""
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
flow._user_data = {
|
|
"name": "mdarcy",
|
|
CONF_PROTOCOL_ENABLED: True,
|
|
CONF_HOST: "shield.local",
|
|
}
|
|
|
|
# skip the pairing route: drive _create_entry directly
|
|
result = flow._create_entry()
|
|
|
|
assert result["type"] == "create_entry"
|
|
assert result["data"] == {}
|
|
|
|
|
|
class TestAdbOnlyStandalone:
|
|
async def test_adb_only_standalone_is_valid(self, tmp_path):
|
|
"""F2: ADB media-session mode is a third accepted state source."""
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
|
|
result = await flow.async_step_user(
|
|
_user_input(
|
|
**{
|
|
CONF_PROTOCOL_ENABLED: False,
|
|
"adb_enabled": True,
|
|
}
|
|
)
|
|
)
|
|
|
|
# routed into ADB pairing — no no_state_source error
|
|
assert result["type"] == "form"
|
|
assert result["step_id"] == "adb_pair"
|
|
assert result["errors"] == {}
|
|
|
|
async def test_error_still_fires_with_no_source_at_all(self, tmp_path):
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
|
|
result = await flow.async_step_user(
|
|
_user_input(**{"adb_enabled": False})
|
|
)
|
|
|
|
assert result["errors"] == {"base": "no_state_source"}
|
|
|
|
|
|
class TestMigrationKeepsOrAllowsAbsence:
|
|
async def test_v1_to_v2_keeps_the_androidtv_field(self, tmp_path):
|
|
module = _entry_module()
|
|
hass = FakeHass(str(tmp_path))
|
|
entry = SimpleNamespace(
|
|
version=1,
|
|
data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
|
|
options={},
|
|
title="mdarcy",
|
|
)
|
|
|
|
ok = await module.async_migrate_entry(hass, entry)
|
|
|
|
assert ok is True
|
|
assert entry.data[CONF_ANDROIDTV_ENTITY] == "media_player.shield" # kept
|
|
# options were rewritten with the protocol defaults; the data dict —
|
|
# including the androidtv field — was never touched
|
|
assert hass.config_entries.updating_entry_calls == [
|
|
(
|
|
entry,
|
|
{
|
|
"options": {CONF_PROTOCOL_ENABLED: False, CONF_HOST: ""},
|
|
"version": 2,
|
|
},
|
|
)
|
|
]
|
|
|
|
async def test_v2_without_the_field_is_valid(self, tmp_path):
|
|
"""Standalone entries born v2 must pass through migration untouched."""
|
|
module = _entry_module()
|
|
hass = FakeHass(str(tmp_path))
|
|
entry = SimpleNamespace(
|
|
version=2,
|
|
data={},
|
|
options={CONF_PROTOCOL_ENABLED: True, CONF_HOST: "shield.local"},
|
|
title="mdarcy",
|
|
)
|
|
|
|
ok = await module.async_migrate_entry(hass, entry)
|
|
|
|
assert ok is True
|
|
assert hass.config_entries.updating_entry_calls == [] # no rewrite needed
|
|
|
|
|
|
def _cmd_player(modules):
|
|
"""Standalone media_player entity via the command-routing harness."""
|
|
import test_command_routing as routing
|
|
|
|
player, _served, _intents = routing.make_player(modules)
|
|
return player, None, None
|
|
|
|
|
|
class TestAppIconColumn:
|
|
"""P5-2: the editor's optional icon column (package=name|icon)."""
|
|
|
|
async def test_two_column_line_stores_the_icon(self, tmp_path):
|
|
module = _form_module()
|
|
entry = SimpleNamespace(options={}, title="Shield")
|
|
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}
|
|
|
|
await flow.async_step_init({"app_name_map": "ar.tvplayer.tv=TiviMate|http://i.local/t.png"})
|
|
|
|
assert flow._options["app_name_map"] == {"ar.tvplayer.tv": "TiviMate"}
|
|
assert flow._options["app_icon_map"] == {"ar.tvplayer.tv": "http://i.local/t.png"}
|
|
|
|
async def test_single_column_map_stays_valid(self, tmp_path):
|
|
module = _form_module()
|
|
entry = SimpleNamespace(options={}, title="Shield")
|
|
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}
|
|
|
|
await flow.async_step_init({"app_name_map": "ar.tvplayer.tv=TiviMate"})
|
|
|
|
assert flow._options["app_name_map"] == {"ar.tvplayer.tv": "TiviMate"}
|
|
assert flow._options["app_icon_map"] == {} # blank-safe (no icons)
|
|
|
|
async def test_prefill_round_trips_the_icon_column(self):
|
|
module = _form_module()
|
|
|
|
# the serializer writes the second column through on prefill
|
|
text = module._app_map_to_text(
|
|
{"ar.tvplayer.tv": "TiviMate"},
|
|
{"ar.tvplayer.tv": "http://i.local/t.png"},
|
|
)
|
|
assert text == "ar.tvplayer.tv=TiviMate|http://i.local/t.png"
|
|
|
|
# and the parser round-trips it back
|
|
icons: dict[str, str] = {}
|
|
app_map = module._app_map_from_text(text, icons)
|
|
assert app_map == {"ar.tvplayer.tv": "TiviMate"}
|
|
assert icons == {"ar.tvplayer.tv": "http://i.local/t.png"}
|
|
|
|
async def test_malformed_icon_invalidates_the_row(self, tmp_path):
|
|
module = _form_module()
|
|
entry = SimpleNamespace(options={}, title="Shield")
|
|
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}
|
|
|
|
result = await flow.async_step_init({"app_name_map": "p=Name|has spaces"})
|
|
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {"base": "invalid_app_map"}
|
|
|
|
|
|
class TestChannelEntryOptions:
|
|
"""P5-2: digit-delay + optional send-enter on the options page."""
|
|
|
|
def _options_flow(self, tmp_path, options):
|
|
from types import SimpleNamespace as DefaultEntry
|
|
|
|
module = _form_module()
|
|
entry = DefaultEntry(options=dict(options) if options else {})
|
|
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 {}
|
|
return module, flow
|
|
|
|
async def test_digit_delay_and_send_enter_store(self, tmp_path):
|
|
module, flow = self._options_flow(tmp_path, options={})
|
|
|
|
flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
|
|
flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}
|
|
|
|
await flow.async_step_init({"channel_digit_delay": "0.3", "channel_send_enter": True})
|
|
|
|
assert flow._options["channel_digit_delay"] == 0.3
|
|
assert flow._options["channel_send_enter"] is True
|
|
|
|
async def test_digit_delay_out_of_window_rejects(self, tmp_path):
|
|
module, flow = self._options_flow(tmp_path, options={})
|
|
|
|
flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
|
|
flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}
|
|
|
|
result = await flow.async_step_init({"channel_digit_delay": "2.0"})
|
|
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {"base": "invalid_digit_delay"}
|
|
|
|
async def test_digit_delay_not_a_number_rejects(self, tmp_path):
|
|
module, flow = self._options_flow(tmp_path, options={})
|
|
|
|
flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
|
|
flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs}
|
|
|
|
result = await flow.async_step_init({"channel_digit_delay": "slow"})
|
|
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {"base": "invalid_digit_delay"}
|
|
|
|
|
|
class TestUniqueIdSource:
|
|
"""t41: no unique_id is derived from another integration, at any stage.
|
|
|
|
t29 G2 replaced the renameable entity_id with a device-MAC derive; t41
|
|
removes the entity path entirely, so the ONLY identity source left is the
|
|
protocol MAC learned during pairing (``async_set_unique_id(mac)``).
|
|
"""
|
|
|
|
def test_no_entity_derived_unique_id_remains(self):
|
|
flow_src = (_SHIELD_DIR / "config_flow.py").read_text(encoding="utf-8")
|
|
|
|
assert "_unique_id_from_atv_device" not in flow_src
|
|
assert "async_set_unique_id(androidtv_entity)" not in flow_src
|
|
# no registry lookups for another integration's devices either
|
|
assert "device_registry" not in flow_src
|
|
|
|
async def test_the_user_step_sets_no_unique_id(self, tmp_path):
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs}
|
|
recorded = []
|
|
flow.async_set_unique_id = _record(recorded)
|
|
|
|
await flow.async_step_user(_user_input(**{CONF_PROTOCOL_ENABLED: True}))
|
|
|
|
assert recorded == []
|
|
|
|
|
|
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."""
|
|
|
|
async def test_migrates_to_entry_id_without_protocol(self, tmp_path):
|
|
import entity_stublib
|
|
|
|
entity_stublib.load_entity_modules()
|
|
modules = sys.modules["nsh.media_player"]
|
|
from types import SimpleNamespace as MigrationEntry
|
|
|
|
hass = FakeHass(str(tmp_path))
|
|
entry = MigrationEntry(
|
|
entry_id="cfgid123",
|
|
data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
|
|
options={},
|
|
title="Shield",
|
|
unique_id="media_player.shield", # legacy
|
|
runtime_data=None,
|
|
)
|
|
|
|
await modules._migrate_legacy_unique_id(hass, entry)
|
|
|
|
assert entry.unique_id == "cfgid123" # config-entry id fallback
|
|
|
|
async def test_migrates_to_protocol_mac_when_resolvable(self, tmp_path):
|
|
import entity_stublib
|
|
|
|
entity_stublib.load_entity_modules()
|
|
modules = sys.modules["nsh.media_player"]
|
|
from types import SimpleNamespace as MigrationEntry
|
|
|
|
hass = FakeHass(str(tmp_path))
|
|
entry = MigrationEntry(
|
|
entry_id="cfgid123",
|
|
data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
|
|
options={},
|
|
title="Shield",
|
|
unique_id="media_player.shield",
|
|
runtime_data=None,
|
|
)
|
|
|
|
class _Manager:
|
|
async def async_protocol_device_info(self):
|
|
return "Shield TV", "aa:bb:cc:dd:ee:ff"
|
|
|
|
entry.runtime_data = _Manager()
|
|
|
|
await modules._migrate_legacy_unique_id(hass, entry)
|
|
|
|
assert entry.unique_id == "aa:bb:cc:dd:ee:ff" # MAC preferred
|
|
|
|
async def test_mac_collision_falls_back_to_entry_id(self, tmp_path):
|
|
"""t26 F4: an existing entry owning the target MAC blocks the move."""
|
|
from types import SimpleNamespace as MigrationEntry
|
|
|
|
import entity_stublib
|
|
|
|
entity_stublib.load_entity_modules()
|
|
modules = sys.modules["nsh.media_player"]
|
|
entry = MigrationEntry(
|
|
entry_id="cfgid123",
|
|
data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
|
|
options={},
|
|
title="Shield",
|
|
unique_id="media_player.shield",
|
|
runtime_data=None,
|
|
)
|
|
|
|
class _Manager:
|
|
async def async_protocol_device_info(self):
|
|
return "Shield TV", "aa:bb:cc:dd:ee:ff"
|
|
|
|
entry.runtime_data = _Manager()
|
|
|
|
class _Entries:
|
|
entries = [
|
|
SimpleNamespace(entry_id="cfgid999", unique_id="aa:bb:cc:dd:ee:ff")
|
|
]
|
|
updating_entry_calls: list = []
|
|
|
|
def async_update_entry(self, entry, **kwargs):
|
|
self.updating_entry_calls.append((entry, kwargs))
|
|
for key, value in kwargs.items():
|
|
if key in ("options", "data", "unique_id"):
|
|
setattr(entry, key, value)
|
|
|
|
entries = _Entries()
|
|
entries.hass = FakeHass(str(tmp_path))
|
|
hass = entries.hass
|
|
hass.config_entries = entries
|
|
|
|
await modules._migrate_legacy_unique_id(hass, entry)
|
|
|
|
# collision → the config-entry id is used instead
|
|
assert entry.unique_id == "cfgid123"
|
|
|
|
async def test_free_mac_is_taken_without_collision(self, tmp_path):
|
|
"""t26 F4: no other entry holds the MAC → the MAC is used."""
|
|
from types import SimpleNamespace as MigrationEntry
|
|
|
|
import entity_stublib
|
|
|
|
entity_stublib.load_entity_modules()
|
|
modules = sys.modules["nsh.media_player"]
|
|
entry = MigrationEntry(
|
|
entry_id="cfgid123",
|
|
data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
|
|
options={},
|
|
title="Shield",
|
|
unique_id="media_player.shield",
|
|
runtime_data=None,
|
|
)
|
|
|
|
class _Manager:
|
|
async def async_protocol_device_info(self):
|
|
return "Shield TV", "aa:bb:cc:dd:ee:ff"
|
|
|
|
entry.runtime_data = _Manager()
|
|
hass = FakeHass(str(tmp_path))
|
|
hass.config_entries.entries = [
|
|
SimpleNamespace(entry_id="cfgid999", unique_id="other:mac") # different MAC
|
|
]
|
|
|
|
await modules._migrate_legacy_unique_id(hass, entry)
|
|
|
|
assert entry.unique_id == "aa:bb:cc:dd:ee:ff" # free MAC taken
|
|
|
|
async def test_never_runs_twice_for_a_migrated_entry(self, tmp_path):
|
|
"""The legacy pattern no longer matches after the one-time move."""
|
|
import entity_stublib
|
|
|
|
entity_stublib.load_entity_modules()
|
|
modules = sys.modules["nsh.media_player"]
|
|
from types import SimpleNamespace as MigrationEntry
|
|
|
|
hass = FakeHass(str(tmp_path))
|
|
entry = MigrationEntry(
|
|
entry_id="cfgid123",
|
|
data={CONF_ANDROIDTV_ENTITY: "media_player.shield"},
|
|
options={},
|
|
title="Shield",
|
|
unique_id="aa:bb:cc:dd:ee:ff", # already migrated
|
|
runtime_data=None,
|
|
)
|
|
|
|
await modules._migrate_legacy_unique_id(hass, entry)
|
|
|
|
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={},
|
|
options={
|
|
CONF_ADB_ENABLED: True,
|
|
CONF_HOST: host, # t41: the entry's one address
|
|
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_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 {}
|
|
)
|
|
|
|
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))
|
|
self._holder().discovery_without_unique_id_calls = 0
|
|
|
|
result = await flow.async_step_zeroconf(self._discovery())
|
|
|
|
assert result["type"] == "form" # prefill still offered once
|
|
# 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_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))
|
|
entry = SimpleNamespace(
|
|
entry_id="e1",
|
|
unique_id=None, # MAC-less entry
|
|
options={CONF_HOST: "192.168.1.50"},
|
|
title="Shield",
|
|
)
|
|
self._holder().entries[:] = [entry]
|
|
|
|
with pytest.raises(DiscoveryDedupAbortError) as raised:
|
|
await flow.async_step_zeroconf(self._discovery())
|
|
|
|
assert raised.value.reason == "already_configured"
|
|
# the entries list is untouched: one-entry outcome
|
|
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:
|
|
"""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]["host"] == "shield.local" # t41: one address field
|
|
|
|
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."""
|
|
|
|
def _discovery(self, host="192.168.1.50", props=None):
|
|
return SimpleNamespace(
|
|
ip_address=host, host=host, properties=props or {}
|
|
)
|
|
|
|
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))
|
|
|
|
result = await flow.async_step_zeroconf(self._discovery())
|
|
|
|
assert result["type"] == "form"
|
|
assert result["step_id"] == "user"
|
|
assert flow._discovered_host == "192.168.1.50"
|
|
defaults = vol_mirror.defaults_by_key(result["data_schema"])
|
|
assert defaults[CONF_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()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
|
|
result = await flow.async_step_zeroconf(
|
|
SimpleNamespace(ip_address=None, host=None, properties={})
|
|
)
|
|
|
|
assert result == {"type": "abort", "reason": "zeroconf_no_host"}
|
|
|
|
async def test_known_mac_ip_change_updates_entry_hosts(self, tmp_path):
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
hass = FakeHass(str(tmp_path))
|
|
flow.hass = hass
|
|
|
|
entry = SimpleNamespace(
|
|
unique_id="aa:bb:cc:dd:ee:ff",
|
|
options={CONF_PROTOCOL_ENABLED: True, CONF_HOST: "10.0.0.1"},
|
|
entry_id="e1",
|
|
title="Shield",
|
|
)
|
|
sys.modules[__name__]._zc_holder.entries[:] = [entry]
|
|
|
|
result = await flow.async_step_zeroconf(
|
|
self._discovery(props={"mac": "aa:bb:cc:dd:ee:ff"})
|
|
)
|
|
|
|
assert result == {"type": "abort", "reason": "already_configured"}
|
|
# t41: the ONE address field moved to the discovery address
|
|
assert entry.options[CONF_HOST] == "192.168.1.50"
|
|
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()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
hass = FakeHass(str(tmp_path))
|
|
flow.hass = hass
|
|
|
|
# the address already holds the discovered value: nothing changes
|
|
entry = SimpleNamespace(
|
|
unique_id="aa:bb:cc:dd:ee:ff",
|
|
options={CONF_PROTOCOL_ENABLED: True, CONF_HOST: "192.168.1.50"},
|
|
entry_id="e1",
|
|
title="Shield",
|
|
)
|
|
sys.modules[__name__]._zc_holder.entries[:] = [entry]
|
|
|
|
result = await flow.async_step_zeroconf(
|
|
self._discovery(props={"mac": "aa:bb:cc:dd:ee:ff"})
|
|
)
|
|
|
|
assert result == {"type": "abort", "reason": "already_configured"}
|
|
assert hass.config_entries.updating_entry_calls == [] # no IP change
|
|
assert hass.config_entries.reloaded_entry_ids == []
|
|
|
|
async def test_the_per_host_heal_cases_collapsed_into_one_rule(self):
|
|
"""t41: the blank / self-heal / deliberate-override cases are gone.
|
|
|
|
They only existed because the address was stored twice (t21 F2). With one
|
|
field the discovery address simply IS the entry's address, so this pins
|
|
the simplification: a second address field cannot quietly come back.
|
|
"""
|
|
source = (_SHIELD_DIR / "config_flow.py").read_text(encoding="utf-8")
|
|
|
|
assert "CONF_ADB_HOST" not in source
|
|
assert "CONF_PROTOCOL_HOST" not in source
|
|
|
|
async def test_an_explicit_address_follows_the_device(self, tmp_path):
|
|
"""t41: one Shield, one address — discovery is its current source.
|
|
|
|
The old rule kept a "deliberately different ADB host" untouched; that
|
|
override is precisely what the single-field ruling removes.
|
|
"""
|
|
module = _form_module()
|
|
flow = module.NvidiaShieldConfigFlow()
|
|
hass = FakeHass(str(tmp_path))
|
|
flow.hass = hass
|
|
|
|
entry = SimpleNamespace(
|
|
unique_id="aa:bb:cc:dd:ee:ff",
|
|
options={CONF_PROTOCOL_ENABLED: True, CONF_HOST: "10.0.0.9"}, # stale
|
|
entry_id="e1",
|
|
title="Shield",
|
|
)
|
|
sys.modules[__name__]._zc_holder.entries[:] = [entry]
|
|
|
|
result = await flow.async_step_zeroconf(
|
|
self._discovery(props={"mac": "aa:bb:cc:dd:ee:ff"})
|
|
)
|
|
|
|
assert result == {"type": "abort", "reason": "already_configured"}
|
|
update = hass.config_entries.updating_entry_calls[0][1]["options"]
|
|
assert update[CONF_HOST] == "192.168.1.50"
|
|
assert hass.config_entries.reloaded_entry_ids == [] # update-only (G3a)
|
|
|
|
|
|
class TestAppsMapOptionsEditor:
|
|
"""P4-1b: the apps name/icon map editor in the options flow."""
|
|
|
|
def _options_flow(self, tmp_path, options):
|
|
module = _form_module()
|
|
entry = SimpleNamespace(options=dict(options), title="Shield")
|
|
flow = module.NvidiaShieldOptionsFlow()
|
|
flow.config_entry = entry # t29 G3b: HA assigns config_entry
|
|
flow.hass = FakeHass(str(tmp_path))
|
|
return flow
|
|
|
|
async def test_map_text_stored_as_dict(self, tmp_path):
|
|
flow = self._options_flow(tmp_path, options={})
|
|
|
|
result = await flow.async_step_init(
|
|
{"app_name_map": "ar.tvplayer.tv=TiviMate\ncom.plex.hdx=Plex"}
|
|
)
|
|
|
|
assert result["type"] == "create_entry"
|
|
assert flow._options["app_name_map"] == {
|
|
"ar.tvplayer.tv": "TiviMate",
|
|
"com.plex.hdx": "Plex",
|
|
}
|
|
|
|
async def test_invalid_map_shows_form_error(self, tmp_path):
|
|
flow = self._options_flow(tmp_path, options={})
|
|
|
|
result = await flow.async_step_init({"app_name_map": "no equals sign here"})
|
|
|
|
assert result["type"] == "form"
|
|
assert result["errors"] == {"base": "invalid_app_map"}
|
|
|
|
async def test_blank_map_clears_the_entries(self, tmp_path):
|
|
flow = self._options_flow(
|
|
tmp_path, options={"app_name_map": {"ar.tvplayer.tv": "TiviMate"}}
|
|
)
|
|
|
|
result = await flow.async_step_init({"app_name_map": " "})
|
|
|
|
assert result["type"] == "create_entry"
|
|
assert flow._options["app_name_map"] == {}
|
|
|
|
async def test_existing_map_prefills_the_text_field(self, tmp_path):
|
|
flow = self._options_flow(
|
|
tmp_path, options={"app_name_map": {"ar.tvplayer.tv": "TiviMate"}}
|
|
)
|
|
|
|
# first visit: the form carries the serialized current map as the
|
|
# field default
|
|
result = await flow.async_step_init(None)
|
|
|
|
assert result["type"] == "form"
|
|
# the editor carried the existing map through the defaults path
|
|
assert flow._options["app_name_map"] == {"ar.tvplayer.tv": "TiviMate"}
|
|
|
|
|
|
class TestAppNameMapResolution:
|
|
"""app_name resolves through the map after the entity/push tiers."""
|
|
|
|
async def test_mapped_package_resolves_the_friendly_name(self):
|
|
import entity_stublib
|
|
|
|
modules = entity_stublib.load_entity_modules()
|
|
player, _s, _i = _cmd_player(modules)
|
|
player._entry.options["app_name_map"] = {"ar.tvplayer.tv": "TiviMate"}
|
|
|
|
manager = player._manager
|
|
manager._on_current_app("ar.tvplayer.tv") # protocol push
|
|
await asyncio.sleep(0.05) # drain the call_soon_threadsafe push
|
|
|
|
assert player.app_name == "TiviMate"
|
|
assert player.app_id == "ar.tvplayer.tv" # app_id stays raw
|
|
|
|
async def test_unmapped_package_stays_raw(self):
|
|
import entity_stublib
|
|
|
|
modules = entity_stublib.load_entity_modules()
|
|
player, _s, _i = _cmd_player(modules)
|
|
player._entry.options["app_name_map"] = {"ar.tvplayer.tv": "TiviMate"}
|
|
|
|
manager = player._manager
|
|
manager._on_current_app("com.google.android.youtube.tv")
|
|
await asyncio.sleep(0.05)
|
|
|
|
assert player.app_name == "com.google.android.youtube.tv"
|
|
|
|
async def test_cached_manifest_label_is_the_second_tier(self):
|
|
"""P5-1: below the explicit map sits the device manifest label."""
|
|
import entity_stublib
|
|
|
|
modules = entity_stublib.load_entity_modules()
|
|
player, _s, _i = _cmd_player(modules)
|
|
player._entry.options["app_name_map"] = {} # no override
|
|
manager = player._manager
|
|
manager._app_labels["ar.tvplayer.tv"] = ("TiviMate", manager._clock())
|
|
|
|
manager._on_current_app("ar.tvplayer.tv")
|
|
await asyncio.sleep(0.05)
|
|
|
|
assert player.app_name == "TiviMate"
|