From b3d1574a7143ae7424043ba282f0464af0c5824b Mon Sep 17 00:00:00 2001 From: Matthias Alphart Date: Sun, 10 Apr 2022 15:56:45 +0200 Subject: [PATCH] Refactor KNX config flow and validate user input (#69698) * validate config flow user input * test flow for invalid user input * validate multicast address blocks * Update homeassistant/components/knx/config_flow.py Co-authored-by: Marvin Wichmann Co-authored-by: Marvin Wichmann --- homeassistant/components/knx/config_flow.py | 254 ++++++++++-------- homeassistant/components/knx/const.py | 5 +- homeassistant/components/knx/schema.py | 20 +- homeassistant/components/knx/strings.json | 18 +- .../components/knx/translations/en.json | 10 +- tests/components/knx/test_config_flow.py | 54 +++- 6 files changed, 226 insertions(+), 135 deletions(-) diff --git a/homeassistant/components/knx/config_flow.py b/homeassistant/components/knx/config_flow.py index e45eb3a87a1..2e3036e9a32 100644 --- a/homeassistant/components/knx/config_flow.py +++ b/homeassistant/components/knx/config_flow.py @@ -24,7 +24,6 @@ from .const import ( CONF_KNX_DEFAULT_RATE_LIMIT, CONF_KNX_DEFAULT_STATE_UPDATER, CONF_KNX_INDIVIDUAL_ADDRESS, - CONF_KNX_INITIAL_CONNECTION_TYPES, CONF_KNX_KNXKEY_FILENAME, CONF_KNX_KNXKEY_PASSWORD, CONF_KNX_LOCAL_IP, @@ -44,18 +43,19 @@ from .const import ( DOMAIN, KNXConfigEntryData, ) +from .schema import ia_validator, ip_v4_validator CONF_KNX_GATEWAY: Final = "gateway" CONF_MAX_RATE_LIMIT: Final = 60 CONF_DEFAULT_LOCAL_IP: Final = "0.0.0.0" -DEFAULT_ENTRY_DATA: KNXConfigEntryData = { - CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER, - CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT, - CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS, - CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, - CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT, -} +DEFAULT_ENTRY_DATA = KNXConfigEntryData( + individual_address=XKNX.DEFAULT_ADDRESS, + multicast_group=DEFAULT_MCAST_GRP, + multicast_port=DEFAULT_MCAST_PORT, + state_updater=CONF_KNX_DEFAULT_STATE_UPDATER, + rate_limit=CONF_KNX_DEFAULT_RATE_LIMIT, +) CONF_KNX_TUNNELING_TYPE: Final = "tunneling_type" CONF_KNX_LABEL_TUNNELING_TCP: Final = "TCP" @@ -101,10 +101,9 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): if user_input is not None: connection_type = user_input[CONF_KNX_CONNECTION_TYPE] if connection_type == CONF_KNX_AUTOMATIC: - entry_data: KNXConfigEntryData = { - **DEFAULT_ENTRY_DATA, # type: ignore[misc] - CONF_KNX_CONNECTION_TYPE: user_input[CONF_KNX_CONNECTION_TYPE], - } + entry_data = DEFAULT_ENTRY_DATA | KNXConfigEntryData( + connection_type=CONF_KNX_AUTOMATIC + ) return self.async_create_entry( title=CONF_KNX_AUTOMATIC.capitalize(), data=entry_data, @@ -118,13 +117,15 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): return await self.async_step_manual_tunnel() - errors: dict = {} - supported_connection_types = CONF_KNX_INITIAL_CONNECTION_TYPES.copy() - gateways = await scan_for_gateways() - - if gateways: - # add automatic only if a gateway responded - supported_connection_types.insert(0, CONF_KNX_AUTOMATIC) + supported_connection_types = { + CONF_KNX_TUNNELING: CONF_KNX_TUNNELING.capitalize(), + CONF_KNX_ROUTING: CONF_KNX_ROUTING.capitalize(), + } + if gateways := await scan_for_gateways(): + # add automatic at first position only if a gateway responded + supported_connection_types = { + CONF_KNX_AUTOMATIC: CONF_KNX_AUTOMATIC.capitalize() + } | supported_connection_types self._found_tunnels = [ gateway for gateway in gateways if gateway.supports_tunnelling ] @@ -132,10 +133,7 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): fields = { vol.Required(CONF_KNX_CONNECTION_TYPE): vol.In(supported_connection_types) } - - return self.async_show_form( - step_id="type", data_schema=vol.Schema(fields), errors=errors - ) + return self.async_show_form(step_id="type", data_schema=vol.Schema(fields)) async def async_step_tunnel(self, user_input: dict | None = None) -> FlowResult: """Select a tunnel from a list. Will be skipped if the gateway scan was unsuccessful or if only one gateway was found.""" @@ -164,37 +162,48 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): self, user_input: dict | None = None ) -> FlowResult: """Manually configure tunnel connection parameters. Fields default to preselected gateway if one was found.""" + errors: dict = {} + if user_input is not None: - connection_type = user_input[CONF_KNX_TUNNELING_TYPE] + try: + _host = ip_v4_validator(user_input[CONF_HOST], multicast=False) + except vol.Invalid: + errors[CONF_HOST] = "invalid_ip_address" - entry_data: KNXConfigEntryData = { - **DEFAULT_ENTRY_DATA, # type: ignore[misc] - CONF_HOST: user_input[CONF_HOST], - CONF_PORT: user_input[CONF_PORT], - CONF_KNX_ROUTE_BACK: ( - connection_type == CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK - ), - CONF_KNX_LOCAL_IP: user_input.get(CONF_KNX_LOCAL_IP), - CONF_KNX_CONNECTION_TYPE: ( - CONF_KNX_TUNNELING_TCP - if connection_type == CONF_KNX_LABEL_TUNNELING_TCP - else CONF_KNX_TUNNELING - ), - } + if _local_ip := user_input.get(CONF_KNX_LOCAL_IP): + try: + _local_ip = ip_v4_validator(_local_ip, multicast=False) + except vol.Invalid: + errors[CONF_KNX_LOCAL_IP] = "invalid_ip_address" - if connection_type == CONF_KNX_LABEL_TUNNELING_TCP_SECURE: - self._tunneling_config = entry_data - return self.async_show_menu( - step_id="secure_tunneling", - menu_options=["secure_knxkeys", "secure_manual"], + if not errors: + connection_type = user_input[CONF_KNX_TUNNELING_TYPE] + entry_data = DEFAULT_ENTRY_DATA | KNXConfigEntryData( + host=_host, + port=user_input[CONF_PORT], + route_back=( + connection_type == CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK + ), + local_ip=_local_ip, + connection_type=( + CONF_KNX_TUNNELING_TCP + if connection_type == CONF_KNX_LABEL_TUNNELING_TCP + else CONF_KNX_TUNNELING + ), ) - return self.async_create_entry( - title=f"{CONF_KNX_TUNNELING.capitalize()} @ {user_input[CONF_HOST]}", - data=entry_data, - ) + if connection_type == CONF_KNX_LABEL_TUNNELING_TCP_SECURE: + self._tunneling_config = entry_data + return self.async_show_menu( + step_id="secure_tunneling", + menu_options=["secure_knxkeys", "secure_manual"], + ) + + return self.async_create_entry( + title=f"Tunneling @ {_host}", + data=entry_data, + ) - errors: dict = {} connection_methods: list[str] = [ CONF_KNX_LABEL_TUNNELING_TCP, CONF_KNX_LABEL_TUNNELING_UDP, @@ -231,20 +240,15 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): if user_input is not None: assert self._tunneling_config - entry_data: KNXConfigEntryData = { - **self._tunneling_config, # type: ignore[misc] - CONF_KNX_SECURE_USER_ID: user_input[CONF_KNX_SECURE_USER_ID], - CONF_KNX_SECURE_USER_PASSWORD: user_input[ - CONF_KNX_SECURE_USER_PASSWORD - ], - CONF_KNX_SECURE_DEVICE_AUTHENTICATION: user_input[ - CONF_KNX_SECURE_DEVICE_AUTHENTICATION - ], - CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, - } + entry_data = self._tunneling_config | KNXConfigEntryData( + connection_type=CONF_KNX_TUNNELING_TCP_SECURE, + device_authentication=user_input[CONF_KNX_SECURE_DEVICE_AUTHENTICATION], + user_id=user_input[CONF_KNX_SECURE_USER_ID], + user_password=user_input[CONF_KNX_SECURE_USER_PASSWORD], + ) return self.async_create_entry( - title=f"Secure {CONF_KNX_TUNNELING.capitalize()} @ {self._tunneling_config[CONF_HOST]}", + title=f"Secure Tunneling @ {self._tunneling_config[CONF_HOST]}", data=entry_data, ) @@ -272,33 +276,29 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): errors = {} if user_input is not None: + assert self._tunneling_config + storage_key = CONST_KNX_STORAGE_KEY + user_input[CONF_KNX_KNXKEY_FILENAME] try: - assert self._tunneling_config - storage_key: str = ( - CONST_KNX_STORAGE_KEY + user_input[CONF_KNX_KNXKEY_FILENAME] - ) load_key_ring( - self.hass.config.path( - STORAGE_DIR, - storage_key, - ), - user_input[CONF_KNX_KNXKEY_PASSWORD], + path=self.hass.config.path(STORAGE_DIR, storage_key), + password=user_input[CONF_KNX_KNXKEY_PASSWORD], + ) + except FileNotFoundError: + errors[CONF_KNX_KNXKEY_FILENAME] = "file_not_found" + except InvalidSignature: + errors[CONF_KNX_KNXKEY_PASSWORD] = "invalid_signature" + + if not errors: + entry_data = self._tunneling_config | KNXConfigEntryData( + connection_type=CONF_KNX_TUNNELING_TCP_SECURE, + knxkeys_filename=storage_key, + knxkeys_password=user_input[CONF_KNX_KNXKEY_PASSWORD], ) - entry_data: KNXConfigEntryData = { - **self._tunneling_config, # type: ignore[misc] - CONF_KNX_KNXKEY_FILENAME: storage_key, - CONF_KNX_KNXKEY_PASSWORD: user_input[CONF_KNX_KNXKEY_PASSWORD], - CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, - } return self.async_create_entry( - title=f"Secure {CONF_KNX_TUNNELING.capitalize()} @ {self._tunneling_config[CONF_HOST]}", + title=f"Secure Tunneling @ {self._tunneling_config[CONF_HOST]}", data=entry_data, ) - except InvalidSignature: - errors["base"] = "invalid_signature" - except FileNotFoundError: - errors["base"] = "file_not_found" fields = { vol.Required(CONF_KNX_KNXKEY_FILENAME): selector.selector({"text": {}}), @@ -311,33 +311,55 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): async def async_step_routing(self, user_input: dict | None = None) -> FlowResult: """Routing setup.""" - if user_input is not None: - return self.async_create_entry( - title=CONF_KNX_ROUTING.capitalize(), - data={ - **DEFAULT_ENTRY_DATA, - CONF_KNX_MCAST_GRP: user_input[CONF_KNX_MCAST_GRP], - CONF_KNX_MCAST_PORT: user_input[CONF_KNX_MCAST_PORT], - CONF_KNX_INDIVIDUAL_ADDRESS: user_input[ - CONF_KNX_INDIVIDUAL_ADDRESS - ], - CONF_KNX_LOCAL_IP: user_input.get(CONF_KNX_LOCAL_IP), - CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, - }, - ) - errors: dict = {} + _individual_address = ( + user_input[CONF_KNX_INDIVIDUAL_ADDRESS] + if user_input + else XKNX.DEFAULT_ADDRESS + ) + _multicast_group = ( + user_input[CONF_KNX_MCAST_GRP] if user_input else DEFAULT_MCAST_GRP + ) + + if user_input is not None: + try: + ia_validator(_individual_address) + except vol.Invalid: + errors[CONF_KNX_INDIVIDUAL_ADDRESS] = "invalid_individual_address" + try: + ip_v4_validator(_multicast_group, multicast=True) + except vol.Invalid: + errors[CONF_KNX_MCAST_GRP] = "invalid_ip_address" + if _local_ip := user_input.get(CONF_KNX_LOCAL_IP): + try: + ip_v4_validator(_local_ip, multicast=False) + except vol.Invalid: + errors[CONF_KNX_LOCAL_IP] = "invalid_ip_address" + + if not errors: + entry_data = DEFAULT_ENTRY_DATA | KNXConfigEntryData( + connection_type=CONF_KNX_ROUTING, + individual_address=_individual_address, + multicast_group=_multicast_group, + multicast_port=user_input[CONF_KNX_MCAST_PORT], + local_ip=_local_ip, + ) + return self.async_create_entry( + title=CONF_KNX_ROUTING.capitalize(), data=entry_data + ) + fields = { vol.Required( - CONF_KNX_INDIVIDUAL_ADDRESS, default=XKNX.DEFAULT_ADDRESS + CONF_KNX_INDIVIDUAL_ADDRESS, default=_individual_address ): _IA_SELECTOR, - vol.Required(CONF_KNX_MCAST_GRP, default=DEFAULT_MCAST_GRP): _IP_SELECTOR, + vol.Required(CONF_KNX_MCAST_GRP, default=_multicast_group): _IP_SELECTOR, vol.Required( CONF_KNX_MCAST_PORT, default=DEFAULT_MCAST_PORT ): _PORT_SELECTOR, } if self.show_advanced_options: + # Optional with default doesn't work properly in flow UI fields[vol.Optional(CONF_KNX_LOCAL_IP)] = _IP_SELECTOR return self.async_show_form( @@ -477,38 +499,34 @@ class KNXOptionsFlowHandler(OptionsFlow): last_step=True, ) - entry_data = { - **DEFAULT_ENTRY_DATA, - **self.general_settings, - CONF_KNX_LOCAL_IP: self.general_settings.get(CONF_KNX_LOCAL_IP) - if self.general_settings.get(CONF_KNX_LOCAL_IP) != CONF_DEFAULT_LOCAL_IP - else None, - CONF_HOST: self.current_config.get(CONF_HOST, ""), - } + _local_ip = self.general_settings.get(CONF_KNX_LOCAL_IP) + entry_data = ( + DEFAULT_ENTRY_DATA + | self.general_settings + | KNXConfigEntryData( + host=self.current_config.get(CONF_HOST, ""), + local_ip=_local_ip if _local_ip != CONF_DEFAULT_LOCAL_IP else None, + ) + ) if user_input is not None: connection_type = user_input[CONF_KNX_TUNNELING_TYPE] - entry_data = { - **entry_data, - CONF_HOST: user_input[CONF_HOST], - CONF_PORT: user_input[CONF_PORT], - CONF_KNX_ROUTE_BACK: ( - connection_type == CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK - ), - CONF_KNX_CONNECTION_TYPE: ( + entry_data = entry_data | KNXConfigEntryData( + host=user_input[CONF_HOST], + port=user_input[CONF_PORT], + route_back=(connection_type == CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK), + connection_type=( CONF_KNX_TUNNELING_TCP if connection_type == CONF_KNX_LABEL_TUNNELING_TCP else CONF_KNX_TUNNELING ), - } + ) entry_title = str(entry_data[CONF_KNX_CONNECTION_TYPE]).capitalize() if entry_data[CONF_KNX_CONNECTION_TYPE] == CONF_KNX_TUNNELING: - entry_title = f"{CONF_KNX_TUNNELING.capitalize()} @ {entry_data[CONF_HOST]}" + entry_title = f"Tunneling @ {entry_data[CONF_HOST]}" if entry_data[CONF_KNX_CONNECTION_TYPE] == CONF_KNX_TUNNELING_TCP: - entry_title = ( - f"{CONF_KNX_TUNNELING.capitalize()} (TCP) @ {entry_data[CONF_HOST]}" - ) + entry_title = f"Tunneling @ {entry_data[CONF_HOST]} (TCP)" self.hass.config_entries.async_update_entry( self.config_entry, diff --git a/homeassistant/components/knx/const.py b/homeassistant/components/knx/const.py index efe091f22a9..44ab2fab33a 100644 --- a/homeassistant/components/knx/const.py +++ b/homeassistant/components/knx/const.py @@ -1,4 +1,6 @@ """Constants for the KNX integration.""" +from __future__ import annotations + from enum import Enum from typing import Final, TypedDict @@ -68,7 +70,6 @@ CONF_RESET_AFTER: Final = "reset_after" CONF_RESPOND_TO_READ: Final = "respond_to_read" CONF_STATE_ADDRESS: Final = "state_address" CONF_SYNC_STATE: Final = "sync_state" -CONF_KNX_INITIAL_CONNECTION_TYPES: Final = [CONF_KNX_TUNNELING, CONF_KNX_ROUTING] # yaml config merged with config entry data DATA_KNX_CONFIG: Final = "knx_config" @@ -84,7 +85,7 @@ class KNXConfigEntryData(TypedDict, total=False): connection_type: str individual_address: str - local_ip: str + local_ip: str | None multicast_group: str multicast_port: int route_back: bool diff --git a/homeassistant/components/knx/schema.py b/homeassistant/components/knx/schema.py index 5c5b05db62d..bec5b5e5bda 100644 --- a/homeassistant/components/knx/schema.py +++ b/homeassistant/components/knx/schema.py @@ -3,6 +3,7 @@ from __future__ import annotations from abc import ABC from collections import OrderedDict +import ipaddress from typing import Any, ClassVar, Final import voluptuous as vol @@ -70,12 +71,29 @@ def ga_validator(value: Any) -> str | int: ga_list_validator = vol.All(cv.ensure_list, [ga_validator]) ia_validator = vol.Any( - cv.matches_regex(IndividualAddress.ADDRESS_RE.pattern), + vol.All(str, str.strip, cv.matches_regex(IndividualAddress.ADDRESS_RE.pattern)), vol.All(vol.Coerce(int), vol.Range(min=1, max=65535)), msg="value does not match pattern for KNX individual address '..' (eg.'1.1.100')", ) +def ip_v4_validator(value: Any, multicast: bool | None = None) -> str: + """ + Validate that value is parsable as IPv4 address. + + Optionally check if address is in a reserved multicast block or is explicitly not. + """ + try: + address = ipaddress.IPv4Address(value) + except ipaddress.AddressValueError as ex: + raise vol.Invalid(f"value '{value}' is not a valid IPv4 address: {ex}") from ex + if multicast is not None and address.is_multicast != multicast: + raise vol.Invalid( + f"value '{value}' is not a valid IPv4 {'multicast' if multicast else 'unicast'} address" + ) + return str(address) + + def number_limit_sub_validator(entity_config: OrderedDict) -> OrderedDict: """Validate a number entity configurations dependent on configured value type.""" value_type = entity_config[CONF_TYPE] diff --git a/homeassistant/components/knx/strings.json b/homeassistant/components/knx/strings.json index 2149dd96a47..018db071adf 100644 --- a/homeassistant/components/knx/strings.json +++ b/homeassistant/components/knx/strings.json @@ -62,8 +62,8 @@ "description": "Please configure the routing options.", "data": { "individual_address": "Individual address", - "multicast_group": "Multicast group used for routing", - "multicast_port": "Multicast port used for routing", + "multicast_group": "Multicast group", + "multicast_port": "Multicast port", "local_ip": "Local IP of Home Assistant" }, "data_description": { @@ -78,8 +78,10 @@ }, "error": { "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", - "invalid_signature": "The password to decrypt the knxkeys file is wrong.", - "file_not_found": "The specified knxkeys file was not found in the path config/.storage/knx/" + "invalid_individual_address": "Value does not match pattern for KNX individual address.\n'area.line.device'", + "invalid_ip_address": "Invalid IPv4 address.", + "invalid_signature": "The password to decrypt the `.knxkeys` file is wrong.", + "file_not_found": "The specified `.knxkeys` file was not found in the path config/.storage/knx/" } }, "options": { @@ -88,8 +90,8 @@ "data": { "connection_type": "KNX Connection Type", "individual_address": "Default individual address", - "multicast_group": "Multicast group", - "multicast_port": "Multicast port", + "multicast_group": "[%key:component::knx::config::step::routing::data::multicast_group%]", + "multicast_port": "[%key:component::knx::config::step::routing::data::multicast_port%]", "local_ip": "Local IP of Home Assistant", "state_updater": "State updater", "rate_limit": "Rate limit" @@ -110,8 +112,8 @@ "host": "[%key:common::config_flow::data::host%]" }, "data_description": { - "port": "Port of the KNX/IP tunneling device.", - "host": "IP address of the KNX/IP tunneling device." + "port": "[%key:component::knx::config::step::manual_tunnel::data_description::port%]", + "host": "[%key:component::knx::config::step::manual_tunnel::data_description::host%]" } } } diff --git a/homeassistant/components/knx/translations/en.json b/homeassistant/components/knx/translations/en.json index 640cb4a5358..538f6b6a5c6 100644 --- a/homeassistant/components/knx/translations/en.json +++ b/homeassistant/components/knx/translations/en.json @@ -6,8 +6,10 @@ }, "error": { "cannot_connect": "Failed to connect", - "file_not_found": "The specified knxkeys file was not found in the path config/.storage/knx/", - "invalid_signature": "The password to decrypt the knxkeys file is wrong." + "file_not_found": "The specified `.knxkeys` file was not found in the path config/.storage/knx/", + "invalid_individual_address": "Value does not match pattern for KNX individual address.\n'..'", + "invalid_ip_address": "Invalid IPv4 address.", + "invalid_signature": "The password to decrypt the `.knxkeys` file is wrong." }, "step": { "manual_tunnel": { @@ -28,8 +30,8 @@ "data": { "individual_address": "Individual address", "local_ip": "Local IP of Home Assistant", - "multicast_group": "Multicast group used for routing", - "multicast_port": "Multicast port used for routing" + "multicast_group": "Multicast group", + "multicast_port": "Multicast port" }, "data_description": { "individual_address": "KNX address to be used by Home Assistant, e.g. `0.0.4`", diff --git a/tests/components/knx/test_config_flow.py b/tests/components/knx/test_config_flow.py index bfdde90cb5b..8d5d57567dc 100644 --- a/tests/components/knx/test_config_flow.py +++ b/tests/components/knx/test_config_flow.py @@ -150,6 +150,26 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None: assert result2["step_id"] == "routing" assert not result2["errors"] + # invalid user input + result_invalid_input = await hass.config_entries.flow.async_configure( + result2["flow_id"], + { + CONF_KNX_MCAST_GRP: "10.1.2.3", # no valid multicast group + CONF_KNX_MCAST_PORT: 3675, + CONF_KNX_INDIVIDUAL_ADDRESS: "not_a_valid_address", + CONF_KNX_LOCAL_IP: "no_local_ip", + }, + ) + await hass.async_block_till_done() + assert result_invalid_input["type"] == RESULT_TYPE_FORM + assert result_invalid_input["step_id"] == "routing" + assert result_invalid_input["errors"] == { + CONF_KNX_MCAST_GRP: "invalid_ip_address", + CONF_KNX_INDIVIDUAL_ADDRESS: "invalid_individual_address", + CONF_KNX_LOCAL_IP: "invalid_ip_address", + } + + # valid user input with patch( "homeassistant.components.knx.async_setup_entry", return_value=True, @@ -297,6 +317,36 @@ async def test_tunneling_setup_for_local_ip(hass: HomeAssistant) -> None: assert result2["step_id"] == "manual_tunnel" assert not result2["errors"] + # invalid host ip address + result_invalid_host = await hass.config_entries.flow.async_configure( + result2["flow_id"], + { + CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP, + CONF_HOST: DEFAULT_MCAST_GRP, # multicast addresses are invalid + CONF_PORT: 3675, + CONF_KNX_LOCAL_IP: "192.168.1.112", + }, + ) + await hass.async_block_till_done() + assert result_invalid_host["type"] == RESULT_TYPE_FORM + assert result_invalid_host["step_id"] == "manual_tunnel" + assert result_invalid_host["errors"] == {CONF_HOST: "invalid_ip_address"} + # invalid local ip address + result_invalid_local = await hass.config_entries.flow.async_configure( + result2["flow_id"], + { + CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP, + CONF_HOST: "192.168.0.2", + CONF_PORT: 3675, + CONF_KNX_LOCAL_IP: "asdf", + }, + ) + await hass.async_block_till_done() + assert result_invalid_local["type"] == RESULT_TYPE_FORM + assert result_invalid_local["step_id"] == "manual_tunnel" + assert result_invalid_local["errors"] == {CONF_KNX_LOCAL_IP: "invalid_ip_address"} + + # valid user input with patch( "homeassistant.components.knx.async_setup_entry", return_value=True, @@ -584,7 +634,7 @@ async def test_configure_secure_knxkeys_file_not_found(hass: HomeAssistant): await hass.async_block_till_done() assert secure_knxkeys["type"] == RESULT_TYPE_FORM assert secure_knxkeys["errors"] - assert secure_knxkeys["errors"]["base"] == "file_not_found" + assert secure_knxkeys["errors"][CONF_KNX_KNXKEY_FILENAME] == "file_not_found" async def test_configure_secure_knxkeys_invalid_signature(hass: HomeAssistant): @@ -613,7 +663,7 @@ async def test_configure_secure_knxkeys_invalid_signature(hass: HomeAssistant): await hass.async_block_till_done() assert secure_knxkeys["type"] == RESULT_TYPE_FORM assert secure_knxkeys["errors"] - assert secure_knxkeys["errors"]["base"] == "invalid_signature" + assert secure_knxkeys["errors"][CONF_KNX_KNXKEY_PASSWORD] == "invalid_signature" async def test_options_flow(