diff --git a/tests/test_config_flow_ux.py b/tests/test_config_flow_ux.py new file mode 100644 index 0000000..8d94859 --- /dev/null +++ b/tests/test_config_flow_ux.py @@ -0,0 +1,935 @@ +"""t33: config-flow UX — discovery prefill, no dead-end pairing, plain copy. + +Three defects from the user's live discovery run are pinned here so they cannot +come back: + +* the discovered address never reached the form — the host keys were + re-declared without a default (a no-op), so the user submitted a blank host + and was told to fix a field on a step they could not return to; +* both pairing steps were dead ends when the Shield was unreachable (no way to + finish the entry); +* the copy pointed at "the previous step" and left the ADB port looking like + required knowledge. + +Schemas are rendered through ``tests/voluptuous_mirror.py``, which keeps the +default on the KEY the way real voluptuous does, so a dropped default FAILS +here. The pre-t33 tests asserted only that ``_discovered_host`` was stored and +stayed green while the form rendered blank. +""" + +import asyncio +import json +import re +import sys +import types +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +import voluptuous_mirror as vol_mirror +from conftest import _SHIELD_DIR, _load_module +from nsh.const import ( + CONF_ADB_ENABLED, + CONF_ADB_HOST, + CONF_ADB_PORT, + CONF_ANDROIDTV_ENTITY, + CONF_PAIRING_CODE, + CONF_PROTOCOL_ENABLED, + CONF_PROTOCOL_HOST, + CONF_SKIP_PAIRING, + DEFAULT_ADB_PORT, +) + +BASE = Path(__file__).resolve().parent.parent / "custom_components" / "nvidia_shield" +DISCOVERED_HOST = "192.168.1.50" +FAKE_PAIRING_CODE = "8421" +FAKE_MAC = "aa:bb:cc:dd:ee:ff" +KEY_PATH = "/tmp/nsh-ux-adbkey" + +_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", + # the integration __init__ module binds stub classes at import time, so it + # is rebuilt per call and must never leak into a later file (t30 G6) + "nsh.init", +) + + +def _install_stubs() -> None: + """Import-time stand-ins for HA + the faithful voluptuous mirror.""" + if sys.modules.get("voluptuous") is not None: + return + + vol = types.ModuleType("voluptuous") + 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 [] + + def _async_handle_discovery_without_unique_id(self): + return None + + def _get_reauth_entry(self, entry=None): + return entry + + def async_update_reload_and_abort(self, entry, **kwargs): + for key, value in (kwargs.get("options_updates") or {}).items(): + entry.options[key] = value + self.hass.config_entries.async_schedule_reload(entry.entry_id) + return self.async_abort(reason="reauth_successful") + + class OptionsFlow: + 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 {}), + } + + 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, OptionsFlow): + 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: + scheduler.async_schedule_reload(entry_id) + return {"type": "create_entry", "title": title, "data": data} + + config_entries.ConfigFlow = ConfigFlow + config_entries.OptionsFlow = OptionsFlow + config_entries.OptionsFlowWithReload = OptionsFlowWithReload + config_entries.ConfigEntry = type("ConfigEntry", (), {}) + config_entries.SOURCE_IMPORT = "import" + 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" + ha_const.Platform = types.SimpleNamespace( + MEDIA_PLAYER="media_player", SENSOR="sensor", REMOTE="remote", + BUTTON="button", + ) + 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={}) + device_registry = types.ModuleType("homeassistant.helpers.device_registry") + device_registry.async_get = lambda hass: SimpleNamespace(devices={}) + helpers.entity_registry = entity_registry + helpers.device_registry = device_registry + sys.modules["homeassistant.helpers.entity_registry"] = entity_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") + helpers.selector = selector + sys.modules["homeassistant.helpers.selector"] = selector + sys.modules["homeassistant.helpers"] = helpers + + +class FakeConfigEntries: + def __init__(self): + self.by_id: dict = {} + self.forwarded: list = [] + self.updating_entry_calls: list = [] + self.reloaded_entry_ids: list[str] = [] + + def async_get_entry(self, entry_id): + return self.by_id.get(entry_id) + + def async_update_entry(self, entry, **kwargs): + self.updating_entry_calls.append((entry, kwargs)) + + def async_schedule_reload(self, entry_id): + self.reloaded_entry_ids.append(entry_id) + + async def async_forward_entry_setups(self, entry, platforms): + self.forwarded.append((entry, list(platforms))) + return True + + async def async_unload_platforms(self, entry, platforms): + return True + + +class FakeHass: + """Minimal hass: config dir, entries, task creation, event bus.""" + + def __init__(self, tmp_dir: str, states: dict | None = None): + self.config = SimpleNamespace(config_dir=tmp_dir) + self.config_entries = FakeConfigEntries() + self.data: dict = {} + self.tasks: list = [] + self._states = states or {} + self.states = SimpleNamespace(get=self._states.get) + self.bus = SimpleNamespace(async_listen_once=lambda *a, **k: (lambda: None)) + + @staticmethod + async def async_add_executor_job(fn, *args): + return fn(*args) + + def async_create_task(self, coro): + task = asyncio.ensure_future(coro) + self.tasks.append(task) + return task + + def async_create_background_task(self, coro, name=None): + return self.async_create_task(coro) + + +class FakeProtocolClient: + """Deterministic androidtvremote2 stand-in for the PIN pairing path.""" + + def __init__(self, fail_code: bool = False): + self.cert_generated = False + self.pairing_started = False + self.finished_codes: list[str] = [] + self.disconnects = 0 + self._fail_code = fail_code + + async def async_generate_cert_if_missing(self): + self.cert_generated = True + return True + + async def async_start_pairing(self): + self.pairing_started = True + return FAKE_PAIRING_CODE + + async def async_finish_pairing(self, pairing_code: str) -> None: + if self._fail_code or pairing_code != FAKE_PAIRING_CODE: + raise RuntimeError("device rejected the pairing code") + self.finished_codes.append(pairing_code) + + async def async_get_name_and_mac(self): + return "mdarcy", FAKE_MAC + + def disconnect(self): + self.disconnects += 1 + + +@pytest.fixture(autouse=True) +def scoped_stubs(): + """Hard-reset the stub set per test, then restore the previous modules. + + Popping first (instead of trusting an idempotence guard) keeps a stub that + another file left behind from shadowing the faithful voluptuous mirror + this file's schema assertions depend on. + """ + snapshot = {name: sys.modules.get(name) for name in _STUB_NAMES} + for name in _STUB_NAMES: + sys.modules.pop(name, None) + _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 _integration_module(): + """The integration __init__ module, rebuilt against the current stubs.""" + _install_stubs() + if "nsh.helpers" not in sys.modules: + _load_module("nsh.helpers", _SHIELD_DIR / "helpers.py") + _load_module("nsh.init", _SHIELD_DIR / "__init__.py") + return sys.modules["nsh.init"] + + +def _new_flow(module, tmp_path): + flow = module.NvidiaShieldConfigFlow() + flow.hass = FakeHass(str(tmp_path)) + return flow, flow.hass + + +def _discovery(host=DISCOVERED_HOST, props=None): + return SimpleNamespace(ip_address=host, host=host, properties=props or {}) + + +def _patch_protocol(monkeypatch, module, client=None): + """Stub the protocol client factory + cert paths; return the fake client.""" + from nsh.connection import ShieldConnectionManager + + client = client or FakeProtocolClient() + monkeypatch.setattr( + ShieldConnectionManager, "build_protocol_client", lambda *a, **k: client + ) + monkeypatch.setattr( + module, "protocol_cert_paths", + lambda hass: ("/tmp/nsh-ux.pem", "/tmp/nsh-ux.key"), + ) + return client + + +def _patch_adb(monkeypatch, module, connected=True): + """Stub the key store + the ADB handshake; return the test_connect mock.""" + from nsh.connection import ShieldConnectionManager + + monkeypatch.setattr(module, "ensure_adb_key", lambda _config_dir: KEY_PATH) + test_connect = AsyncMock(return_value=connected) + monkeypatch.setattr(ShieldConnectionManager, "test_connect", test_connect) + return test_connect + + +def _user_input(**overrides) -> dict: + base = { + "name": "mdarcy", + CONF_ANDROIDTV_ENTITY: "media_player.shield", + CONF_ADB_ENABLED: False, + CONF_ADB_HOST: "", + CONF_PROTOCOL_ENABLED: False, + CONF_PROTOCOL_HOST: "", + } + base.update(overrides) + return base + + +def _entry_from(result): + """The config-entry stand-in an entry flow's result describes.""" + return SimpleNamespace( + entry_id="e1", + title=result["title"], + data=dict(result["data"] or {}), + options=dict(result["options"] or {}), + unique_id=None, + runtime_data=None, + async_on_unload=lambda _cb: None, + async_start_reauth=lambda _hass: None, + ) + + +def _strings(): + return json.loads((BASE / "strings.json").read_text(encoding="utf-8")) + + +def _en_strings(): + return json.loads((BASE / "translations" / "en.json").read_text(encoding="utf-8")) + + +def _key_tree(node, path=()): + if isinstance(node, dict): + tree = set() + for key, value in node.items(): + tree |= _key_tree(value, (*path, key)) + return tree + return {path} + + +def _iter_string_leaves(node, path=()): + if isinstance(node, dict): + for key, value in node.items(): + yield from _iter_string_leaves(value, (*path, key)) + elif isinstance(node, str): + yield path, node + + +class TestVoluptuousMirrorContract: + """The harness must be able to SEE a dropped default (else F-A is blind).""" + + def test_a_default_on_the_declaration_is_visible(self): + vol = sys.modules["voluptuous"] + schema = {vol.Optional(CONF_ADB_HOST, default=DISCOVERED_HOST): str} + + assert vol_mirror.defaults_by_key(schema)[CONF_ADB_HOST] == DISCOVERED_HOST + + def test_a_later_redeclaration_cannot_add_the_default(self): + vol = sys.modules["voluptuous"] + # this is the t33 shape: declare bare, then "fix" it by re-declaring. + # Real voluptuous keeps the FIRST key object, so the default is dropped. + schema = {vol.Optional(CONF_ADB_HOST): str} + schema[vol.Optional(CONF_ADB_HOST, default=DISCOVERED_HOST)] = str + + assert vol_mirror.defaults_by_key(schema)[CONF_ADB_HOST] is vol_mirror.UNDEFINED + + def test_a_bare_string_key_is_rejected_loudly(self): + with pytest.raises(AssertionError): + vol_mirror.declared_fields({CONF_ADB_HOST: str}) + + +class TestDiscoveryPrefill: + """F-A / F-A2 / F-D: the discovered address reaches the rendered FORM.""" + + async def test_discovered_host_is_the_default_of_both_host_fields(self, tmp_path): + module = _form_module() + flow, _hass = _new_flow(module, tmp_path) + + result = await flow.async_step_zeroconf(_discovery()) + + assert result["type"] == "form" + assert result["step_id"] == "user" + defaults = vol_mirror.defaults_by_key(result["data_schema"]) + # schema-level, not "the attribute was stored": this is precisely what + # the old assertion missed while the form rendered blank + assert defaults[CONF_ADB_HOST] == DISCOVERED_HOST + assert defaults[CONF_PROTOCOL_HOST] == DISCOVERED_HOST + + async def test_every_host_field_carries_the_discovered_default(self, tmp_path): + """The keys are declared WITH the default — never re-declared bare.""" + module = _form_module() + flow, _hass = _new_flow(module, tmp_path) + + result = await flow.async_step_zeroconf(_discovery()) + + for field in (CONF_ADB_HOST, CONF_PROTOCOL_HOST): + keys = [ + key for key in vol_mirror.declared_keys(result["data_schema"]) + if key.schema == field + ] + assert len(keys) == 1, f"{field} is declared {len(keys)} times" + assert keys[0].default == DISCOVERED_HOST + + async def test_discovery_step_announces_the_known_address(self, tmp_path): + module = _form_module() + flow, _hass = _new_flow(module, tmp_path) + + result = await flow.async_step_zeroconf(_discovery()) + + note = result["description_placeholders"]["discovery_note"] + assert f"Discovered Shield at {DISCOVERED_HOST}" in note + + async def test_manual_setup_claims_no_discovery(self, tmp_path): + module = _form_module() + flow, _hass = _new_flow(module, tmp_path) + + result = await flow.async_step_user(None) + + assert result["description_placeholders"]["discovery_note"] == "" + assert ( + vol_mirror.defaults_by_key(result["data_schema"])[CONF_ADB_HOST] + is vol_mirror.UNDEFINED + ) + + async def test_seed_hint_prefills_without_claiming_a_discovery(self, tmp_path): + module = _form_module() + flow, _hass = _new_flow(module, tmp_path) + + result = await flow.async_step_import({"name": "Living", "adb_host": "10.0.0.31"}) + + assert result["step_id"] == "user" + defaults = vol_mirror.defaults_by_key(result["data_schema"]) + assert defaults[CONF_ADB_HOST] == "10.0.0.31" + assert defaults["name"] == "Living" + assert result["description_placeholders"]["discovery_note"] == "" + + async def test_standard_adb_port_is_prefilled(self, tmp_path): + """F-D: 5555 rides the form as an advanced-setting default.""" + module = _form_module() + flow, _hass = _new_flow(module, tmp_path) + + manual = await flow.async_step_user(None) + discovered = await flow.async_step_zeroconf(_discovery()) + + for result in (manual, discovered): + defaults = vol_mirror.defaults_by_key(result["data_schema"]) + assert defaults[CONF_ADB_PORT] == DEFAULT_ADB_PORT == 5555 + + async def test_error_re_render_keeps_the_discovered_prefill(self, tmp_path): + module = _form_module() + flow, _hass = _new_flow(module, tmp_path) + await flow.async_step_zeroconf(_discovery()) + + result = await flow.async_step_user( + _user_input(**{CONF_ANDROIDTV_ENTITY: "", CONF_ADB_ENABLED: False}) + ) + + assert result["errors"] == {"base": "no_state_source"} + defaults = vol_mirror.defaults_by_key(result["data_schema"]) + assert defaults[CONF_ADB_HOST] == DISCOVERED_HOST + assert defaults[CONF_PROTOCOL_HOST] == DISCOVERED_HOST + + +class TestNoDeadEndPairing: + """F-B: both pairing steps offer 'continue without pairing'.""" + + async def test_pairing_forms_offer_the_skip_and_the_field_errors_name( + self, tmp_path, monkeypatch + ): + module = _form_module() + _patch_protocol(monkeypatch, module) + flow, _hass = _new_flow(module, tmp_path) + + adb_form = await flow.async_step_adb_pair(None) + protocol_form = await flow.async_step_protocol_pair(None) + + for form in (adb_form, protocol_form): + assert CONF_SKIP_PAIRING in vol_mirror.declared_fields(form["data_schema"]) + + adb_fields = vol_mirror.declared_fields(adb_form["data_schema"]) + assert CONF_ADB_HOST in adb_fields # the field adb_no_host names + protocol_fields = vol_mirror.declared_fields(protocol_form["data_schema"]) + assert CONF_PROTOCOL_HOST in protocol_fields # protocol_no_host names this + + async def test_the_skip_option_is_off_by_default(self, tmp_path, monkeypatch): + module = _form_module() + _patch_protocol(monkeypatch, module) + flow, _hass = _new_flow(module, tmp_path) + + adb_form = await flow.async_step_adb_pair(None) + protocol_form = await flow.async_step_protocol_pair(None) + + for form in (adb_form, protocol_form): + defaults = vol_mirror.defaults_by_key(form["data_schema"]) + assert defaults[CONF_SKIP_PAIRING] is False + + async def test_skip_protocol_pairing_creates_a_passive_entry( + self, tmp_path, monkeypatch + ): + module = _form_module() + client = _patch_protocol(monkeypatch, module) + flow, _hass = _new_flow(module, tmp_path) + + form = await flow.async_step_user( + _user_input( + **{ + CONF_ANDROIDTV_ENTITY: "", + CONF_PROTOCOL_ENABLED: True, + CONF_PROTOCOL_HOST: DISCOVERED_HOST, + } + ) + ) + assert form["step_id"] == "protocol_pair" + + result = await flow.async_step_protocol_pair({CONF_SKIP_PAIRING: True}) + + assert result["type"] == "create_entry" + assert result["options"][CONF_PROTOCOL_ENABLED] is False # passive + assert client.disconnects == 1 # the half-open session is dropped + + init = _integration_module() + hass = FakeHass(str(tmp_path)) + entry = _entry_from(result) + # no protocol transport configured, no ADB → a passive manager + assert await init._build_protocol_settings(hass, entry) is None + assert await init._build_poller_factory(hass, entry) is None + # and setup still accepts the entry (no ConfigEntryNotReady, no crash) + assert await init.async_setup_entry(hass, entry) is True + await entry.runtime_data.async_stop() + + async def test_skip_adb_pairing_creates_a_passive_entry(self, tmp_path, monkeypatch): + module = _form_module() + test_connect = _patch_adb(monkeypatch, module) + flow, _hass = _new_flow(module, tmp_path) + + form = await flow.async_step_user( + _user_input( + **{ + CONF_ANDROIDTV_ENTITY: "", + CONF_ADB_ENABLED: True, + CONF_ADB_HOST: DISCOVERED_HOST, + } + ) + ) + assert form["step_id"] == "adb_pair" + + result = await flow.async_step_adb_pair({CONF_SKIP_PAIRING: True}) + + assert result["type"] == "create_entry" + assert test_connect.await_count == 0 # nothing was paired + # the intent survives, so the Pair ADB button stays available for later + assert result["options"][CONF_ADB_ENABLED] is True + + init = _integration_module() + hass = FakeHass(str(tmp_path)) + entry = _entry_from(result) + # no key material yet → no poller factory, i.e. a passive manager + assert await init._build_poller_factory(hass, entry) is None + assert await init.async_setup_entry(hass, entry) is True + await entry.runtime_data.async_stop() + + async def test_skip_adb_pairing_still_offers_the_protocol_step( + self, tmp_path, monkeypatch + ): + module = _form_module() + _patch_protocol(monkeypatch, module) + flow, _hass = _new_flow(module, tmp_path) + + form = await flow.async_step_user( + _user_input( + **{ + CONF_ANDROIDTV_ENTITY: "", + CONF_ADB_ENABLED: True, + CONF_PROTOCOL_ENABLED: True, + CONF_ADB_HOST: DISCOVERED_HOST, + } + ) + ) + assert form["step_id"] == "adb_pair" + + result = await flow.async_step_adb_pair({CONF_SKIP_PAIRING: True}) + + assert result["step_id"] == "protocol_pair" # not a dead end either + + async def test_skip_protocol_after_skip_adb_creates_the_entry( + self, tmp_path, monkeypatch + ): + module = _form_module() + _patch_protocol(monkeypatch, module) + flow, _hass = _new_flow(module, tmp_path) + + await flow.async_step_user( + _user_input( + **{ + CONF_ANDROIDTV_ENTITY: "", + CONF_ADB_ENABLED: True, + CONF_PROTOCOL_ENABLED: True, + CONF_ADB_HOST: DISCOVERED_HOST, + } + ) + ) + await flow.async_step_adb_pair({CONF_SKIP_PAIRING: True}) + + result = await flow.async_step_protocol_pair({CONF_SKIP_PAIRING: True}) + + assert result["type"] == "create_entry" + assert result["options"][CONF_PROTOCOL_ENABLED] is False + + async def test_normal_protocol_pair_path_still_creates_the_entry( + self, tmp_path, monkeypatch + ): + module = _form_module() + client = _patch_protocol(monkeypatch, module) + flow, _hass = _new_flow(module, tmp_path) + + form = await flow.async_step_user( + _user_input( + **{ + CONF_ANDROIDTV_ENTITY: "", + CONF_PROTOCOL_ENABLED: True, + CONF_PROTOCOL_HOST: DISCOVERED_HOST, + } + ) + ) + assert form["step_id"] == "protocol_pair" + + result = await flow.async_step_protocol_pair( + {CONF_PAIRING_CODE: FAKE_PAIRING_CODE} + ) + + assert result["type"] == "create_entry" + assert result["options"][CONF_PROTOCOL_ENABLED] is True + assert client.finished_codes == [FAKE_PAIRING_CODE] + assert flow._unique_id == FAKE_MAC + + async def test_normal_adb_pair_path_still_creates_the_entry(self, tmp_path, monkeypatch): + module = _form_module() + test_connect = _patch_adb(monkeypatch, module) + flow, _hass = _new_flow(module, tmp_path) + + form = await flow.async_step_user( + _user_input( + **{ + CONF_ANDROIDTV_ENTITY: "", + CONF_ADB_ENABLED: True, + CONF_ADB_HOST: DISCOVERED_HOST, + } + ) + ) + assert form["step_id"] == "adb_pair" + + result = await flow.async_step_adb_pair({CONF_ADB_PORT: 5555}) + + assert result["type"] == "create_entry" + assert result["options"][CONF_ADB_ENABLED] is True + test_connect.assert_awaited_once_with(DISCOVERED_HOST, 5555, KEY_PATH) + + async def test_adb_form_host_correction_is_accepted_in_place(self, tmp_path, monkeypatch): + """The user step is gone — a typed host must be usable where it is typed.""" + module = _form_module() + test_connect = _patch_adb(monkeypatch, module) + flow, _hass = _new_flow(module, tmp_path) + await flow.async_step_user( + _user_input(**{CONF_ANDROIDTV_ENTITY: "", CONF_ADB_ENABLED: True}) + ) + + result = await flow.async_step_adb_pair( + {CONF_ADB_HOST: "10.0.0.77", CONF_ADB_PORT: 5555} + ) + + assert result["type"] == "create_entry" + assert result["options"][CONF_ADB_HOST] == "10.0.0.77" + assert test_connect.await_args.args[0] == "10.0.0.77" + + async def test_protocol_form_host_correction_restarts_pairing(self, tmp_path, monkeypatch): + module = _form_module() + first = FakeProtocolClient() + _patch_protocol(monkeypatch, module, first) + flow, _hass = _new_flow(module, tmp_path) + await flow.async_step_user( + _user_input( + **{ + CONF_ANDROIDTV_ENTITY: "", + CONF_PROTOCOL_ENABLED: True, + CONF_PROTOCOL_HOST: DISCOVERED_HOST, + } + ) + ) + assert first.pairing_started is True + + # the host typed HERE replaces the wrong one and voids the session + second = FakeProtocolClient() + from nsh.connection import ShieldConnectionManager + + monkeypatch.setattr( + ShieldConnectionManager, "build_protocol_client", lambda *a, **k: second + ) + form = await flow.async_step_protocol_pair({CONF_PROTOCOL_HOST: "10.0.0.77"}) + + assert form["step_id"] == "protocol_pair" + assert first.disconnects == 1 + assert second.pairing_started is True + assert flow._user_data[CONF_PROTOCOL_HOST] == "10.0.0.77" + + async def test_adb_no_host_error_keeps_the_user_on_a_fixable_form( + self, tmp_path, monkeypatch + ): + module = _form_module() + _patch_adb(monkeypatch, module, connected=False) + flow, _hass = _new_flow(module, tmp_path) + await flow.async_step_user( + _user_input(**{CONF_ANDROIDTV_ENTITY: "", CONF_ADB_ENABLED: True}) + ) + + result = await flow.async_step_adb_pair({CONF_ADB_PORT: 5555}) + + assert result["errors"] == {"base": "adb_no_host"} + assert CONF_ADB_HOST in vol_mirror.declared_fields(result["data_schema"]) + + async def test_adb_failed_error_keeps_the_user_on_the_form(self, tmp_path, monkeypatch): + module = _form_module() + _patch_adb(monkeypatch, module, connected=False) + flow, _hass = _new_flow(module, tmp_path) + await flow.async_step_user( + _user_input( + **{ + CONF_ANDROIDTV_ENTITY: "", + CONF_ADB_ENABLED: True, + CONF_ADB_HOST: DISCOVERED_HOST, + } + ) + ) + + result = await flow.async_step_adb_pair({CONF_ADB_PORT: 5555}) + + assert result["errors"] == {"base": "adb_failed"} + assert CONF_SKIP_PAIRING in vol_mirror.declared_fields(result["data_schema"]) + + async def test_missing_protocol_library_reports_instead_of_crashing( + self, tmp_path, monkeypatch + ): + """No androidtvremote2 (the default test variant) must not raise.""" + module = _form_module() + from nsh.connection import ShieldConnectionManager + + monkeypatch.setattr( + ShieldConnectionManager, "build_protocol_client", lambda *a, **k: None + ) + monkeypatch.setattr( + module, "protocol_cert_paths", + lambda hass: ("/tmp/nsh-ux.pem", "/tmp/nsh-ux.key"), + ) + flow, _hass = _new_flow(module, tmp_path) + # the address IS known — only the library is missing + flow._user_data = { + "name": "mdarcy", + CONF_PROTOCOL_ENABLED: True, + CONF_PROTOCOL_HOST: DISCOVERED_HOST, + } + + result = await flow.async_step_protocol_pair(None) + + assert result["errors"] == {"base": "protocol_unavailable"} + assert CONF_SKIP_PAIRING in vol_mirror.declared_fields(result["data_schema"]) + + async def test_reauth_skip_aborts_honestly(self, tmp_path, monkeypatch): + """An existing entry cannot be re-created — say so, never fake a repair.""" + module = _form_module() + _patch_protocol(monkeypatch, module) + flow, hass = _new_flow(module, tmp_path) + entry = SimpleNamespace( + entry_id="e1", + title="mdarcy", + data={}, + options={CONF_PROTOCOL_ENABLED: True, CONF_PROTOCOL_HOST: DISCOVERED_HOST}, + ) + hass.config_entries.by_id["e1"] = entry + flow.context = {"entry_id": "e1"} + + shown = await flow.async_step_reauth({}) + assert shown["step_id"] == "reauth_confirm" + pair_form = await flow.async_step_reauth_confirm({"confirm": True}) + assert pair_form["step_id"] == "protocol_pair" + + result = await flow.async_step_protocol_pair({CONF_SKIP_PAIRING: True}) + + assert result == {"type": "abort", "reason": "reauth_skipped"} + assert entry.options[CONF_PROTOCOL_ENABLED] is True # untouched + + +class TestCopyContract: + """F-C: plain, self-consistent copy with no unreachable-step pointers.""" + + def test_no_string_sends_the_user_to_the_previous_step(self): + for path, text in _iter_string_leaves(_strings()): + assert "previous step" not in text.lower(), ".".join(path) + + def test_no_host_errors_name_the_field_and_the_discovery_prefill(self): + for file in (_strings(), _en_strings()): + errors = file["config"]["error"] + for key in ("adb_no_host", "protocol_no_host"): + text = errors[key].lower() + assert "host" in text, key + assert "discovered" in text, key + + def test_every_error_and_abort_key_the_flow_raises_exists_in_both_files(self): + source = (BASE / "config_flow.py").read_text(encoding="utf-8") + raised = set(re.findall(r'errors\["base"\] = "([a-z_]+)"', source)) + aborted = set(re.findall(r'reason="([a-z_]+)"', source)) + + assert raised, "no flow error keys found — the pattern drifted" + assert aborted, "no abort reasons found — the pattern drifted" + for file in (_strings(), _en_strings()): + assert raised <= set(file["config"]["error"]) + assert aborted <= set(file["config"]["abort"]) + + async def test_every_rendered_setup_field_has_a_label(self, tmp_path, monkeypatch): + module = _form_module() + _patch_protocol(monkeypatch, module) + flow, _hass = _new_flow(module, tmp_path) + + rendered = { + "user": vol_mirror.declared_fields( + (await flow.async_step_user(None))["data_schema"] + ), + "adb_pair": vol_mirror.declared_fields( + (await flow.async_step_adb_pair(None))["data_schema"] + ), + "protocol_pair": vol_mirror.declared_fields( + (await flow.async_step_protocol_pair(None))["data_schema"] + ), + } + + for file in (_strings(), _en_strings()): + steps = file["config"]["step"] + for step, fields in rendered.items(): + assert steps[step]["title"], step + assert steps[step]["description"], step + for field in fields: + assert field in steps[step]["data"], f"{step}.{field} has no label" + + async def test_every_rendered_options_field_has_a_label(self, tmp_path): + module = _form_module() + flow = module.NvidiaShieldOptionsFlow() + flow.config_entry = SimpleNamespace(options={}, entry_id="e1", title="Shield") + flow.hass = FakeHass(str(tmp_path)) + + form = await flow.async_step_init(None) + fields = vol_mirror.declared_fields(form["data_schema"]) + + for file in (_strings(), _en_strings()): + step = file["options"]["step"]["init"] + assert step["description"] + for field in fields: + assert field in step["data"], f"options.{field} has no label" + + +class TestStringsTranslationsParity: + """Both copies of the UI copy must expose the same key tree.""" + + def test_named_sections_exist_in_both_files(self): + for file in (_strings(), _en_strings()): + for section in ("config", "options", "entity", "services"): + assert section in file, section + + def test_key_trees_match(self): + assert _key_tree(_strings()) == _key_tree(_en_strings()) + + +class TestPassiveSetupNeverDemandsReadiness: + """The skip paths rely on setup accepting an unpaired entry.""" + + def test_async_setup_entry_raises_nothing(self): + """Setup accepts an unpaired entry — that is what the skip path needs.""" + source = (BASE / "__init__.py").read_text(encoding="utf-8") + body = source.split("async def async_setup_entry", 1)[1] + body = body.split("\nasync def ", 1)[0].split("\ndef ", 1)[0] + + assert "ConfigEntryNotReady" not in body # never demands readiness + assert "raise " not in body