Not-Expected-Yet/ha-nvidia-shield
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity
Files
ee3f2845af3fa72f4b9392603170875f81e733f8
ha-nvidia-shield/tests/test_config_flow_real_ha.py
T
James Mackay ca3fdec218 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.
2026-10-12 07:17:41 +01:00

481 lines
19 KiB
Python
Raw Blame History

"""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.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 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",
"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 == "finish":
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: SET, CLEAR, SAVE, REOPEN — the live symptom, end to end.
Scoped deliberately: the whole Scripts page goes through the real flow
manager with every selector cleared (which is the shape the live bug was
about), and the SET half is proven separately against the real selector
objects in TestRealSelectorValidation below. A page carrying ENTITY IDS
cannot be driven here — real HA validates an entity selector against the
loaded entity/device registries and a stubbed core refuses the ids; the
selector-object test is the honest place for that half.
"""
entry = make_entry(ha)
manager, scripts = await open_scripts(ha, entry)
# 1. CLEAR: the frontend submits the whole page, all selectors empty
cleared = full_script_page("")
after_clear = await manager.async_configure(scripts["flow_id"], cleared)
assert after_clear["type"] == "menu", after_clear
created = await manager.async_configure(
after_clear["flow_id"], {"next_step_id": "finish"}
)
assert created["type"] == "create_entry"
assert created["data"]["command_scripts"] == {}
# 2. PERSIST: re-open the page over the SAVED options — the cleared
# commands must not come back, and no flat script_* key was stored
saved = dict(entry.options)
saved.update(created["data"])
object.__setattr__(entry, "_options", saved)
assert not [k for k in entry.options if k.startswith(SCRIPT_FIELD_PREFIX)]
manager2, scripts2 = await open_scripts(ha, entry)
values = suggested(scripts2["data_schema"])
assert not [v for k, v in values.items() if k.startswith(SCRIPT_FIELD_PREFIX) and v]
@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 ONE place the real selectors are exercised with real entity ids.
Driving an entity-valued page through a stubbed FlowManager is not possible
(see TestRealScriptClearing), so the distinction this task turns on is pinned
here directly against the real selector classes: 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"
Reference in new issue View Git Blame Copy Permalink