test(config-flow): drive the menus, pin the tables, probe real HA

The flow tests drove `async_step_init` as if it were the one form, so they
had to learn the menu: a shared helper picks the sub-screen that owns a
payload's keys out of the flow's own `menu_options` (no test re-implements
the routing), and a `save=True` flag walks to "Save and close" so the
assertions read as they did before.

New coverage for this task:

- the menu offers every sub-screen and every row has a handler named
  after it, with the rendered step_id equal to the row id (the invariant
  real HA enforces and the first draft broke);
- edits on one sub-screen are not lost by another, and a page that is
  never saved never writes the entry;
- the apps table: add / edit / remove a row, the icon column round-trips,
  the empty state renders without data, a blank field clears the map, and
  the suggested value is TEXT (the `[object Object]` layer);
- a receiver the selector offers `switch` for, stored and pre-filled;
- the setup menu's rows, their handlers and a connection+receiver merge.

tests/test_config_flow_real_ha.py is new and is the empirical half: it
drives the real `FlowManager`, real probatio validation and the real
frontend serialiser (`to_field_list` + `cv.custom_serializer`) — the call
the HTTP flow view makes — so the menu dispatch, the cleared-selector
shapes and the wire serialisation are established rather than assumed.
It skips when homeassistant is absent; CI runs it with:

    uvx --with homeassistant --with adb-shell pytest tests/test_config_flow_real_ha.py

53 of the stub tests and 11 of the real-HA ones fail on the pre-fix tree.
This commit is contained in:
James Mackay committed 2026-10-12 07:17:41 +01:00
1 parent b0adbd2dfd
commit ca3fdec218
6 files changed
+1146 -74

No files matched your search

