James Mackay
3a55519b74
t61 F1 (BLOCKER) + F2 + F3 + F5, all reproduced against a real FlowManager
before fixing. The OPTIONS flow was verified good and is untouched by this
commit; the SETUP flow was broken in three independent ways.
F1 — `_show_setup_menu` passed `errors=` to `async_show_menu`, which has no
such parameter (HA 2026.10.1: step_id / menu_options / sort /
description_placeholders). Every path that went near the menu therefore
raised `TypeError` the moment the first form was submitted or ADB pairing
was skipped. The path map, driven end to end: FOUR of the five ordinary
paths died before create_entry ("ADB skip", "ADB paired + protocol off",
"ADB skip + protocol skip", "pair ADB"); only "protocol only" survived, and
it never showed the menu — so a new entry could only be created on that one
path and the sub-screens were unreachable.
F2 — the menu's rows broke the rule this repo documents: HA dispatches a row
to `async_step_<row id>` AND dispatches a submitted form by its OWN step_id,
so a row named `adb` with a handler called `async_step_adb_pair` is an
UnknownStep, and a row named `receiver` rendering `step_id="setup_receiver"`
is a form that cannot be submitted. The rows are now their handler names
(`receiver`, `connection`, `adb_pair`, `protocol_pair`, `finish`) and the
two sub-screens render their row id. There is also a `_paired_adb` guard
now, so a completed ADB pairing stops offering that row — the rule the
protocol row already followed.
F3 — both menus keyed their row labels under a bare `menu`, which core does
not read (56 core strings files use `menu_options`), so every row fell back
to its raw id in the browser. Renamed in both strings files.
F5 — the per-command script merge is field-wise now: a `script_<command>`
key that is present and empty removes that command, one with a value sets
it, and a key the submission does not carry is untouched — the same rule as
every other option. The wholesale rebuild depended on the page always being
complete, so a partial submission silently wiped the commands it omitted.
THE SYSTEMIC HALF (F1), closed as asked: `ha_api_drift.HA_CALL_OWNERS` now
covers the flow-display APIs (`async_show_menu`, `async_show_progress`,
`async_show_progress_done`, plus the existing form/abort/entry owners), so
the drift engine resolves their REAL signatures and flags an unknown
keyword. `tests/test_ha_api_drift.py` carries a pre-fix source using
`errors=` to prove the new owner is non-vacuous. Every harness stub now
mirrors the real signature (no `errors` on a menu) and a stub-level test
pins that parameter set, so a stub cannot go back to accepting anything.
On the pre-fix tree the drift engine reports exactly this:
config_flow.py:898: unknown-kwarg: async_show_menu() has no 'errors'
parameter; real signature: (self, *, step_id=None, menu_options, sort,
description_placeholders) -> _FlowResultT
TESTS: tests/test_config_flow_real_ha.py gains a SETUP section that drives
the real flow — the first form, the ADB skip, the pairing guard, EVERY menu
row (dispatch AND re-submittable step_id), an accepted sub-screen
submission, a full connection+receiver+finish walk merging into one entry,
the menu's pairing note, and the five-path map as a parametrised test.
11 of those fail on the pre-fix tree; the drift engine failure above is the
12th. The stub suite additionally pins the menu stub's signature and the
`menu_options` key in both strings files.
Also in this commit (t57 F4): the real-HA file claimed entity-valued pages
could not be driven there, which was false (it has always submitted
`switch.soundbar`). The claim is gone and the scripts page is driven through
SET (two scripts selected, real entity ids) → CLEAR (one selector emptied)
→ SAVE → REOPEN, so the SET half is pinned in the real flow and not only
against the selector objects.
818 lines
32 KiB
Python
818 lines
32 KiB
Python
"""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
|
|
|
|
|
|
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}
|
|
)
|
|
assert opened["type"] == "form", row
|
|
# 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, menu = await open_options(ha, entry)
|
|
apps = await manager.async_configure(menu["flow_id"], {"next_step_id": "apps"})
|
|
|
|
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, menu = await open_options(ha, entry)
|
|
apps = await manager.async_configure(menu["flow_id"], {"next_step_id": "apps"})
|
|
|
|
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, menu = await open_options(ha, entry)
|
|
apps = await manager.async_configure(menu["flow_id"], {"next_step_id": "apps"})
|
|
|
|
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 menu, nothing committed yet
|
|
|
|
created = await manager.async_configure(
|
|
saved["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"}
|
|
)
|
|
after_apps = await manager.async_configure(
|
|
apps["flow_id"], {"app_name_map": "com.plex.hdx=Plex"}
|
|
)
|
|
assert after_apps["type"] == "menu"
|
|
|
|
created = await manager.async_configure(
|
|
after_apps["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
|