"""t56: the config flow against a REAL Home Assistant install. The clearing semantics the live bug turned on ("a configured script cannot be removed") cannot be established from hand-written stubs: what matters is what Home Assistant itself does when a selector is cleared and how it serialises the form it hands the browser. So this file drives the REAL ``homeassistant.data_entry_flow.FlowManager`` with the REAL integration, real ``probatio`` validation and the real frontend serialiser (``probatio.to_field_list`` + ``cv.custom_serializer``) — the exact call the HTTP flow view makes. Skips cleanly when homeassistant is not installed (every other test file runs against stubs). Run it with: uvx --with homeassistant --with adb-shell pytest tests/test_config_flow_real_ha.py -q What this file CANNOT prove: there is no browser here, so nothing about how a form LOOKS. What it proves is the wire contract — the field list the frontend receives and the exact ``user_input`` a submission produces. """ from __future__ import annotations import pytest import pytest_asyncio pytest.importorskip( "homeassistant", reason="the real FlowManager is the point of this file", ) probatio = pytest.importorskip("probatio") import importlib.util # noqa: E402 import sys # noqa: E402 from pathlib import Path # noqa: E402 from homeassistant.config_entries import ( # noqa: E402 SOURCE_USER, ConfigEntries, ConfigEntry, OptionsFlowManager, ) from homeassistant.core import HomeAssistant # noqa: E402 from homeassistant.data_entry_flow import FlowManager # noqa: E402 from homeassistant.helpers import config_validation as cv # noqa: E402 from homeassistant.helpers import device_registry as device_registry # noqa: E402 from homeassistant.helpers import entity_registry as er # noqa: E402 from homeassistant.helpers.entity_registry import ( # noqa: E402 EntityRegistryItems, EntityRegistrySettings, ) # Load the integration by PATH under its real package name: the repo is not an # installed distribution, and the dotted import would need custom_components/ on # sys.path plus every one of its siblings present. _ROOT = Path(__file__).resolve().parent.parent if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) cf = importlib.import_module("custom_components.nvidia_shield.config_flow") SCRIPTS = {"turn_on": "script.shield_wake", "volume_set": "script.shield_volume"} OPTIONS = { "app_name_map": {"ar.tvplayer.tv": "TiviMate"}, "command_scripts": dict(SCRIPTS), "host": "10.0.0.9", "adb_enabled": True, } SCRIPT_COMMANDS = ( "turn_on", "turn_off", "volume_up", "volume_down", "volume_set", "volume_mute", "select_source", ) SCRIPT_FIELD_PREFIX = "script_" def field_list(schema): """EXACTLY what the HTTP flow view serialises for the browser.""" if schema is None: return [] return probatio.to_field_list(schema, custom_serializer=cv.custom_serializer) def field_names(schema): return [field["name"] for field in field_list(schema)] def suggested(schema): return { field["name"]: (field.get("description") or {}).get("suggested_value") for field in field_list(schema) } #: Entities the probe refers to. The real EntitySelector validates an id #: against the entity registry, so these have to exist for a set (as opposed to #: a clear) to be a legal submission. ENTITIES = ( "script.shield_wake", "script.shield_volume", "script.kept", "script.wake", "media_player.jellyfin", "switch.soundbar", "switch.plug", ) @pytest_asyncio.fixture(name="ha") async def ha_fixture(tmp_path): """A real HomeAssistant with just enough machinery to submit real forms. HomeAssistant binds the running loop in __init__, so this has to be an async fixture — which is also why the tests below are coroutines. """ hass = HomeAssistant(str(tmp_path)) # the real manager, exactly as core builds it — minus the reload SCHEDULE: # core would immediately set the domain up, which needs the integration # loader and the whole component bootstrap this probe deliberately skips. hass.config_entries = ConfigEntries(hass, {}) hass.config_entries.async_schedule_reload = lambda entry_id: None # the real registry, populated with the entities the probe uses: a real # EntitySelector refuses an id the registry does not know device_registry.async_setup(hass) await device_registry.async_load(hass) # EntityRegistry's own loader reaches deep into core (device migration, # restore, store), so the two containers it fills are set up directly — # same objects, no core bootstrap. registry = er.async_get(hass) registry.entities = EntityRegistryItems(hass) registry.deleted_entities = {} registry.settings = EntityRegistrySettings() registry._entities_data = registry.entities.data hass.data[er.DATA_REGISTRY] = registry for entity_id in ENTITIES: domain, object_id = entity_id.split(".", 1) registry.async_get_or_create( domain, "test", entity_id, suggested_object_id=object_id ) assert registry.async_get("script.kept") is not None return hass def make_entry(hass, options=None): entry = ConfigEntry( version=1, minor_version=1, domain="nvidia_shield", title=f"Shield {len(hass.config_entries._entries) + 1}", data={}, options=dict(options if options is not None else OPTIONS), source=SOURCE_USER, unique_id=None, subentries_data=(), discovery_keys={}, ) # the probe "saves" by writing back what create_entry returned, exactly as # core's async_update_entry would; ConfigEntry exposes options through a # read-only property, so the private store is written instead object.__setattr__(entry, "_options", dict(entry.options)) hass.config_entries._entries[entry.entry_id] = entry return entry class SetupManager(FlowManager): """A real FlowManager that builds the integration's real config flow. Overrides only the handler construction, and does there what core does at config_entries.py:1959 — the flow's ``init_step`` is the SOURCE, which is why a real config flow starts at ``async_step_user`` and not at ``init``. """ async def async_create_flow(self, handler_key, *, context=None, data=None): flow = cf.NvidiaShieldConfigFlow() flow.init_step = (context or {}).get("source", SOURCE_USER) return flow async def async_finish_flow(self, flow, result): return result def setup_manager(hass): return SetupManager(hass) async def open_setup(hass): """Start a real user-initiated setup flow and return (manager, result).""" manager = setup_manager(hass) result = await manager.async_init( "nvidia_shield", context={"source": SOURCE_USER} ) return manager, result #: The first setup form, as the real frontend submits it: EVERY rendered field. def first_form(**overrides): payload = { "name": "Shield", "jellyfin_entity": "", "receiver_entity": "", "receiver_input": "", "power_entity": "", "adb_enabled": False, "adb_store_app_info": True, "host": "10.0.0.9", "adb_port": 5555, "protocol_enabled": False, } payload.update(overrides) return payload class OptionsManager(OptionsFlowManager): """OptionsFlowManager that builds the integration's real options flow. Overrides ONLY the handler lookup: real HA resolves it through ``async_get_options_flow`` from the registered integration, which needs a fully loaded core. Everything else — menu dispatch, schema validation, ``next_step_id`` routing — is the real implementation. """ async def async_create_flow(self, handler_key, *, context=None, data=None): return cf.NvidiaShieldOptionsFlow() def options_manager(hass): return OptionsManager(hass) async def open_options(hass, entry): manager = options_manager(hass) result = await manager.async_init( entry.entry_id, context={"source": SOURCE_USER, "entry_id": entry.entry_id}, ) return manager, result def full_script_page(value=""): """A real submission carries EVERY rendered field — this is that page.""" return {f"{SCRIPT_FIELD_PREFIX}{c}": value for c in SCRIPT_COMMANDS} async def open_scripts(hass, entry): """Open the Scripts sub-screen through the real menu.""" manager, menu = await open_options(hass, entry) scripts = await manager.async_configure(menu["flow_id"], {"next_step_id": "scripts"}) assert scripts["type"] == "form", scripts return manager, scripts async def open_apps_menu(hass, entry): """Open the Apps editor (t64: a MENU of row actions) through the real menu.""" manager, menu = await open_options(hass, entry) apps = await manager.async_configure(menu["flow_id"], {"next_step_id": "apps"}) assert apps["type"] == "menu", apps return manager, apps async def open_apps_text(hass, entry): """Open the Apps editor's TEXT step — the whole-list multiline field (t64).""" manager, apps = await open_apps_menu(hass, entry) text = await manager.async_configure( apps["flow_id"], {"next_step_id": "apps_text"} ) assert text["type"] == "form", text return manager, text class TestRealMenuFlow: """The options page is a real menu with real sub-screens.""" async def test_the_page_is_a_menu_and_every_row_opens(self, ha): entry = make_entry(ha) manager, menu = await open_options(ha, entry) assert menu["type"] == "menu" assert menu["step_id"] == "init" assert set(menu["menu_options"]) == { "connection", "receiver", "apps", "discover", "scripts", "finish", } # HA renders the rows from the flow result, and validates the choice # against the menu's own schema assert field_names(menu["data_schema"]) == ["next_step_id"] # each row is opened on its own flow, because a real flow keeps ONE # step at a time — reopening from a fresh menu is what the browser does for row in menu["menu_options"]: if row in ("finish", "discover"): # actions: `finish` commits and `discover` sweeps, neither renders continue _manager, fresh = await open_options(ha, entry) opened = await _manager.async_configure( fresh["flow_id"], {"next_step_id": row} ) # a sub-screen is a FORM — except Apps, which is a MENU of row # actions: HA has no table selector, so a table is edited one row per # step (t64), and every row it offers has to be dispatchable too expected = "menu" if row == "apps" else "form" assert opened["type"] == expected, (row, opened) # HA dispatches a submitted form by its OWN step_id, so the rendered # step_id must be the menu row id (the method name) assert opened["step_id"] == row async def test_an_unknown_menu_row_is_rejected_by_the_real_schema(self, ha): entry = make_entry(ha) manager, menu = await open_options(ha, entry) from homeassistant.data_entry_flow import InvalidData # real HA validates the choice against the menu schema (probatio.In on # the offered ids) before the flow is even dispatched with pytest.raises(InvalidData): await manager.async_configure( menu["flow_id"], {"next_step_id": "not_a_step"} ) class TestRealAppsField: """[object Object]: the field list must carry TEXT, never a dict.""" async def test_the_apps_suggestion_is_the_serialized_text(self, ha): entry = make_entry(ha) manager, apps = await open_apps_text(ha, entry) suggested_values = suggested(apps["data_schema"]) assert suggested_values["app_name_map"] == "ar.tvplayer.tv=TiviMate" assert isinstance(suggested_values["app_name_map"], str) # and nothing in the whole field list carries a dict for a text field for field in field_list(apps["data_schema"]): value = (field.get("description") or {}).get("suggested_value") if field.get("type") == "string": assert not isinstance(value, (dict, list)), field["name"] async def test_the_empty_apps_state_renders_without_a_suggestion(self, ha): entry = make_entry(ha, options={"host": "10.0.0.9"}) manager, apps = await open_apps_text(ha, entry) fields = field_list(apps["data_schema"]) assert [f["name"] for f in fields] == ["app_name_map"] assert (fields[0].get("description") or {}).get("suggested_value") in (None, "") async def test_a_table_row_round_trips_through_the_real_flow(self, ha): entry = make_entry(ha, options={"host": "10.0.0.9"}) manager, apps = await open_apps_text(ha, entry) saved = await manager.async_configure( apps["flow_id"], {"app_name_map": "com.plex.hdx=Plex|http://icons.local/plex.png"}, ) assert saved["type"] == "menu" # back at the APPS editor, nothing committed assert saved["step_id"] == "apps" back = await manager.async_configure( saved["flow_id"], {"next_step_id": "menu_again"} ) assert back["type"] == "menu" and back["step_id"] == "init" created = await manager.async_configure( back["flow_id"], {"next_step_id": "finish"} ) assert created["type"] == "create_entry" assert created["data"]["app_name_map"] == {"com.plex.hdx": "Plex"} assert created["data"]["app_icon_map"] == { "com.plex.hdx": "http://icons.local/plex.png" } class TestRealScriptClearing: """The live bug: what the REAL flow stores when a selector is cleared.""" async def test_set_then_cleared_persists_through_save_and_reopen(self, ha): """t56 unit 0 / t57 F4: SET, CLEAR, SAVE, REOPEN — the live symptom. The whole page goes through the real flow manager: first with a script SELECTED (an entity id, which the seeded registry resolves), then with that one selector cleared while the other keeps its value. Both halves are driven here — the earlier version of this file claimed entity ids could not be driven, which was simply false (the receiver page in this file has always submitted ``switch.soundbar``). """ entry = make_entry(ha) manager, scripts = await open_scripts(ha, entry) # 1. SET: two commands selected through the real flow, from real ids page = full_script_page("") page["script_turn_on"] = "script.kept" page["script_volume_set"] = SCRIPTS["volume_set"] after_set = await manager.async_configure(scripts["flow_id"], page) assert after_set["type"] == "menu", after_set created = await manager.async_configure( after_set["flow_id"], {"next_step_id": "finish"} ) assert created["type"] == "create_entry" assert created["data"]["command_scripts"] == { "turn_on": "script.kept", "volume_set": SCRIPTS["volume_set"], } # 2. CLEAR: the same page with ONE selector emptied (a cleared selector # arrives as ''), the other left set saved = dict(entry.options) saved.update(created["data"]) object.__setattr__(entry, "_options", saved) manager2, scripts2 = await open_scripts(ha, entry) assert suggested(scripts2["data_schema"])["script_turn_on"] == "script.kept" cleared = full_script_page("") cleared["script_turn_on"] = "" cleared["script_volume_set"] = SCRIPTS["volume_set"] after_clear = await manager2.async_configure(scripts2["flow_id"], cleared) assert after_clear["type"] == "menu", after_clear created2 = await manager2.async_configure( after_clear["flow_id"], {"next_step_id": "finish"} ) assert created2["data"]["command_scripts"] == { "volume_set": SCRIPTS["volume_set"], } # 3. PERSIST: re-open over the SAVED options — the cleared command must # not come back, and no flat script_* key was ever stored saved2 = dict(entry.options) saved2.update(created2["data"]) object.__setattr__(entry, "_options", saved2) assert not [k for k in entry.options if k.startswith(SCRIPT_FIELD_PREFIX)] manager3, scripts3 = await open_scripts(ha, entry) values = suggested(scripts3["data_schema"]) assert values["script_volume_set"] == SCRIPTS["volume_set"] assert values.get("script_turn_on") in (None, "") @pytest.mark.parametrize("shape", ["empty_string", "null"]) async def test_a_cleared_selector_is_accepted_by_the_real_schema(self, ha, shape): """Real HA rejects '' / None for a plain EntitySelector; ours tolerates it.""" entry = make_entry(ha) manager, scripts = await open_scripts(ha, entry) page = full_script_page("") page["script_turn_on"] = "" if shape == "empty_string" else None saved = await manager.async_configure(scripts["flow_id"], page) assert saved["type"] == "menu", saved created = await manager.async_configure( saved["flow_id"], {"next_step_id": "finish"} ) assert created["data"]["command_scripts"] == {} async def test_a_never_set_field_stays_out_of_the_dict(self, ha): entry = make_entry(ha, options={"host": "10.0.0.9", "adb_enabled": True}) manager, scripts = await open_scripts(ha, entry) # a real submission carries every rendered field, all empty here saved = await manager.async_configure(scripts["flow_id"], full_script_page("")) created = await manager.async_configure( saved["flow_id"], {"next_step_id": "finish"} ) assert created["data"]["command_scripts"] == {} class TestRealSelectorValidation: """The selector distinction, pinned on the selector classes themselves. ``TestRealScriptClearing`` drives a cleared page end to end; this class pins the underlying difference in isolation, so a regression says WHICH rule broke: the built-in EntitySelector rejects a cleared field, ours accepts it and still validates a real id. """ def test_real_entity_selector_rejects_a_cleared_field(self): from homeassistant.helpers.selector import ( EntitySelector, EntitySelectorConfig, ) strict = EntitySelector(EntitySelectorConfig(domain="script")) for cleared in ("", None): with pytest.raises(probatio.Invalid): strict(cleared) def test_the_clearing_selector_accepts_cleared_and_real_values(self): from custom_components.nvidia_shield.config_flow import _build_selectors selector = _build_selectors()["script"] assert selector("") == "" assert selector(None) == "" assert selector(" ") == "" assert selector("script.whatever") == "script.whatever" class TestRealNoLostEdits: """Edits made on one sub-screen survive a submission on another.""" async def test_connection_then_scripts_then_apps_all_survive(self, ha): entry = make_entry(ha, options={}) manager, menu = await open_options(ha, entry) connection = await manager.async_configure( menu["flow_id"], {"next_step_id": "connection"} ) after_connection = await manager.async_configure( connection["flow_id"], { "host": "10.0.0.44", "adb_port": 5555, "adb_enabled": True, "protocol_enabled": False, "jellyfin_entity": "media_player.jellyfin", }, ) assert after_connection["type"] == "menu" scripts = await manager.async_configure( after_connection["flow_id"], {"next_step_id": "scripts"} ) page = full_script_page("") page["script_turn_on"] = "script.wake" after_scripts = await manager.async_configure(scripts["flow_id"], page) assert after_scripts["type"] == "menu" apps = await manager.async_configure( after_scripts["flow_id"], {"next_step_id": "apps"} ) assert apps["type"] == "menu" and apps["step_id"] == "apps" text = await manager.async_configure( apps["flow_id"], {"next_step_id": "apps_text"} ) after_apps = await manager.async_configure( text["flow_id"], {"app_name_map": "com.plex.hdx=Plex"} ) assert after_apps["type"] == "menu" and after_apps["step_id"] == "apps" back = await manager.async_configure( after_apps["flow_id"], {"next_step_id": "menu_again"} ) assert back["type"] == "menu" and back["step_id"] == "init" created = await manager.async_configure( back["flow_id"], {"next_step_id": "finish"} ) assert created["type"] == "create_entry" data = created["data"] # every earlier sub-screen's edit is still there assert data["host"] == "10.0.0.44" assert data["jellyfin_entity"] == "media_player.jellyfin" assert data["command_scripts"] == {"turn_on": "script.wake"} assert data["app_name_map"] == {"com.plex.hdx": "Plex"} class TestRealReceiverSelector: """The Receiver field accepts a switch, not only a media_player.""" async def test_the_selector_offers_switch_domains(self, ha): entry = make_entry(ha, options={}) manager, menu = await open_options(ha, entry) receiver = await manager.async_configure( menu["flow_id"], {"next_step_id": "receiver"} ) fields = {f["name"]: f for f in field_list(receiver["data_schema"])} selector = fields["receiver_entity"]["selector"]["entity"] assert "switch" in selector["domain"] assert "media_player" in selector["domain"] async def test_a_switch_is_accepted_and_stored(self, ha): entry = make_entry(ha, options={}) manager, menu = await open_options(ha, entry) receiver = await manager.async_configure( menu["flow_id"], {"next_step_id": "receiver"} ) saved = await manager.async_configure( receiver["flow_id"], {"receiver_entity": "switch.soundbar", "receiver_input": "HDMI 2"}, ) created = await manager.async_configure( saved["flow_id"], {"next_step_id": "finish"} ) assert created["data"]["receiver_entity"] == "switch.soundbar" assert created["data"]["receiver_input"] == "HDMI 2" class TestRealSetupMenu: """t57 UNIT 0a / F2: the SETUP flow under a REAL FlowManager. The stub suite drove `async_step_menu` directly, which is exactly how a broken menu stays green: it never asked Home Assistant to dispatch a row. These tests do, for every row, and they are the red proof for the TypeError the first form used to hit on the way to the menu. """ @staticmethod def _no_connect(result=True): async def _connect(*args, **kwargs): return result return staticmethod(_connect) async def test_the_first_form_leads_into_pairing_then_the_menu(self, ha, monkeypatch): """The whole ordinary path: first form → ADB pairing → the menu.""" monkeypatch.setattr( cf.ShieldConnectionManager, "test_connect", self._no_connect(True) ) manager, form = await open_setup(ha) assert form["type"] == "form" assert form["step_id"] == "user" pairing = await manager.async_configure( form["flow_id"], first_form(adb_enabled=True, protocol_enabled=False) ) assert pairing["type"] == "form" assert pairing["step_id"] == "adb_pair" result = await manager.async_configure(pairing["flow_id"], {"adb_port": 5555}) assert result["type"] == "menu", result # ADB is paired, the protocol is off: only the two sub-screens + finish assert list(result["menu_options"]) == ["receiver", "connection", "finish"] async def test_the_adb_skip_reaches_the_menu(self, ha, monkeypatch): """The path that used to die: skip ADB pairing, then render the menu.""" monkeypatch.setattr( cf.ShieldConnectionManager, "test_connect", self._no_connect(False) ) manager, form = await open_setup(ha) pairing = await manager.async_configure( form["flow_id"], first_form(adb_enabled=True, protocol_enabled=False) ) assert pairing["step_id"] == "adb_pair" result = await manager.async_configure( pairing["flow_id"], {"skip_pairing": True} ) assert result["type"] == "menu", result assert "adb_pair" in result["menu_options"] # not paired: still offered async def test_adb_pairing_success_stops_offering_that_row(self, ha, monkeypatch): monkeypatch.setattr( cf.ShieldConnectionManager, "test_connect", self._no_connect(True) ) manager, form = await open_setup(ha) pairing = await manager.async_configure( form["flow_id"], first_form(adb_enabled=True, protocol_enabled=False) ) paired = await manager.async_configure(pairing["flow_id"], {"adb_port": 5555}) assert paired["type"] == "menu" assert "adb_pair" not in paired["menu_options"] # F2: the guard async def test_every_setup_menu_row_dispatches_and_renders(self, ha, monkeypatch): """Every row HA offers must have a handler AND render a form whose step_id is that same row, or the page dead-ends on submit.""" monkeypatch.setattr( cf.ShieldConnectionManager, "test_connect", self._no_connect(False) ) # the protocol row builds the androidtvremote2 client; this test is about # DISPATCH, and the pairing internals have their own stub coverage. The # UNREACHABLE branch is a real code path needing no library, so the row # renders its real form. async def _unreachable(self): return None, "protocol_unreachable" monkeypatch.setattr( cf.NvidiaShieldConfigFlow, "_async_begin_protocol_pairing", _unreachable ) manager, form = await open_setup(ha) menu_result = await manager.async_configure( form["flow_id"], first_form(adb_enabled=True, protocol_enabled=True), ) assert menu_result["type"] == "form" # the ADB pairing form comes first # hand the ADB form a skip so the MENU renders with every row menu_result = await manager.async_configure( menu_result["flow_id"], {"skip_pairing": True} ) assert menu_result["type"] == "menu", menu_result rows = list(menu_result["menu_options"]) assert rows == ["receiver", "connection", "adb_pair", "protocol_pair", "finish"] for row in rows: # each row on its own flow (a flow is at one step at a time) AND on # the manager that owns it: a flow lives in its manager's progress row_manager, fresh = await open_setup(ha) pairing = await row_manager.async_configure( fresh["flow_id"], first_form(adb_enabled=True, protocol_enabled=True) ) menu = await row_manager.async_configure( pairing["flow_id"], {"skip_pairing": True} ) opened = await row_manager.async_configure( menu["flow_id"], {"next_step_id": row} ) assert opened["type"] in ("form", "menu", "create_entry"), (row, opened) if opened["type"] == "form": assert opened["step_id"] == row, (row, opened["step_id"]) async def test_a_setup_sub_screen_submission_is_accepted(self, ha, monkeypatch): """A rendered sub-screen must be re-submittable (row id == step_id).""" monkeypatch.setattr( cf.ShieldConnectionManager, "test_connect", self._no_connect(True) ) manager, form = await open_setup(ha) pairing = await manager.async_configure( form["flow_id"], first_form(adb_enabled=True, protocol_enabled=False) ) menu = await manager.async_configure(pairing["flow_id"], {"adb_port": 5555}) assert menu["type"] == "menu" opened = await manager.async_configure( menu["flow_id"], {"next_step_id": "receiver"} ) assert opened["type"] == "form" assert opened["step_id"] == "receiver" back = await manager.async_configure( opened["flow_id"], {"receiver_entity": "switch.soundbar"} ) assert back["type"] == "menu", back created = await manager.async_configure( back["flow_id"], {"next_step_id": "finish"} ) assert created["type"] == "create_entry" assert created["options"]["receiver_entity"] == "switch.soundbar" async def test_the_full_menu_walk_merges_and_creates_one_entry(self, ha, monkeypatch): """connection + receiver + finish → ONE entry carrying both edits.""" monkeypatch.setattr( cf.ShieldConnectionManager, "test_connect", self._no_connect(True) ) manager, form = await open_setup(ha) pairing = await manager.async_configure( form["flow_id"], first_form(adb_enabled=True, protocol_enabled=False) ) menu = await manager.async_configure(pairing["flow_id"], {"adb_port": 5555}) connection = await manager.async_configure( menu["flow_id"], {"next_step_id": "connection"} ) assert connection["step_id"] == "connection" menu2 = await manager.async_configure( connection["flow_id"], {"host": "10.0.0.55", "adb_port": 5555, "adb_enabled": True, "protocol_enabled": False} ) assert menu2["type"] == "menu" receiver = await manager.async_configure( menu2["flow_id"], {"next_step_id": "receiver"} ) menu3 = await manager.async_configure( receiver["flow_id"], {"receiver_entity": "switch.soundbar"} ) created = await manager.async_configure( menu3["flow_id"], {"next_step_id": "finish"} ) assert created["type"] == "create_entry" assert created["options"]["host"] == "10.0.0.55" assert created["options"]["receiver_entity"] == "switch.soundbar" @pytest.mark.parametrize( ("adb_enabled", "protocol_enabled", "connect", "pairs_protocol"), [ (True, False, False, False), # ADB skipped (True, False, True, False), # ADB paired, protocol off (True, True, False, False), # ADB skipped, protocol skipped (True, True, True, False), # ADB paired, protocol skipped (False, True, False, False), # protocol only ], ids=["adb-skip", "adb-paired", "adb-skip+protocol-skip", "pair-adb", "protocol-only"], ) async def test_every_ordinary_path_reaches_create_entry( self, ha, monkeypatch, adb_enabled, protocol_enabled, connect, pairs_protocol ): """t57 UNIT 0a: the path map. Under the pre-fix tree FOUR of these five died with `TypeError: async_show_menu() got an unexpected keyword argument 'errors'` the moment the ADB pairing form was submitted or skipped — so a new entry could not be created on any path that went near the menu, and the sub-screens were unreachable. Each path is driven here to `create_entry`. """ monkeypatch.setattr( cf.ShieldConnectionManager, "test_connect", self._no_connect(connect) ) async def _unreachable(self): return None, "protocol_unreachable" monkeypatch.setattr( cf.NvidiaShieldConfigFlow, "_async_begin_protocol_pairing", _unreachable ) manager, form = await open_setup(ha) result = await manager.async_configure( form["flow_id"], first_form(adb_enabled=adb_enabled, protocol_enabled=protocol_enabled), ) assert result["type"] == "form", result for _ in range(12): if result["type"] == "create_entry": break step = result.get("step_id") if step == "adb_pair": result = await manager.async_configure( result["flow_id"], {"skip_pairing": True} ) elif step == "protocol_pair": result = await manager.async_configure( result["flow_id"], {"pairing_code": "1234", "host": "10.0.0.9", "skip_pairing": True}, ) elif result["type"] == "menu": # the menu must always offer a way out assert "finish" in result["menu_options"], result["menu_options"] result = await manager.async_configure( result["flow_id"], {"next_step_id": "finish"} ) else: # pragma: no cover - the assertion below reports it break assert result["type"] == "create_entry", result async def test_the_menu_description_reports_what_is_unpaired(self, ha, monkeypatch): monkeypatch.setattr( cf.ShieldConnectionManager, "test_connect", self._no_connect(False) ) manager, form = await open_setup(ha) pairing = await manager.async_configure( form["flow_id"], first_form(adb_enabled=True, protocol_enabled=True) ) menu = await manager.async_configure(pairing["flow_id"], {"skip_pairing": True}) assert menu["type"] == "menu" note = (menu.get("description_placeholders") or {}).get("pairing", "") assert "ADB" in note and "remote protocol" in note # ------------------------------------------- t64: the apps editor (real HA) # # Both live reports are about this screen: the map arrived as ONE LONG LINE with # rows that could not be added, and "Discover installed apps seems to load then do # nothing". The fixes are pinned here, over the REAL FlowManager, because that is # where the field list the browser renders and the exact ``user_input`` a # submission produces are decided. What is NOT provable here: how the field LOOKS. # There is no browser, so "multiline" is proven through the serialised selector # (``selector.text.multiline``) — the value the frontend actually reads. class TestRealAppsMultilineField: """t64: the apps field is a MULTILINE text field — the live bug's cause.""" async def test_the_rendered_field_carries_the_multiline_text_selector(self, ha): entry = make_entry(ha, options={"app_name_map": {"ar.tvplayer.tv": "TiviMate"}}) manager, apps = await open_apps_text(ha, entry) fields = {f["name"]: f for f in field_list(apps["data_schema"])} assert list(fields) == ["app_name_map"] # A bare `str` field with no selector renders as a SINGLE-LINE input (HA # docs), which is why the live page showed the map flattened and refused # new lines. The field must carry a real selector, and it is the text one; # a selector-less field would have no ``selector`` key at all. assert fields["app_name_map"]["selector"] == { "text": {"multiline": True, "multiple": False} } async def test_a_saved_page_keeps_one_row_per_line_and_reopens_unchanged(self, ha): entry = make_entry(ha, options={"host": "10.0.0.9"}) manager, apps = await open_apps_text(ha, entry) page = "com.a=Alpha\ncom.b=Beta|http://icons.local/b.png\ncom.c=Gamma" saved = await manager.async_configure(apps["flow_id"], {"app_name_map": page}) back = await manager.async_configure( saved["flow_id"], {"next_step_id": "menu_again"} ) created = await manager.async_configure( back["flow_id"], {"next_step_id": "finish"} ) assert created["data"]["app_name_map"] == { "com.a": "Alpha", "com.b": "Beta", "com.c": "Gamma", } assert created["data"]["app_icon_map"] == {"com.b": "http://icons.local/b.png"} # reopen over the SAVED options: the same three lines come back, in order reopened = dict(entry.options) reopened.update(created["data"]) object.__setattr__(entry, "_options", reopened) _manager, apps2 = await open_apps_text(ha, entry) assert suggested(apps2["data_schema"])["app_name_map"] == page class TestRealAppsRowEditor: """t64: add / edit / remove ONE row — HA has no table selector.""" async def test_the_editor_offers_only_the_rows_that_can_do_something(self, ha): entry = make_entry(ha, options={}) _manager, apps = await open_apps_menu(ha, entry) # nothing stored AND nothing detected: no edit/remove row could do # anything, so none is offered (t64) assert list(apps["menu_options"]) == ["apps_text", "apps_add", "menu_again"] assert "no apps known yet" in apps["description_placeholders"]["apps"] entry = make_entry( ha, options={"app_name_map": {"com.a": "Alpha", "com.b": "Beta"}} ) _manager, apps = await open_apps_menu(ha, entry) assert list(apps["menu_options"]) == [ "apps_text", "apps_add", "apps_edit", "apps_remove", "menu_again", ] assert apps["description_placeholders"]["apps"] == ( "2 apps known — Alpha (com.a), Beta (com.b)" ) async def test_the_dropdown_carries_the_detected_apps_and_names_them(self, ha): """t64: the picked-from list is the detected list, named where known. The user's own Shield exposes no device label at all, so the unnamed case is the common one: it must read as unnamed (not as its own id twice), it must still be pickable, and saving a name for it must store the row. """ entry = make_entry(ha, options={"app_name_map": {"com.a": "Alpha"}}) entry.runtime_data = TestRealDiscoverFeedback._Manager( [ {"package": "org.xbmc.kodi", "name": "Kodi", "icon": "mdi:kodi"}, {"package": "com.unnamed.tv", "name": "com.unnamed.tv", "icon": ""}, {"package": "com.a", "name": "Wrong name", "icon": "mdi:x"}, ] ) manager, apps = await open_apps_menu(ha, entry) note = apps["description_placeholders"]["apps"] assert note.startswith("3 apps known — ") assert "Kodi (org.xbmc.kodi) [detected]" in note assert "com.unnamed.tv (no name yet) [detected]" in note assert "Alpha (com.a)" in note # a saved row keeps the user's name choose = await manager.async_configure(apps["flow_id"], {"next_step_id": "apps_edit"}) options = field_list(choose["data_schema"])[0]["selector"]["select"]["options"] assert options == [ {"value": "com.a", "label": "Alpha (com.a)"}, {"value": "org.xbmc.kodi", "label": "Kodi (org.xbmc.kodi)"}, {"value": "com.unnamed.tv", "label": "com.unnamed.tv (no name yet)"}, ] form = await manager.async_configure( choose["flow_id"], {"app_package": "com.unnamed.tv"} ) assert form["step_id"] == "apps_edit_form" assert "com.unnamed.tv (no name yet)" in form["description_placeholders"]["app"] assert suggested(form["data_schema"])["app_name"] == "" # nothing invented assert suggested(form["data_schema"])["app_icon"] == "" # never the mdi: icon edited = await manager.async_configure( form["flow_id"], {"app_name": "Unnamed TV", "app_icon": "http://icons.local/u.png"}, ) back = await manager.async_configure( edited["flow_id"], {"next_step_id": "menu_again"} ) created = await manager.async_configure( back["flow_id"], {"next_step_id": "finish"} ) assert created["data"]["app_name_map"] == { "com.a": "Alpha", "com.unnamed.tv": "Unnamed TV", } assert created["data"]["app_icon_map"] == {"com.unnamed.tv": "http://icons.local/u.png"} async def test_add_persists_one_row_and_leaves_the_others_alone(self, ha): entry = make_entry( ha, options={"app_name_map": {"com.a": "Alpha"}, "host": "10.0.0.9"} ) manager, apps = await open_apps_menu(ha, entry) add = await manager.async_configure(apps["flow_id"], {"next_step_id": "apps_add"}) assert add["type"] == "form" assert add["step_id"] == "apps_add" assert field_names(add["data_schema"]) == ["app_package", "app_name", "app_icon"] edited = await manager.async_configure( add["flow_id"], {"app_package": "com.b", "app_name": "Beta", "app_icon": ""}, ) assert edited["type"] == "menu" assert edited["step_id"] == "apps" back = await manager.async_configure( edited["flow_id"], {"next_step_id": "menu_again"} ) created = await manager.async_configure( back["flow_id"], {"next_step_id": "finish"} ) assert created["data"]["app_name_map"] == {"com.a": "Alpha", "com.b": "Beta"} assert created["data"]["host"] == "10.0.0.9" # nothing else moved async def test_an_invalid_row_is_refused_with_a_message(self, ha): entry = make_entry(ha, options={}) manager, apps = await open_apps_menu(ha, entry) add = await manager.async_configure(apps["flow_id"], {"next_step_id": "apps_add"}) refused = await manager.async_configure( add["flow_id"], {"app_package": "has spaces", "app_name": "Nope", "app_icon": ""}, ) assert refused["type"] == "form" assert refused["step_id"] == "apps_add" assert refused["errors"] == {"base": "invalid_app_row"} async def test_edit_lists_the_stored_rows_and_prefills_the_chosen_one(self, ha): entry = make_entry( ha, options={ "app_name_map": {"com.a": "Alpha", "com.b": "Beta"}, "app_icon_map": {"com.b": "http://icons.local/b.png"}, }, ) manager, apps = await open_apps_menu(ha, entry) choose = await manager.async_configure( apps["flow_id"], {"next_step_id": "apps_edit"} ) assert choose["type"] == "form" assert choose["step_id"] == "apps_edit" options = field_list(choose["data_schema"])[0]["selector"]["select"]["options"] assert options == [ {"value": "com.a", "label": "Alpha (com.a)"}, {"value": "com.b", "label": "Beta (com.b)"}, ] form = await manager.async_configure(choose["flow_id"], {"app_package": "com.b"}) assert form["type"] == "form" assert form["step_id"] == "apps_edit_form" assert "Beta (com.b)" in form["description_placeholders"]["app"] # PREFILLED from the stored row — name and icon both assert suggested(form["data_schema"])["app_name"] == "Beta" assert suggested(form["data_schema"])["app_icon"] == "http://icons.local/b.png" edited = await manager.async_configure( form["flow_id"], {"app_name": "Beta (living room)", "app_icon": "http://icons.local/b2.png"}, ) assert edited["type"] == "menu" and edited["step_id"] == "apps" back = await manager.async_configure( edited["flow_id"], {"next_step_id": "menu_again"} ) created = await manager.async_configure( back["flow_id"], {"next_step_id": "finish"} ) assert created["data"]["app_name_map"] == { "com.a": "Alpha", "com.b": "Beta (living room)", } assert created["data"]["app_icon_map"] == {"com.b": "http://icons.local/b2.png"} async def test_remove_confirms_first_and_a_cancel_keeps_the_row(self, ha): entry = make_entry( ha, options={"app_name_map": {"com.a": "Alpha", "com.b": "Beta"}} ) manager, apps = await open_apps_menu(ha, entry) choose = await manager.async_configure( apps["flow_id"], {"next_step_id": "apps_remove"} ) assert choose["type"] == "form" and choose["step_id"] == "apps_remove" confirm = await manager.async_configure(choose["flow_id"], {"app_package": "com.a"}) assert confirm["type"] == "menu" assert confirm["step_id"] == "apps_remove_confirm" assert set(confirm["menu_options"]) == {"apps_remove_yes", "apps_remove_no"} assert "Alpha (com.a)" in confirm["description_placeholders"]["app"] cancelled = await manager.async_configure( confirm["flow_id"], {"next_step_id": "apps_remove_no"} ) assert cancelled["type"] == "menu" and cancelled["step_id"] == "apps" # ...and the confirmed path drops exactly that row choose2 = await manager.async_configure( cancelled["flow_id"], {"next_step_id": "apps_remove"} ) confirm2 = await manager.async_configure( choose2["flow_id"], {"app_package": "com.a"} ) after = await manager.async_configure( confirm2["flow_id"], {"next_step_id": "apps_remove_yes"} ) assert after["type"] == "menu" and after["step_id"] == "apps" back = await manager.async_configure( after["flow_id"], {"next_step_id": "menu_again"} ) created = await manager.async_configure( back["flow_id"], {"next_step_id": "finish"} ) assert created["data"]["app_name_map"] == {"com.b": "Beta"} class TestRealDiscoverFeedback: """t64: the Discover row reports WHAT IT DID — never a silent no-op.""" class _Manager: """A stand-in connection manager with a canned discovered table.""" def __init__(self, rows=None, connected=True, fail=False): self.rows = list(rows or []) self.connected = connected self.fail = fail self.refresh_calls: list[bool] = [] def installed_apps(self): return [dict(row) for row in self.rows] async def async_refresh_installed_apps(self, force=False): self.refresh_calls.append(force) if self.fail: raise RuntimeError("adb went away") return self.installed_apps() async def _discover(self, ha, entry): manager, menu = await open_options(ha, entry) return manager, await manager.async_configure( menu["flow_id"], {"next_step_id": "discover"} ) async def test_an_unreachable_sweep_says_so(self, ha): entry = make_entry(ha, options={}) entry.runtime_data = self._Manager(connected=False, fail=True) _manager, after = await self._discover(ha, entry) assert after["type"] == "menu" and after["step_id"] == "init" note = after["description_placeholders"]["discovery"] assert "could not reach" in note and "unchanged" in note async def test_finding_apps_reports_the_count(self, ha): entry = make_entry(ha, options={}) manager = self._Manager( [ {"package": "com.a", "name": "Alpha", "icon": "mdi:alpha"}, {"package": "com.b", "name": "Beta", "icon": "mdi:beta"}, ] ) entry.runtime_data = manager _manager, after = await self._discover(ha, entry) note = after["description_placeholders"]["discovery"] assert note.startswith("2 apps found on the Shield") assert "updated" in note assert manager.refresh_calls == [True] # explicit force, exactly once async def test_finding_nothing_says_the_rows_are_kept(self, ha): entry = make_entry(ha, options={"app_name_map": {"com.a": "Alpha"}}) entry.runtime_data = self._Manager([]) _manager, after = await self._discover(ha, entry) note = after["description_placeholders"]["discovery"] assert "no installed apps" in note and "keeps your" in note async def test_without_a_manager_it_says_adb_is_not_set_up(self, ha): entry = make_entry(ha, options={}) _manager, after = await self._discover(ha, entry) note = after["description_placeholders"]["discovery"] assert note == "ADB is not set up for this entry — nothing could be discovered."