+499 -55
View File
@@ -31,11 +31,15 @@ import pytest
import voluptuous_mirror as vol_mirror
from conftest import _SHIELD_DIR, _load_module
from nsh.const import (
CONF_ADB_COMMAND_ENABLED,
CONF_ADB_ENABLED,
CONF_ADB_PORT,
CONF_ADB_STORE_APP_INFO,
CONF_CHANNEL_DIGIT_DELAY,
CONF_COMMAND_SCRIPTS,
CONF_HOST,
CONF_JELLYFIN_ENTITY,
CONF_LOGCAT_EVENTS,
CONF_PAIRING_CODE,
CONF_POWER_ENTITY,
CONF_PROTOCOL_ENABLED,
@@ -103,6 +107,19 @@ def _install_stubs() -> None:
"description_placeholders": dict(description_placeholders or {}),
}
def async_show_menu(self, *, step_id, menu_options, sort=False,
description_placeholders=None, errors=None):
# t56: the real result shape. The flow manager reads
# `next_step_id` from the MENU result and calls
# `async_step_<choice>(None)`, so the stub reports the menu and the
# test driver makes the choice.
return {
"type": "menu",
"step_id": step_id,
"menu_options": list(menu_options),
"errors": dict(errors or {}),
}
def async_create_entry(self, *, title, data, options=None):
return {"type": "create_entry", "title": title, "data": data,
"options": options}
@@ -182,6 +199,17 @@ def _install_stubs() -> None:
"description_placeholders": dict(description_placeholders or {}),
}
def async_show_menu(self, *, step_id, menu_options, sort=False,
description_placeholders=None, errors=None):
# t56: the real result shape (data_entry_flow sets type "menu" from
# FlowResultType.MENU, which is a StrEnum — "menu" on the wire).
return {
"type": "menu",
"step_id": step_id,
"menu_options": list(menu_options),
"errors": dict(errors or {}),
}
class _SuggestedMixin:
def add_suggested_values_to_schema(self, schema, suggested):
schema_dict = getattr(schema, "schema", None) or (
@@ -864,8 +892,10 @@ class TestNoDeadEndPairing:
)
assert form["step_id"] == "adb_pair"
result = await flow.async_step_adb_pair({CONF_SKIP_PAIRING: True})
skipped = await flow.async_step_adb_pair({CONF_SKIP_PAIRING: True})
assert skipped["type"] == "menu" # t56: the setup menu takes over
result = await flow.async_step_finish(None)
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
@@ -899,7 +929,11 @@ class TestNoDeadEndPairing:
result = await flow.async_step_adb_pair({CONF_SKIP_PAIRING: True})
assert result["step_id"] == "protocol_pair" # not a dead end either
# t56: the skip hands over to the setup MENU, and the unpaired protocol
# is still one row away — the escape hatch is intact, not a dead end
assert result["type"] == "menu"
assert "protocol" in result["menu_options"]
assert (await flow.async_step_protocol_pair(None))["step_id"] == "protocol_pair"
async def test_skip_protocol_after_skip_adb_creates_the_entry(
self, tmp_path, monkeypatch
@@ -941,15 +975,21 @@ class TestNoDeadEndPairing:
)
assert form["step_id"] == "protocol_pair"
result = await flow.async_step_protocol_pair(
paired = 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
# t56: a finished pairing returns to the setup menu (the just-paired
# transport is no longer offered), and the menu's Finish creates the entry
assert paired["type"] == "menu"
assert "protocol" not in paired["menu_options"]
assert client.finished_codes == [FAKE_PAIRING_CODE]
assert flow._unique_id == FAKE_MAC
result = await flow.async_step_finish(None)
assert result["type"] == "create_entry"
assert result["options"][CONF_PROTOCOL_ENABLED] is True
async def test_normal_adb_pair_path_still_creates_the_entry(self, tmp_path, monkeypatch):
module = _form_module()
test_connect = _patch_adb(monkeypatch, module)
@@ -965,11 +1005,17 @@ class TestNoDeadEndPairing:
)
assert form["step_id"] == "adb_pair"
result = await flow.async_step_adb_pair({CONF_ADB_PORT: 5555})
paired = await flow.async_step_adb_pair({CONF_ADB_PORT: 5555})
# t56: the ADB pairing is done, so the setup menu takes over; Finish
# (its own row, HA dispatches it) creates the entry
assert paired["type"] == "menu"
assert "finish" in paired["menu_options"]
result = await flow.async_step_finish(None)
test_connect.assert_awaited_once_with(DISCOVERED_HOST, 5555, KEY_PATH)
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."""
@@ -983,9 +1029,11 @@ class TestNoDeadEndPairing:
result = await flow.async_step_adb_pair(
{CONF_HOST: "10.0.0.77", CONF_ADB_PORT: 5555}
)
assert result["type"] == "menu"
created = await flow.async_step_finish(None)
assert result["type"] == "create_entry"
assert result["options"][CONF_HOST] == "10.0.0.77"
assert created["type"] == "create_entry"
assert created["options"][CONF_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):
@@ -1286,8 +1334,13 @@ class TestUnreachableShieldDuringPairing:
)
assert adb_form["step_id"] == "adb_pair"
result = await flow.async_step_adb_pair({CONF_SKIP_PAIRING: True})
skipped = await flow.async_step_adb_pair({CONF_SKIP_PAIRING: True})
assert skipped["type"] == "menu"
assert "protocol" in skipped["menu_options"]
# choosing it re-renders the reachable-again pairing form with the
# unreachable error and both escapes, exactly as before
result = await flow.async_step_protocol_pair(None)
assert result["type"] == "form"
assert result["step_id"] == "protocol_pair"
assert result["errors"] == {"base": "protocol_unreachable"}
@@ -1434,19 +1487,43 @@ class TestCopyContract:
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):
"""t56: EVERY sub-screen's fields carry a label, in both string files.
The page is a menu now, so the check walks the real menu: each rendered
sub-screen must have a step with a title, a description and a label for
every field it renders — and the menu itself must label every row it
offers.
"""
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"])
menu = await flow.async_step_init(None)
# the rendered sub-screens, keyed by the i18n step id the menu row maps to
rendered = {
step: vol_mirror.declared_fields(
(await getattr(flow, f"async_step_{step}")(None))["data_schema"]
)
for step in menu["menu_options"]
if step != "finish"
}
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"
menu_step = file["options"]["step"]["init"]
assert menu_step["description"]
# every row the menu offers has a label (HA renders these as names)
for option in menu["menu_options"]:
assert option in menu_step.get("menu", {}), f"menu row {option}"
steps = file["options"]["step"]
for step, fields in rendered.items():
# a sub-screen's i18n step key is its menu row id (HA dispatches
# a submitted form by its own step_id, so the two must match)
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"
def _options_flow(tmp_path, options=None):
@@ -1460,6 +1537,86 @@ def _options_flow(tmp_path, options=None):
return flow
# ── t56: driving the options MENU from tests ──
#
# The page is a menu now, so a test navigates two levels: choose a sub-screen
# (the menu result's own `menu_options` are the source of truth, the same list
# Home Assistant renders), then submit that sub-screen's form. Nothing here
# re-implements the flow's routing — a wrong step id fails loudly.
OPTIONS_SUBSTEP_FIELDS = {
"connection": {
CONF_HOST,
CONF_ADB_PORT,
CONF_ADB_ENABLED,
CONF_PROTOCOL_ENABLED,
CONF_ADB_STORE_APP_INFO,
CONF_JELLYFIN_ENTITY,
CONF_POWER_ENTITY,
CONF_LOGCAT_EVENTS,
CONF_ADB_COMMAND_ENABLED,
},
"receiver": {"receiver_entity", "receiver_input"},
"apps": {"app_name_map", "app_icon_map"},
"scripts": {
CONF_REMOTE_COMMAND_SCRIPT,
CONF_CHANNEL_DIGIT_DELAY,
CONF_COMMAND_SCRIPTS,
*{f"{SCRIPT_FIELD_PREFIX}{command}" for command in SCRIPT_COMMANDS},
},
}
def _options_substep_of(key: str) -> str:
for step, fields in OPTIONS_SUBSTEP_FIELDS.items():
if key in fields:
return step
if key.startswith(SCRIPT_FIELD_PREFIX):
return "scripts"
raise AssertionError(f"no options sub-screen owns the field {key!r}")
async def _options_submit(flow, payload: dict, *, save=False):
"""Submit `payload` through the MENU into the sub-screens that own its keys.
Returns the result of the LAST submitted sub-screen (the menu again after a
successful save), so a test can assert on form errors the way it did when the
page was one form. With ``save=True`` the menu's "Save and close" row is
chosen afterwards and the create_entry result comes back — that is what
actually commits the merged options.
"""
grouped: dict[str, dict] = {}
for key, value in payload.items():
grouped.setdefault(_options_substep_of(key), {})[key] = value
result = await flow.async_step_init(None)
for step, sub_payload in grouped.items():
assert step in result["menu_options"], (
f"{step} is not offered by the menu: {result['menu_options']}"
)
result = await getattr(flow, f"async_step_{step}")(sub_payload)
if save:
assert "finish" in result["menu_options"], result["menu_options"]
result = await flow.async_step_finish(None)
return result
def _all_script_fields(value):
"""Every per-command script field set to `value` — a FULL page submission.
The frontend submits every field the step rendered, so this is what a real
submit of the Scripts screen looks like; a partial dict is the artificial
shape only tests produce.
"""
return {f"{SCRIPT_FIELD_PREFIX}{command}": value for command in SCRIPT_COMMANDS}
async def _options_open(flow, step: str):
"""Choose a sub-screen from the menu and open its form."""
result = await flow.async_step_init(None)
assert step in result["menu_options"], result["menu_options"]
return await getattr(flow, f"async_step_{step}")(None)
class TestCommandScriptOptions:
"""t44: one script selector per command, stored as one dict option.
@@ -1478,7 +1635,7 @@ class TestCommandScriptOptions:
async def test_a_script_selector_exists_for_every_command(self, tmp_path):
flow = self._flow(tmp_path)
form = await flow.async_step_init(None)
form = await _options_open(flow, "scripts")
fields = set(vol_mirror.declared_fields(form["data_schema"]))
assert {f"{SCRIPT_FIELD_PREFIX}{command}" for command in SCRIPT_COMMANDS} <= fields
@@ -1486,11 +1643,13 @@ class TestCommandScriptOptions:
async def test_saving_stores_one_dict_option_and_no_flat_keys(self, tmp_path):
flow = self._flow(tmp_path)
result = await flow.async_step_init(
result = await _options_submit(
flow,
{
f"{SCRIPT_FIELD_PREFIX}volume_set": "script.shield_volume",
f"{SCRIPT_FIELD_PREFIX}turn_on": "script.shield_wake",
}
},
save=True,
)
assert result["type"] == "create_entry"
@@ -1506,7 +1665,7 @@ class TestCommandScriptOptions:
tmp_path, {CONF_COMMAND_SCRIPTS: {"turn_on": "script.shield_wake"}}
)
form = await flow.async_step_init(None)
form = await _options_open(flow, "scripts")
suggested = self._submitted(form["data_schema"]).get("__suggested__", {})
assert suggested[f"{SCRIPT_FIELD_PREFIX}turn_on"] == "script.shield_wake"
@@ -1514,12 +1673,12 @@ class TestCommandScriptOptions:
assert f"{SCRIPT_FIELD_PREFIX}volume_up" not in suggested
async def test_clearing_every_script_removes_the_option(self, tmp_path):
"""The editor is the truth: a save without scripts clears them."""
"""The editor is the truth: clearing every selector removes them all."""
flow = self._flow(
tmp_path, {CONF_COMMAND_SCRIPTS: {"turn_on": "script.shield_wake"}}
)
result = await flow.async_step_init({})
result = await _options_submit(flow, _all_script_fields(""), save=True)
assert result["data"][CONF_COMMAND_SCRIPTS] == {}
@@ -1529,15 +1688,34 @@ class TestCommandScriptOptions:
{CONF_COMMAND_SCRIPTS: {"turn_on": "script.shield_wake", "gone": "script.x"}},
)
result = await flow.async_step_init({f"{SCRIPT_FIELD_PREFIX}turn_on": "script.a"})
payload = _all_script_fields("")
payload[f"{SCRIPT_FIELD_PREFIX}turn_on"] = "script.a"
result = await _options_submit(flow, payload, save=True)
assert result["data"][CONF_COMMAND_SCRIPTS] == {"turn_on": "script.a"}
async def test_a_full_clear_of_a_freshly_typed_script_persists(self, tmp_path):
"""t56 live bug: a configured script must be removable from the UI.
The frontend submits the whole Scripts page, so "typed then cleared"
arrives as an empty value for that command — and the save must drop it
rather than keep a stale key.
"""
flow = self._flow(
tmp_path, {CONF_COMMAND_SCRIPTS: {"turn_on": "script.shield_wake"}}
)
payload = _all_script_fields("")
payload[f"{SCRIPT_FIELD_PREFIX}turn_on"] = ""
result = await _options_submit(flow, payload, save=True)
assert result["data"][CONF_COMMAND_SCRIPTS] == {}
async def test_the_page_still_renders_with_a_hand_edited_option(self, tmp_path):
"""A string where the dict belongs must not break the form."""
flow = self._flow(tmp_path, {CONF_COMMAND_SCRIPTS: "script.shield_wake"})
form = await flow.async_step_init(None)
form = await _options_open(flow, "scripts")
assert form["type"] == "form"
assert f"{SCRIPT_FIELD_PREFIX}turn_on" in vol_mirror.declared_fields(
@@ -1582,7 +1760,7 @@ class TestClearingAnEntityField:
flow = _options_flow(
tmp_path, {CONF_COMMAND_SCRIPTS: {"turn_on": "script.watch"}}
)
form = await flow.async_step_init(None)
form = await _options_open(flow, "scripts")
for value in ("", None):
assert self._validate(form, {f"{SCRIPT_FIELD_PREFIX}turn_on": value}) == {
@@ -1616,15 +1794,18 @@ class TestClearingAnEntityField:
flow = _options_flow(
tmp_path, {CONF_COMMAND_SCRIPTS: {"turn_on": "script.watch"}}
)
form = await flow.async_step_init(None)
form = await _options_open(flow, "scripts")
# what HA does: validate the submitted page, then run the step
result = await flow.async_step_init(
self._validate(form, {f"{SCRIPT_FIELD_PREFIX}turn_on": value})
)
# what HA does: validate the submitted page, then run the step — the
# whole Scripts page, with the one selector cleared
page = _all_script_fields("")
page[f"{SCRIPT_FIELD_PREFIX}turn_on"] = value
result = await flow.async_step_scripts(self._validate(form, page))
assert result["type"] == "menu" # back to the menu, ready to save
assert result["type"] == "create_entry"
assert result["data"][CONF_COMMAND_SCRIPTS] == {}
created = await flow.async_step_finish(None)
assert created["type"] == "create_entry"
assert created["data"][CONF_COMMAND_SCRIPTS] == {}
@pytest.mark.parametrize("value", ["", None])
async def test_an_explicit_empty_value_clears_a_preexisting_entity_field(
@@ -1635,34 +1816,42 @@ class TestClearingAnEntityField:
tmp_path,
{CONF_JELLYFIN_ENTITY: "media_player.jellyfin", CONF_HOST: "10.0.0.9"},
)
form = await flow.async_step_init(None)
form = await _options_open(flow, "connection")
result = await flow.async_step_init(
await flow.async_step_connection(
self._validate(form, {CONF_JELLYFIN_ENTITY: value})
)
result = await flow.async_step_finish(None)
assert not result["data"][CONF_JELLYFIN_ENTITY] # cleared, not kept
assert result["data"][CONF_HOST] == "10.0.0.9" # untouched fields survive
async def test_an_omitted_script_field_is_cleared_as_well(self, tmp_path):
"""The script dict is rebuilt from the page, so omitting a command clears it."""
"""The script dict is rebuilt from the page, so an empty page clears it."""
flow = _options_flow(
tmp_path, {CONF_COMMAND_SCRIPTS: {"turn_on": "script.watch"}}
)
form = await flow.async_step_init(None)
form = await _options_open(flow, "scripts")
result = await flow.async_step_init(self._validate(form, {}))
# a full page submission with every selector empty (the frontend's shape
# when the user clears them all)
page = self._validate(form, _all_script_fields(""))
await flow.async_step_scripts(page)
result = await flow.async_step_finish(None)
assert result["data"][CONF_COMMAND_SCRIPTS] == {}
async def test_an_omitted_entity_field_keeps_its_stored_value(self, tmp_path):
"""An omitted key is "not touched" for the patched options, not a clear."""
"""A key the submission does NOT carry is untouched, not a clear (t56)."""
flow = _options_flow(
tmp_path, {CONF_JELLYFIN_ENTITY: "media_player.jellyfin"}
)
form = await flow.async_step_init(None)
form = await _options_open(flow, "connection")
result = await flow.async_step_init(self._validate(form, {CONF_HOST: "10.0.0.9"}))
await flow.async_step_connection(
self._validate(form, {CONF_HOST: "10.0.0.9"})
)
result = await flow.async_step_finish(None)
assert result["data"][CONF_JELLYFIN_ENTITY] == "media_player.jellyfin"
@@ -1674,22 +1863,29 @@ class TestClearingAnEntityField:
CONF_HOST: "10.0.0.9",
}
flow = _options_flow(tmp_path, stored)
form = await flow.async_step_init(None)
suggested = form["data_schema"].get("__suggested__", {})
# the page comes back carrying the stored values, and the user submits them
scripts_form = await _options_open(flow, "scripts")
suggested = scripts_form["data_schema"].get("__suggested__", {})
assert suggested[f"{SCRIPT_FIELD_PREFIX}turn_on"] == "script.watch"
result = await flow.async_step_init(
connection_form = await _options_open(flow, "connection")
# the two sub-screens come back carrying the stored values, and the user
# submits both pages unchanged
await flow.async_step_scripts(
self._validate(
form,
scripts_form, {f"{SCRIPT_FIELD_PREFIX}turn_on": "script.watch"}
)
)
await flow.async_step_connection(
self._validate(
connection_form,
{
f"{SCRIPT_FIELD_PREFIX}turn_on": "script.watch",
CONF_JELLYFIN_ENTITY: "media_player.jellyfin",
CONF_POWER_ENTITY: "switch.plug",
CONF_HOST: "10.0.0.9",
},
)
)
result = await flow.async_step_finish(None)
assert result["data"][CONF_COMMAND_SCRIPTS] == {"turn_on": "script.watch"}
assert result["data"][CONF_JELLYFIN_ENTITY] == "media_player.jellyfin"
@@ -1702,14 +1898,14 @@ class TestRemoteCommandScriptOption:
async def test_the_field_is_offered_and_stored_as_a_string(self, tmp_path):
flow = _options_flow(tmp_path)
form = await flow.async_step_init(None)
form = await _options_open(flow, "scripts")
assert CONF_REMOTE_COMMAND_SCRIPT in vol_mirror.declared_fields(
form["data_schema"]
)
result = await flow.async_step_init(
{CONF_REMOTE_COMMAND_SCRIPT: "script.shield_remote"}
result = await _options_submit(
flow, {CONF_REMOTE_COMMAND_SCRIPT: "script.shield_remote"}, save=True
)
assert result["data"][CONF_REMOTE_COMMAND_SCRIPT] == "script.shield_remote"
@@ -1718,7 +1914,7 @@ class TestRemoteCommandScriptOption:
flow = _options_flow(
tmp_path, {CONF_REMOTE_COMMAND_SCRIPT: "script.shield_remote"}
)
form = await flow.async_step_init(None)
form = await _options_open(flow, "scripts")
assert form["data_schema"]["__suggested__"][CONF_REMOTE_COMMAND_SCRIPT] == (
"script.shield_remote"
@@ -1727,8 +1923,8 @@ class TestRemoteCommandScriptOption:
async def test_a_padded_value_is_trimmed(self, tmp_path):
flow = _options_flow(tmp_path)
result = await flow.async_step_init(
{CONF_REMOTE_COMMAND_SCRIPT: " script.shield_remote "}
result = await _options_submit(
flow, {CONF_REMOTE_COMMAND_SCRIPT: " script.shield_remote "}, save=True
)
assert result["data"][CONF_REMOTE_COMMAND_SCRIPT] == "script.shield_remote"
@@ -1737,12 +1933,13 @@ class TestRemoteCommandScriptOption:
flow = _options_flow(
tmp_path, {CONF_REMOTE_COMMAND_SCRIPT: "script.shield_remote"}
)
form = await flow.async_step_init(None)
form = await _options_open(flow, "scripts")
validated = TestClearingAnEntityField._validate(
form, {CONF_REMOTE_COMMAND_SCRIPT: ""}
)
result = await flow.async_step_init(validated)
await flow.async_step_scripts(validated)
result = await flow.async_step_finish(None)
assert result["data"][CONF_REMOTE_COMMAND_SCRIPT] == ""
@@ -1770,3 +1967,250 @@ class TestPassiveSetupNeverDemandsReadiness:
assert "ConfigEntryNotReady" not in body # never demands readiness
assert "raise " not in body
class TestOptionsMenu:
"""t56: the options page is a MENU of sub-screens, not one long form."""
async def test_the_menu_offers_every_sub_screen_and_save(self, tmp_path):
flow = _options_flow(tmp_path)
menu = await flow.async_step_init(None)
assert menu["type"] == "menu"
assert menu["step_id"] == "init"
assert list(menu["menu_options"]) == [
"connection",
"receiver",
"apps",
"scripts",
"finish",
]
async def test_every_menu_row_has_a_handler_named_after_it(self, tmp_path):
"""The invariant that makes a menu work at all (found via real HA).
Home Assistant dispatches a menu choice to ``async_step_<row id>`` and a
submitted form to ``async_step_<its own step_id>``, so a row whose id has
no matching method is an UnknownStep, and a sub-screen whose step_id is
not its row id cannot be re-submitted. Both are pinned here.
"""
flow = _options_flow(tmp_path)
menu = await flow.async_step_init(None)
for row in menu["menu_options"]:
assert hasattr(flow, f"async_step_{row}"), row
if row == "finish":
continue
form = await getattr(flow, f"async_step_{row}")(None)
assert form["type"] == "form"
assert form["step_id"] == row, form["step_id"]
async def test_every_sub_screen_is_reachable_and_returns_to_the_menu(self, tmp_path):
flow = _options_flow(tmp_path)
for row in ("connection", "receiver", "apps", "scripts"):
opened = await _options_open(flow, row)
assert opened["type"] == "form"
# a submission that changes nothing still returns to the menu
back = await getattr(flow, f"async_step_{row}")({})
assert back["type"] == "menu", (row, back)
async def test_edits_on_one_sub_screen_are_not_lost_by_another(self, tmp_path):
"""The merge is key-scoped, so the order of sub-screens cannot matter."""
flow = _options_flow(tmp_path, options={})
await _options_submit(
flow,
{"host": "10.0.0.44", "receiver_entity": "switch.soundbar"},
)
await _options_submit(
flow,
{"app_name_map": "com.plex.hdx=Plex"},
)
result = await flow.async_step_finish(None)
assert result["data"]["host"] == "10.0.0.44"
assert result["data"]["receiver_entity"] == "switch.soundbar"
assert result["data"]["app_name_map"] == {"com.plex.hdx": "Plex"}
async def test_closing_without_finish_keeps_the_entry_untouched(self, tmp_path):
"""Only "Save and close" commits; visiting a sub-screen never does."""
stored = {CONF_HOST: "10.0.0.9"}
flow = _options_flow(tmp_path, stored)
await _options_submit(flow, {"host": "10.0.0.44"})
# the flow's pending buffer changed, the entry's options did not
assert flow._options_dict[CONF_HOST] == "10.0.0.44"
assert flow.config_entry.options == stored
class TestAppsTable:
"""t56: the Apps sub-screen is an editable one-row-per-app table."""
async def test_the_suggestion_is_text_never_a_dict(self, tmp_path):
"""The [object Object] root cause, pinned at the layer that caused it."""
flow = _options_flow(
tmp_path,
{"app_name_map": {"ar.tvplayer.tv": "TiviMate", "com.plex.hdx": "Plex"}},
)
form = await _options_open(flow, "apps")
suggested = form["data_schema"]["__suggested__"]
assert isinstance(suggested["app_name_map"], str)
assert not isinstance(suggested["app_name_map"], (dict, list))
assert suggested["app_name_map"] == "ar.tvplayer.tv=TiviMate\ncom.plex.hdx=Plex"
async def test_a_row_is_added_with_its_icon_and_prefilled(self, tmp_path):
flow = _options_flow(tmp_path, options={})
await _options_submit(
flow, {"app_name_map": "com.plex.hdx=Plex|http://icons.local/plex.png"}
)
result = await flow.async_step_finish(None)
assert result["data"]["app_name_map"] == {"com.plex.hdx": "Plex"}
assert result["data"]["app_icon_map"] == {
"com.plex.hdx": "http://icons.local/plex.png"
}
# re-open: the row (icon included) comes back in the text field
reopened = _options_flow(tmp_path, result["data"])
form = await _options_open(reopened, "apps")
assert (
form["data_schema"]["__suggested__"]["app_name_map"]
== "com.plex.hdx=Plex|http://icons.local/plex.png"
)
async def test_a_row_is_edited_in_place(self, tmp_path):
flow = _options_flow(
tmp_path, {"app_name_map": {"com.plex.hdx": "Plex"}, "host": "10.0.0.9"}
)
await _options_submit(flow, {"app_name_map": "com.plex.hdx=Plex (living room)"})
result = await flow.async_step_finish(None)
assert result["data"]["app_name_map"] == {"com.plex.hdx": "Plex (living room)"}
assert result["data"]["host"] == "10.0.0.9" # another screen's field
async def test_a_row_is_removed_by_dropping_its_line(self, tmp_path):
flow = _options_flow(
tmp_path,
{"app_name_map": {"com.plex.hdx": "Plex", "ar.tvplayer.tv": "TiviMate"}},
)
await _options_submit(flow, {"app_name_map": "ar.tvplayer.tv=TiviMate"})
result = await flow.async_step_finish(None)
assert result["data"]["app_name_map"] == {"ar.tvplayer.tv": "TiviMate"}
async def test_the_empty_state_renders_without_data_or_a_crash(self, tmp_path):
flow = _options_flow(tmp_path, options={})
form = await _options_open(flow, "apps")
assert form["type"] == "form"
assert form["step_id"] == "apps"
assert form["errors"] == {}
assert list(vol_mirror.declared_fields(form["data_schema"])) == ["app_name_map"]
# no discovery data yet, so nothing is suggested — an empty text field,
# which is exactly the documented "blank uses the built-in list"
assert "app_name_map" not in form["data_schema"].get("__suggested__", {})
async def test_a_cleared_table_is_an_empty_map_and_not_an_error(self, tmp_path):
flow = _options_flow(tmp_path, {"app_name_map": {"com.plex.hdx": "Plex"}})
result = await _options_submit(flow, {"app_name_map": " "}, save=True)
assert result["type"] == "create_entry"
assert result["data"]["app_name_map"] == {}
class TestReceiverAcceptsASwitch:
"""t56: a soundbar that only knows on/off is a valid receiver."""
def test_the_selector_offers_switch_and_media_player(self):
module = _form_module()
config = module._build_selectors()["receiver"].config
assert sorted(config.domain) == ["media_player", "switch"]
async def test_a_switch_stores_and_prefills(self, tmp_path):
flow = _options_flow(tmp_path, options={})
result = await _options_submit(
flow,
{"receiver_entity": "switch.soundbar", "receiver_input": "HDMI 2"},
save=True,
)
assert result["data"]["receiver_entity"] == "switch.soundbar"
assert result["data"]["receiver_input"] == "HDMI 2"
reopened = _options_flow(tmp_path, result["data"])
suggested = (await _options_open(reopened, "receiver"))["data_schema"][
"__suggested__"
]
assert suggested["receiver_entity"] == "switch.soundbar"
class TestSetupMenu:
"""t56: the setup flow keeps its first form and adds a menu after it."""
async def test_the_setup_menu_offers_the_sub_screens_and_finish(
self, tmp_path, monkeypatch
):
module = _form_module()
_patch_adb(monkeypatch, module)
flow, _hass = _new_flow(module, tmp_path)
await flow.async_step_user(
_user_input(**{CONF_ADB_ENABLED: False, CONF_PROTOCOL_ENABLED: False})
)
menu = await flow.async_step_menu(None)
assert menu["type"] == "menu"
assert menu["step_id"] == "menu"
assert list(menu["menu_options"]) == ["receiver", "connection", "finish"]
async def test_every_setup_row_has_a_handler_named_after_it(
self, tmp_path, monkeypatch
):
module = _form_module()
_patch_adb(monkeypatch, module)
flow, _hass = _new_flow(module, tmp_path)
await flow.async_step_user(
_user_input(**{CONF_ADB_ENABLED: False, CONF_PROTOCOL_ENABLED: False})
)
menu = await flow.async_step_menu(None)
for row in menu["menu_options"]:
assert hasattr(flow, f"async_step_{row}"), row
# ...and the row ids never include the menu itself, which would loop
assert "menu" not in menu["menu_options"]
async def test_connection_and_receiver_edits_both_land_in_the_entry(
self, tmp_path, monkeypatch
):
module = _form_module()
_patch_adb(monkeypatch, module)
flow, _hass = _new_flow(module, tmp_path)
await flow.async_step_user(
_user_input(
**{
CONF_ADB_ENABLED: False,
CONF_PROTOCOL_ENABLED: False,
CONF_HOST: DISCOVERED_HOST,
}
)
)
await flow.async_step_connection({CONF_HOST: "10.0.0.77"})
await flow.async_step_receiver(
{"receiver_entity": "switch.soundbar", "receiver_input": "HDMI 1"}
)
result = await flow.async_step_finish(None)
assert result["type"] == "create_entry"
assert flow._host() == "10.0.0.77"
assert result["options"].get("receiver_entity") == "switch.soundbar"
assert result["options"].get("receiver_input") == "HDMI 1"