diff --git a/homeassistant/components/knx/config_flow.py b/homeassistant/components/knx/config_flow.py index d6516d1d4ef..3d046ecaeec 100644 --- a/homeassistant/components/knx/config_flow.py +++ b/homeassistant/components/knx/config_flow.py @@ -1,6 +1,7 @@ """Config flow for KNX.""" from __future__ import annotations +from abc import ABC, abstractmethod from typing import Any, Final import voluptuous as vol @@ -10,13 +11,13 @@ from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT from xknx.io.gateway_scanner import GatewayDescriptor, GatewayScanner from xknx.secure import load_key_ring -from homeassistant import config_entries -from homeassistant.config_entries import ConfigEntry, OptionsFlow +from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow from homeassistant.const import CONF_HOST, CONF_PORT from homeassistant.core import callback -from homeassistant.data_entry_flow import FlowResult +from homeassistant.data_entry_flow import FlowHandler, FlowResult from homeassistant.helpers import selector from homeassistant.helpers.storage import STORAGE_DIR +from homeassistant.helpers.typing import UNDEFINED from .const import ( CONF_KNX_AUTOMATIC, @@ -47,21 +48,25 @@ 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( individual_address=XKNX.DEFAULT_ADDRESS, + local_ip=None, multicast_group=DEFAULT_MCAST_GRP, multicast_port=DEFAULT_MCAST_PORT, - state_updater=CONF_KNX_DEFAULT_STATE_UPDATER, rate_limit=CONF_KNX_DEFAULT_RATE_LIMIT, + route_back=False, + state_updater=CONF_KNX_DEFAULT_STATE_UPDATER, ) CONF_KNX_TUNNELING_TYPE: Final = "tunneling_type" -CONF_KNX_LABEL_TUNNELING_TCP: Final = "TCP" -CONF_KNX_LABEL_TUNNELING_TCP_SECURE: Final = "TCP with IP Secure" -CONF_KNX_LABEL_TUNNELING_UDP: Final = "UDP" -CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK: Final = "UDP with route back / NAT mode" +CONF_KNX_TUNNELING_TYPE_LABELS: Final = { + CONF_KNX_TUNNELING: "UDP (Tunnelling v1)", + CONF_KNX_TUNNELING_TCP: "TCP (Tunnelling v2)", + CONF_KNX_TUNNELING_TCP_SECURE: "Secure Tunnelling (TCP)", +} + +OPTION_MANUAL_TUNNEL: Final = "Manual" _IA_SELECTOR = selector.TextSelector() _IP_SELECTOR = selector.TextSelector() @@ -75,88 +80,111 @@ _PORT_SELECTOR = vol.All( ) -class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): - """Handle a KNX config flow.""" +class KNXCommonFlow(ABC, FlowHandler): + """Base class for KNX flows.""" - VERSION = 1 + def __init__(self, initial_data: KNXConfigEntryData) -> None: + """Initialize KNXCommonFlow.""" + self.initial_data = initial_data + self._found_gateways: list[GatewayDescriptor] = [] + self._found_tunnels: list[GatewayDescriptor] = [] + self._selected_tunnel: GatewayDescriptor | None = None + self._tunneling_config: KNXConfigEntryData | None = None - _found_tunnels: list[GatewayDescriptor] - _selected_tunnel: GatewayDescriptor | None - _tunneling_config: KNXConfigEntryData | None + @abstractmethod + def finish_flow(self, new_entry_data: KNXConfigEntryData, title: str) -> FlowResult: + """Finish the flow.""" - @staticmethod - @callback - def async_get_options_flow(config_entry: ConfigEntry) -> KNXOptionsFlowHandler: - """Get the options flow for this handler.""" - return KNXOptionsFlowHandler(config_entry) - - async def async_step_user(self, user_input: dict | None = None) -> FlowResult: - """Handle a flow initialized by the user.""" - if self._async_current_entries(): - return self.async_abort(reason="single_instance_allowed") - - self._found_tunnels = [] - self._selected_tunnel = None - self._tunneling_config = None - return await self.async_step_type() - - async def async_step_type(self, user_input: dict | None = None) -> FlowResult: + async def async_step_connection_type( + self, user_input: dict | None = None + ) -> FlowResult: """Handle connection type configuration.""" if user_input is not None: connection_type = user_input[CONF_KNX_CONNECTION_TYPE] - if connection_type == CONF_KNX_AUTOMATIC: - entry_data = DEFAULT_ENTRY_DATA | KNXConfigEntryData( - connection_type=CONF_KNX_AUTOMATIC - ) - return self.async_create_entry( - title=CONF_KNX_AUTOMATIC.capitalize(), - data=entry_data, - ) - if connection_type == CONF_KNX_ROUTING: return await self.async_step_routing() - if connection_type == CONF_KNX_TUNNELING and self._found_tunnels: + if connection_type == CONF_KNX_TUNNELING: + self._found_tunnels = [ + gateway + for gateway in self._found_gateways + if gateway.supports_tunnelling + ] + self._found_tunnels.sort( + key=lambda tunnel: tunnel.individual_address.raw + if tunnel.individual_address + else 0 + ) return await self.async_step_tunnel() - return await self.async_step_manual_tunnel() + # Automatic connection type + entry_data = KNXConfigEntryData(connection_type=CONF_KNX_AUTOMATIC) + return self.finish_flow( + new_entry_data=entry_data, + title=CONF_KNX_AUTOMATIC.capitalize(), + ) supported_connection_types = { CONF_KNX_TUNNELING: CONF_KNX_TUNNELING.capitalize(), CONF_KNX_ROUTING: CONF_KNX_ROUTING.capitalize(), } - if gateways := await scan_for_gateways(): + self._found_gateways = await scan_for_gateways() + if self._found_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 - ] 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)) + return self.async_show_form( + step_id="connection_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.""" if user_input is not None: + if user_input[CONF_KNX_GATEWAY] == OPTION_MANUAL_TUNNEL: + if self._found_tunnels: + self._selected_tunnel = self._found_tunnels[0] + return await self.async_step_manual_tunnel() + self._selected_tunnel = next( tunnel for tunnel in self._found_tunnels if user_input[CONF_KNX_GATEWAY] == str(tunnel) ) - return await self.async_step_manual_tunnel() + connection_type = ( + CONF_KNX_TUNNELING_TCP_SECURE + if self._selected_tunnel.tunnelling_requires_secure + else CONF_KNX_TUNNELING_TCP + if self._selected_tunnel.supports_tunnelling_tcp + else CONF_KNX_TUNNELING + ) + self._tunneling_config = KNXConfigEntryData( + host=self._selected_tunnel.ip_addr, + port=self._selected_tunnel.port, + route_back=False, + connection_type=connection_type, + ) + if connection_type == CONF_KNX_TUNNELING_TCP_SECURE: + return self.async_show_menu( + step_id="secure_tunneling", + menu_options=["secure_knxkeys", "secure_tunnel_manual"], + ) + return self.finish_flow( + new_entry_data=self._tunneling_config, + title=f"Tunneling @ {self._selected_tunnel}", + ) - # skip this step if the user has only one unique gateway. - if len(self._found_tunnels) == 1: - self._selected_tunnel = self._found_tunnels[0] + if not self._found_tunnels: return await self.async_step_manual_tunnel() errors: dict = {} - tunnels_repr = {str(tunnel) for tunnel in self._found_tunnels} - fields = {vol.Required(CONF_KNX_GATEWAY): vol.In(tunnels_repr)} + tunnel_options = [str(tunnel) for tunnel in self._found_tunnels] + tunnel_options.append(OPTION_MANUAL_TUNNEL) + fields = {vol.Required(CONF_KNX_GATEWAY): vol.In(tunnel_options)} return self.async_show_form( step_id="tunnel", data_schema=vol.Schema(fields), errors=errors @@ -182,61 +210,83 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): if not errors: connection_type = user_input[CONF_KNX_TUNNELING_TYPE] - entry_data = DEFAULT_ENTRY_DATA | KNXConfigEntryData( + self._tunneling_config = KNXConfigEntryData( host=_host, port=user_input[CONF_PORT], - route_back=( - connection_type == CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK - ), + route_back=user_input[CONF_KNX_ROUTE_BACK], local_ip=_local_ip, - connection_type=( - CONF_KNX_TUNNELING_TCP - if connection_type == CONF_KNX_LABEL_TUNNELING_TCP - else CONF_KNX_TUNNELING - ), + connection_type=connection_type, ) - if connection_type == CONF_KNX_LABEL_TUNNELING_TCP_SECURE: - self._tunneling_config = entry_data + if connection_type == CONF_KNX_TUNNELING_TCP_SECURE: return self.async_show_menu( step_id="secure_tunneling", - menu_options=["secure_knxkeys", "secure_manual"], + menu_options=["secure_knxkeys", "secure_tunnel_manual"], ) - - return self.async_create_entry( + return self.finish_flow( + new_entry_data=self._tunneling_config, title=f"Tunneling @ {_host}", - data=entry_data, ) - connection_methods: list[str] = [ - CONF_KNX_LABEL_TUNNELING_TCP, - CONF_KNX_LABEL_TUNNELING_UDP, - CONF_KNX_LABEL_TUNNELING_TCP_SECURE, - CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK, - ] - ip_address = "" - port = DEFAULT_MCAST_PORT - if self._selected_tunnel is not None: + _reconfiguring_existing_tunnel = ( + self.initial_data.get(CONF_KNX_CONNECTION_TYPE) + in CONF_KNX_TUNNELING_TYPE_LABELS + ) + if ( # initial attempt on ConfigFlow or coming from automatic / routing + (isinstance(self, ConfigFlow) or not _reconfiguring_existing_tunnel) + and not user_input + and self._selected_tunnel is not None + ): # default to first found tunnel ip_address = self._selected_tunnel.ip_addr port = self._selected_tunnel.port - if not self._selected_tunnel.supports_tunnelling_tcp: - connection_methods.remove(CONF_KNX_LABEL_TUNNELING_TCP) - connection_methods.remove(CONF_KNX_LABEL_TUNNELING_TCP_SECURE) + if self._selected_tunnel.tunnelling_requires_secure: + default_type = CONF_KNX_TUNNELING_TCP_SECURE + elif self._selected_tunnel.supports_tunnelling_tcp: + default_type = CONF_KNX_TUNNELING_TCP + else: + default_type = CONF_KNX_TUNNELING + else: # OptionFlow, no tunnel discovered or user input + ip_address = ( + user_input[CONF_HOST] + if user_input + else self.initial_data.get(CONF_HOST) + ) + port = ( + user_input[CONF_PORT] + if user_input + else self.initial_data.get(CONF_PORT, DEFAULT_MCAST_PORT) + ) + default_type = ( + user_input[CONF_KNX_TUNNELING_TYPE] + if user_input + else self.initial_data[CONF_KNX_CONNECTION_TYPE] + if _reconfiguring_existing_tunnel + else CONF_KNX_TUNNELING + ) + _route_back: bool = self.initial_data.get( + CONF_KNX_ROUTE_BACK, not bool(self._selected_tunnel) + ) fields = { - vol.Required(CONF_KNX_TUNNELING_TYPE): vol.In(connection_methods), + vol.Required(CONF_KNX_TUNNELING_TYPE, default=default_type): vol.In( + CONF_KNX_TUNNELING_TYPE_LABELS + ), vol.Required(CONF_HOST, default=ip_address): _IP_SELECTOR, vol.Required(CONF_PORT, default=port): _PORT_SELECTOR, + vol.Required( + CONF_KNX_ROUTE_BACK, default=_route_back + ): selector.BooleanSelector(), } - if self.show_advanced_options: fields[vol.Optional(CONF_KNX_LOCAL_IP)] = _IP_SELECTOR + if not self._found_tunnels: + errors["base"] = "no_tunnel_discovered" return self.async_show_form( step_id="manual_tunnel", data_schema=vol.Schema(fields), errors=errors ) - async def async_step_secure_manual( + async def async_step_secure_tunnel_manual( self, user_input: dict | None = None ) -> FlowResult: """Configure ip secure manually.""" @@ -250,14 +300,16 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): user_id=user_input[CONF_KNX_SECURE_USER_ID], user_password=user_input[CONF_KNX_SECURE_USER_PASSWORD], ) - - return self.async_create_entry( + return self.finish_flow( + new_entry_data=entry_data, title=f"Secure Tunneling @ {self._tunneling_config[CONF_HOST]}", - data=entry_data, ) fields = { - vol.Required(CONF_KNX_SECURE_USER_ID, default=2): vol.All( + vol.Required( + CONF_KNX_SECURE_USER_ID, + default=self.initial_data.get(CONF_KNX_SECURE_USER_ID, 2), + ): vol.All( selector.NumberSelector( selector.NumberSelectorConfig( min=1, max=127, mode=selector.NumberSelectorMode.BOX @@ -265,16 +317,24 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): ), vol.Coerce(int), ), - vol.Required(CONF_KNX_SECURE_USER_PASSWORD): selector.TextSelector( + vol.Required( + CONF_KNX_SECURE_USER_PASSWORD, + default=self.initial_data.get(CONF_KNX_SECURE_USER_PASSWORD), + ): selector.TextSelector( selector.TextSelectorConfig(type=selector.TextSelectorType.PASSWORD), ), - vol.Required(CONF_KNX_SECURE_DEVICE_AUTHENTICATION): selector.TextSelector( + vol.Required( + CONF_KNX_SECURE_DEVICE_AUTHENTICATION, + default=self.initial_data.get(CONF_KNX_SECURE_DEVICE_AUTHENTICATION), + ): selector.TextSelector( selector.TextSelectorConfig(type=selector.TextSelectorType.PASSWORD), ), } return self.async_show_form( - step_id="secure_manual", data_schema=vol.Schema(fields), errors=errors + step_id="secure_tunnel_manual", + data_schema=vol.Schema(fields), + errors=errors, ) async def async_step_secure_knxkeys( @@ -302,15 +362,20 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): knxkeys_filename=storage_key, knxkeys_password=user_input[CONF_KNX_KNXKEY_PASSWORD], ) - - return self.async_create_entry( + return self.finish_flow( + new_entry_data=entry_data, title=f"Secure Tunneling @ {self._tunneling_config[CONF_HOST]}", - data=entry_data, ) fields = { - vol.Required(CONF_KNX_KNXKEY_FILENAME): selector.TextSelector(), - vol.Required(CONF_KNX_KNXKEY_PASSWORD): selector.TextSelector(), + vol.Required( + CONF_KNX_KNXKEY_FILENAME, + default=self.initial_data.get(CONF_KNX_KNXKEY_FILENAME), + ): selector.TextSelector(), + vol.Required( + CONF_KNX_KNXKEY_PASSWORD, + default=self.initial_data.get(CONF_KNX_KNXKEY_PASSWORD), + ): selector.TextSelector(), } return self.async_show_form( @@ -323,10 +388,17 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): _individual_address = ( user_input[CONF_KNX_INDIVIDUAL_ADDRESS] if user_input - else XKNX.DEFAULT_ADDRESS + else self.initial_data[CONF_KNX_INDIVIDUAL_ADDRESS] ) _multicast_group = ( - user_input[CONF_KNX_MCAST_GRP] if user_input else DEFAULT_MCAST_GRP + user_input[CONF_KNX_MCAST_GRP] + if user_input + else self.initial_data[CONF_KNX_MCAST_GRP] + ) + _multicast_port = ( + user_input[CONF_KNX_MCAST_PORT] + if user_input + else self.initial_data[CONF_KNX_MCAST_PORT] ) if user_input is not None: @@ -345,15 +417,16 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): errors[CONF_KNX_LOCAL_IP] = "invalid_ip_address" if not errors: - entry_data = DEFAULT_ENTRY_DATA | KNXConfigEntryData( + entry_data = KNXConfigEntryData( connection_type=CONF_KNX_ROUTING, individual_address=_individual_address, multicast_group=_multicast_group, - multicast_port=user_input[CONF_KNX_MCAST_PORT], + multicast_port=_multicast_port, local_ip=_local_ip, ) - return self.async_create_entry( - title=CONF_KNX_ROUTING.capitalize(), data=entry_data + return self.finish_flow( + new_entry_data=entry_data, + title=f"Routing as {_individual_address}", ) fields = { @@ -361,101 +434,112 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): CONF_KNX_INDIVIDUAL_ADDRESS, default=_individual_address ): _IA_SELECTOR, vol.Required(CONF_KNX_MCAST_GRP, default=_multicast_group): _IP_SELECTOR, - vol.Required( - CONF_KNX_MCAST_PORT, default=DEFAULT_MCAST_PORT - ): _PORT_SELECTOR, + vol.Required(CONF_KNX_MCAST_PORT, default=_multicast_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 + if not any( + router for router in self._found_gateways if router.supports_routing + ): + errors["base"] = "no_router_discovered" return self.async_show_form( step_id="routing", data_schema=vol.Schema(fields), errors=errors ) -class KNXOptionsFlowHandler(OptionsFlow): +class KNXConfigFlow(KNXCommonFlow, ConfigFlow, domain=DOMAIN): + """Handle a KNX config flow.""" + + VERSION = 1 + + def __init__(self) -> None: + """Initialize KNX options flow.""" + super().__init__(initial_data=DEFAULT_ENTRY_DATA) + + @staticmethod + @callback + def async_get_options_flow(config_entry: ConfigEntry) -> KNXOptionsFlow: + """Get the options flow for this handler.""" + return KNXOptionsFlow(config_entry) + + @callback + def finish_flow(self, new_entry_data: KNXConfigEntryData, title: str) -> FlowResult: + """Create the ConfigEntry.""" + return self.async_create_entry( + title=title, + data=DEFAULT_ENTRY_DATA | new_entry_data, + ) + + async def async_step_user(self, user_input: dict | None = None) -> FlowResult: + """Handle a flow initialized by the user.""" + if self._async_current_entries(): + return self.async_abort(reason="single_instance_allowed") + return await self.async_step_connection_type() + + +class KNXOptionsFlow(KNXCommonFlow, OptionsFlow): """Handle KNX options.""" general_settings: dict - current_config: dict def __init__(self, config_entry: ConfigEntry) -> None: """Initialize KNX options flow.""" self.config_entry = config_entry + super().__init__(initial_data=config_entry.data) # type: ignore[arg-type] + + @callback + def finish_flow( + self, new_entry_data: KNXConfigEntryData, title: str | None + ) -> FlowResult: + """Update the ConfigEntry and finish the flow.""" + new_data = DEFAULT_ENTRY_DATA | self.initial_data | new_entry_data + self.hass.config_entries.async_update_entry( + self.config_entry, + data=new_data, + title=title or UNDEFINED, + ) + return self.async_create_entry(title="", data={}) async def async_step_init( self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Manage KNX options.""" - if user_input is not None: - self.general_settings = user_input - return await self.async_step_tunnel() + return self.async_show_menu( + step_id="options_init", + menu_options=["connection_type", "communication_settings"], + ) - supported_connection_types = [ - CONF_KNX_AUTOMATIC, - CONF_KNX_TUNNELING, - CONF_KNX_ROUTING, - ] - self.current_config = self.config_entry.data # type: ignore[assignment] + async def async_step_communication_settings( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Manage KNX communication settings.""" + if user_input is not None: + return self.finish_flow( + new_entry_data=KNXConfigEntryData( + state_updater=user_input[CONF_KNX_STATE_UPDATER], + rate_limit=user_input[CONF_KNX_RATE_LIMIT], + ), + title=None, + ) data_schema = { vol.Required( - CONF_KNX_CONNECTION_TYPE, - default=( - CONF_KNX_TUNNELING - if self.current_config.get(CONF_KNX_CONNECTION_TYPE) - == CONF_KNX_TUNNELING_TCP - else self.current_config.get(CONF_KNX_CONNECTION_TYPE) - ), - ): vol.In(supported_connection_types), - vol.Required( - CONF_KNX_INDIVIDUAL_ADDRESS, - default=self.current_config[CONF_KNX_INDIVIDUAL_ADDRESS], - ): selector.TextSelector(), - vol.Required( - CONF_KNX_MCAST_GRP, - default=self.current_config.get(CONF_KNX_MCAST_GRP, DEFAULT_MCAST_GRP), - ): _IP_SELECTOR, - vol.Required( - CONF_KNX_MCAST_PORT, - default=self.current_config.get( - CONF_KNX_MCAST_PORT, DEFAULT_MCAST_PORT - ), - ): _PORT_SELECTOR, - } - - if self.show_advanced_options: - local_ip = ( - self.current_config.get(CONF_KNX_LOCAL_IP) - if self.current_config.get(CONF_KNX_LOCAL_IP) is not None - else CONF_DEFAULT_LOCAL_IP - ) - data_schema[ - vol.Required( - CONF_KNX_LOCAL_IP, - default=local_ip, - ) - ] = _IP_SELECTOR - data_schema[ - vol.Required( + CONF_KNX_STATE_UPDATER, + default=self.initial_data.get( CONF_KNX_STATE_UPDATER, - default=self.current_config.get( - CONF_KNX_STATE_UPDATER, - CONF_KNX_DEFAULT_STATE_UPDATER, - ), - ) - ] = selector.BooleanSelector() - data_schema[ - vol.Required( + CONF_KNX_DEFAULT_STATE_UPDATER, + ), + ): selector.BooleanSelector(), + vol.Required( + CONF_KNX_RATE_LIMIT, + default=self.initial_data.get( CONF_KNX_RATE_LIMIT, - default=self.current_config.get( - CONF_KNX_RATE_LIMIT, - CONF_KNX_DEFAULT_RATE_LIMIT, - ), - ) - ] = vol.All( + CONF_KNX_DEFAULT_RATE_LIMIT, + ), + ): vol.All( selector.NumberSelector( selector.NumberSelectorConfig( min=0, @@ -464,96 +548,14 @@ class KNXOptionsFlowHandler(OptionsFlow): ), ), vol.Coerce(int), - ) - + ), + } return self.async_show_form( - step_id="init", + step_id="communication_settings", data_schema=vol.Schema(data_schema), - last_step=self.current_config.get(CONF_KNX_CONNECTION_TYPE) - != CONF_KNX_TUNNELING, + last_step=True, ) - async def async_step_tunnel( - self, user_input: dict[str, Any] | None = None - ) -> FlowResult: - """Manage KNX tunneling options.""" - if ( - self.general_settings.get(CONF_KNX_CONNECTION_TYPE) == CONF_KNX_TUNNELING - and user_input is None - ): - connection_methods: list[str] = [ - CONF_KNX_LABEL_TUNNELING_TCP, - CONF_KNX_LABEL_TUNNELING_UDP, - CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK, - ] - return self.async_show_form( - step_id="tunnel", - data_schema=vol.Schema( - { - vol.Required( - CONF_KNX_TUNNELING_TYPE, - default=get_knx_tunneling_type(self.current_config), - ): vol.In(connection_methods), - vol.Required( - CONF_HOST, default=self.current_config.get(CONF_HOST) - ): _IP_SELECTOR, - vol.Required( - CONF_PORT, default=self.current_config.get(CONF_PORT, 3671) - ): _PORT_SELECTOR, - } - ), - last_step=True, - ) - - _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 | 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"Tunneling @ {entry_data[CONF_HOST]}" - if entry_data[CONF_KNX_CONNECTION_TYPE] == CONF_KNX_TUNNELING_TCP: - entry_title = f"Tunneling @ {entry_data[CONF_HOST]} (TCP)" - - self.hass.config_entries.async_update_entry( - self.config_entry, - data=entry_data, - title=entry_title, - ) - - return self.async_create_entry(title="", data={}) - - -def get_knx_tunneling_type(config_entry_data: dict) -> str: - """Obtain the knx tunneling type based on the data in the config entry data.""" - connection_type = config_entry_data[CONF_KNX_CONNECTION_TYPE] - route_back = config_entry_data.get(CONF_KNX_ROUTE_BACK, False) - if route_back and connection_type == CONF_KNX_TUNNELING: - return CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK - if connection_type == CONF_KNX_TUNNELING_TCP: - return CONF_KNX_LABEL_TUNNELING_TCP - - return CONF_KNX_LABEL_TUNNELING_UDP - async def scan_for_gateways(stop_on_found: int = 0) -> list[GatewayDescriptor]: """Scan for gateways within the network.""" diff --git a/homeassistant/components/knx/strings.json b/homeassistant/components/knx/strings.json index c8161462d66..d87ad6ac177 100644 --- a/homeassistant/components/knx/strings.json +++ b/homeassistant/components/knx/strings.json @@ -1,7 +1,7 @@ { "config": { "step": { - "type": { + "connection_type": { "description": "Please enter the connection type we should use for your KNX connection. \n AUTOMATIC - The integration takes care of the connectivity to your KNX Bus by performing a gateway scan. \n TUNNELING - The integration will connect to your KNX bus via tunneling. \n ROUTING - The integration will connect to your KNX bus via routing.", "data": { "connection_type": "KNX Connection Type" @@ -19,11 +19,13 @@ "tunneling_type": "KNX Tunneling Type", "port": "[%key:common::config_flow::data::port%]", "host": "[%key:common::config_flow::data::host%]", + "route_back": "Route back / NAT mode", "local_ip": "Local IP of Home Assistant" }, "data_description": { "port": "Port of the KNX/IP tunneling device.", "host": "IP address of the KNX/IP tunneling device.", + "route_back": "Enable if your KNXnet/IP tunneling server is behind NAT. Only applies for UDP connections.", "local_ip": "Leave blank to use auto-discovery." } }, @@ -31,7 +33,7 @@ "description": "Select how you want to configure KNX/IP Secure.", "menu_options": { "secure_knxkeys": "Use a `.knxkeys` file containing IP secure keys", - "secure_manual": "Configure IP secure keys manually" + "secure_tunnel_manual": "Configure IP secure keys manually" } }, "secure_knxkeys": { @@ -45,7 +47,7 @@ "knxkeys_password": "This was set when exporting the file from ETS." } }, - "secure_manual": { + "secure_tunnel_manual": { "description": "Please enter your IP secure information.", "data": { "user_id": "User ID", @@ -81,41 +83,110 @@ "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/" + "file_not_found": "The specified `.knxkeys` file was not found in the path config/.storage/knx/", + "no_router_discovered": "No KNXnet/IP router was discovered on the network.", + "no_tunnel_discovered": "Could not find a KNX tunneling server on your network." } }, "options": { "step": { - "init": { + "options_init": { + "menu_options": { + "connection_type": "Configure KNX interface", + "communication_settings": "Communication settings" + } + }, + "communication_settings": { "data": { - "connection_type": "KNX Connection Type", - "individual_address": "Default individual address", - "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" }, "data_description": { - "individual_address": "KNX address to be used by Home Assistant, e.g. `0.0.4`", - "multicast_group": "Used for routing and discovery. Default: `224.0.23.12`", - "multicast_port": "Used for routing and discovery. Default: `3671`", - "local_ip": "Use `0.0.0.0` for auto-discovery.", "state_updater": "Set default for reading states from the KNX Bus. When disabled, Home Assistant will not actively retrieve entity states from the KNX Bus. Can be overridden by `sync_state` entity options.", - "rate_limit": "Maximum outgoing telegrams per second.\nRecommended: 20 to 40" + "rate_limit": "Maximum outgoing telegrams per second.\n`0` to disable limit. Recommended: 0 or 20 to 40" + } + }, + "connection_type": { + "description": "Please enter the connection type we should use for your KNX connection. \n AUTOMATIC - The integration takes care of the connectivity to your KNX Bus by performing a gateway scan. \n TUNNELING - The integration will connect to your KNX bus via tunneling. \n ROUTING - The integration will connect to your KNX bus via routing.", + "data": { + "connection_type": "KNX Connection Type" } }, "tunnel": { + "description": "[%key:component::knx::config::step::tunnel::description%]", "data": { - "tunneling_type": "KNX Tunneling Type", + "gateway": "[%key:component::knx::config::step::tunnel::data::gateway%]" + } + }, + "manual_tunnel": { + "description": "[%key:component::knx::config::step::manual_tunnel::description%]", + "data": { + "tunneling_type": "[%key:component::knx::config::step::manual_tunnel::data::tunneling_type%]", "port": "[%key:common::config_flow::data::port%]", - "host": "[%key:common::config_flow::data::host%]" + "host": "[%key:common::config_flow::data::host%]", + "route_back": "[%key:component::knx::config::step::manual_tunnel::data::route_back%]", + "local_ip": "[%key:component::knx::config::step::manual_tunnel::data::local_ip%]" }, "data_description": { "port": "[%key:component::knx::config::step::manual_tunnel::data_description::port%]", - "host": "[%key:component::knx::config::step::manual_tunnel::data_description::host%]" + "host": "[%key:component::knx::config::step::manual_tunnel::data_description::host%]", + "route_back": "[%key:component::knx::config::step::manual_tunnel::data_description::route_back%]", + "local_ip": "[%key:component::knx::config::step::manual_tunnel::data_description::local_ip%]" + } + }, + "secure_tunneling": { + "description": "[%key:component::knx::config::step::secure_tunneling::description%]", + "menu_options": { + "secure_knxkeys": "[%key:component::knx::config::step::secure_tunneling::menu_options::secure_knxkeys%]", + "secure_tunnel_manual": "[%key:component::knx::config::step::secure_tunneling::menu_options::secure_tunnel_manual%]" + } + }, + "secure_knxkeys": { + "description": "[%key:component::knx::config::step::secure_knxkeys::description%]", + "data": { + "knxkeys_filename": "[%key:component::knx::config::step::secure_knxkeys::data::knxkeys_filename%]", + "knxkeys_password": "[%key:component::knx::config::step::secure_knxkeys::data::knxkeys_password%]" + }, + "data_description": { + "knxkeys_filename": "[%key:component::knx::config::step::secure_knxkeys::data_description::knxkeys_filename%]", + "knxkeys_password": "[%key:component::knx::config::step::secure_knxkeys::data_description::knxkeys_password%]" + } + }, + "secure_tunnel_manual": { + "description": "[%key:component::knx::config::step::secure_tunnel_manual::description%]", + "data": { + "user_id": "[%key:component::knx::config::step::secure_tunnel_manual::data::user_id%]", + "user_password": "[%key:component::knx::config::step::secure_tunnel_manual::data::user_password%]", + "device_authentication": "[%key:component::knx::config::step::secure_tunnel_manual::data::device_authentication%]" + }, + "data_description": { + "user_id": "[%key:component::knx::config::step::secure_tunnel_manual::data_description::user_id%]", + "user_password": "[%key:component::knx::config::step::secure_tunnel_manual::data_description::user_password%]", + "device_authentication": "[%key:component::knx::config::step::secure_tunnel_manual::data_description::device_authentication%]" + } + }, + "routing": { + "description": "[%key:component::knx::config::step::routing::description%]", + "data": { + "individual_address": "[%key:component::knx::config::step::routing::data::individual_address%]", + "multicast_group": "[%key:component::knx::config::step::routing::data::multicast_group%]", + "multicast_port": "[%key:component::knx::config::step::routing::data::multicast_port%]", + "local_ip": "[%key:component::knx::config::step::routing::data::local_ip%]" + }, + "data_description": { + "individual_address": "[%key:component::knx::config::step::routing::data_description::individual_address%]", + "local_ip": "[%key:component::knx::config::step::routing::data_description::local_ip%]" } } + }, + "error": { + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "invalid_individual_address": "[%key:component::knx::config::error::invalid_individual_address%]", + "invalid_ip_address": "[%key:component::knx::config::error::invalid_ip_address%]", + "invalid_signature": "[%key:component::knx::config::error::invalid_signature%]", + "file_not_found": "[%key:component::knx::config::error::file_not_found%]", + "no_router_discovered": "[%key:component::knx::config::error::no_router_discovered%]", + "no_tunnel_discovered": "[%key:component::knx::config::error::no_tunnel_discovered%]" } } } diff --git a/homeassistant/components/knx/translations/en.json b/homeassistant/components/knx/translations/en.json index 6dffe059b2a..c45c98b070a 100644 --- a/homeassistant/components/knx/translations/en.json +++ b/homeassistant/components/knx/translations/en.json @@ -9,20 +9,30 @@ "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." + "invalid_signature": "The password to decrypt the `.knxkeys` file is wrong.", + "no_router_discovered": "No KNXnet/IP router was discovered on the network.", + "no_tunnel_discovered": "Could not find a KNX tunneling server on your network." }, "step": { + "connection_type": { + "data": { + "connection_type": "KNX Connection Type" + }, + "description": "Please enter the connection type we should use for your KNX connection. \n AUTOMATIC - The integration takes care of the connectivity to your KNX Bus by performing a gateway scan. \n TUNNELING - The integration will connect to your KNX bus via tunneling. \n ROUTING - The integration will connect to your KNX bus via routing." + }, "manual_tunnel": { "data": { "host": "Host", "local_ip": "Local IP of Home Assistant", "port": "Port", + "route_back": "Route back / NAT mode", "tunneling_type": "KNX Tunneling Type" }, "data_description": { "host": "IP address of the KNX/IP tunneling device.", "local_ip": "Leave blank to use auto-discovery.", - "port": "Port of the KNX/IP tunneling device." + "port": "Port of the KNX/IP tunneling device.", + "route_back": "Enable if your KNXnet/IP tunneling server is behind NAT. Only applies for UDP connections." }, "description": "Please enter the connection information of your tunneling device." }, @@ -50,7 +60,7 @@ }, "description": "Please enter the information for your `.knxkeys` file." }, - "secure_manual": { + "secure_tunnel_manual": { "data": { "device_authentication": "Device authentication password", "user_id": "User ID", @@ -67,7 +77,7 @@ "description": "Select how you want to configure KNX/IP Secure.", "menu_options": { "secure_knxkeys": "Use a `.knxkeys` file containing IP secure keys", - "secure_manual": "Configure IP secure keys manually" + "secure_tunnel_manual": "Configure IP secure keys manually" } }, "tunnel": { @@ -75,46 +85,107 @@ "gateway": "KNX Tunnel Connection" }, "description": "Please select a gateway from the list." - }, - "type": { - "data": { - "connection_type": "KNX Connection Type" - }, - "description": "Please enter the connection type we should use for your KNX connection. \n AUTOMATIC - The integration takes care of the connectivity to your KNX Bus by performing a gateway scan. \n TUNNELING - The integration will connect to your KNX bus via tunneling. \n ROUTING - The integration will connect to your KNX bus via routing." } } }, "options": { + "error": { + "cannot_connect": "Failed to connect", + "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.", + "no_router_discovered": "No KNXnet/IP router was discovered on the network.", + "no_tunnel_discovered": "Could not find a KNX tunneling server on your network." + }, "step": { - "init": { + "communication_settings": { "data": { - "connection_type": "KNX Connection Type", - "individual_address": "Default individual address", - "local_ip": "Local IP of Home Assistant", - "multicast_group": "Multicast group", - "multicast_port": "Multicast port", "rate_limit": "Rate limit", "state_updater": "State updater" }, "data_description": { - "individual_address": "KNX address to be used by Home Assistant, e.g. `0.0.4`", - "local_ip": "Use `0.0.0.0` for auto-discovery.", - "multicast_group": "Used for routing and discovery. Default: `224.0.23.12`", - "multicast_port": "Used for routing and discovery. Default: `3671`", - "rate_limit": "Maximum outgoing telegrams per second.\nRecommended: 20 to 40", + "rate_limit": "Maximum outgoing telegrams per second.\n`0` to disable limit. Recommended: 0 or 20 to 40", "state_updater": "Set default for reading states from the KNX Bus. When disabled, Home Assistant will not actively retrieve entity states from the KNX Bus. Can be overridden by `sync_state` entity options." } }, - "tunnel": { + "connection_type": { + "data": { + "connection_type": "KNX Connection Type" + }, + "description": "Please enter the connection type we should use for your KNX connection. \n AUTOMATIC - The integration takes care of the connectivity to your KNX Bus by performing a gateway scan. \n TUNNELING - The integration will connect to your KNX bus via tunneling. \n ROUTING - The integration will connect to your KNX bus via routing." + }, + "manual_tunnel": { "data": { "host": "Host", + "local_ip": "Local IP of Home Assistant", "port": "Port", + "route_back": "Route back / NAT mode", "tunneling_type": "KNX Tunneling Type" }, "data_description": { "host": "IP address of the KNX/IP tunneling device.", - "port": "Port of the KNX/IP tunneling device." + "local_ip": "Leave blank to use auto-discovery.", + "port": "Port of the KNX/IP tunneling device.", + "route_back": "Enable if your KNXnet/IP tunneling server is behind NAT. Only applies for UDP connections." + }, + "description": "Please enter the connection information of your tunneling device." + }, + "options_init": { + "menu_options": { + "communication_settings": "Communication settings", + "connection_type": "Configure KNX interface" } + }, + "routing": { + "data": { + "individual_address": "Individual address", + "local_ip": "Local IP of Home Assistant", + "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`", + "local_ip": "Leave blank to use auto-discovery." + }, + "description": "Please configure the routing options." + }, + "secure_knxkeys": { + "data": { + "knxkeys_filename": "The filename of your `.knxkeys` file (including extension)", + "knxkeys_password": "The password to decrypt the `.knxkeys` file" + }, + "data_description": { + "knxkeys_filename": "The file is expected to be found in your config directory in `.storage/knx/`.\nIn Home Assistant OS this would be `/config/.storage/knx/`\nExample: `my_project.knxkeys`", + "knxkeys_password": "This was set when exporting the file from ETS." + }, + "description": "Please enter the information for your `.knxkeys` file." + }, + "secure_tunnel_manual": { + "data": { + "device_authentication": "Device authentication password", + "user_id": "User ID", + "user_password": "User password" + }, + "data_description": { + "device_authentication": "This is set in the 'IP' panel of the interface in ETS.", + "user_id": "This is often tunnel number +1. So 'Tunnel 2' would have User-ID '3'.", + "user_password": "Password for the specific tunnel connection set in the 'Properties' panel of the tunnel in ETS." + }, + "description": "Please enter your IP secure information." + }, + "secure_tunneling": { + "description": "Select how you want to configure KNX/IP Secure.", + "menu_options": { + "secure_knxkeys": "Use a `.knxkeys` file containing IP secure keys", + "secure_tunnel_manual": "Configure IP secure keys manually" + } + }, + "tunnel": { + "data": { + "gateway": "KNX Tunnel Connection" + }, + "description": "Please select a gateway from the list." } } } diff --git a/tests/components/knx/test_config_flow.py b/tests/components/knx/test_config_flow.py index 30b8aa537a6..fdef15ed217 100644 --- a/tests/components/knx/test_config_flow.py +++ b/tests/components/knx/test_config_flow.py @@ -3,24 +3,20 @@ from unittest.mock import patch import pytest from xknx.exceptions.exception import InvalidSignature -from xknx.io import DEFAULT_MCAST_GRP +from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT from xknx.io.gateway_scanner import GatewayDescriptor from homeassistant import config_entries from homeassistant.components.knx.config_flow import ( - CONF_DEFAULT_LOCAL_IP, CONF_KNX_GATEWAY, - CONF_KNX_LABEL_TUNNELING_TCP, - CONF_KNX_LABEL_TUNNELING_TCP_SECURE, - CONF_KNX_LABEL_TUNNELING_UDP, - CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK, CONF_KNX_TUNNELING_TYPE, DEFAULT_ENTRY_DATA, - get_knx_tunneling_type, + OPTION_MANUAL_TUNNEL, ) from homeassistant.components.knx.const import ( CONF_KNX_AUTOMATIC, CONF_KNX_CONNECTION_TYPE, + CONF_KNX_DEFAULT_STATE_UPDATER, CONF_KNX_INDIVIDUAL_ADDRESS, CONF_KNX_KNXKEY_FILENAME, CONF_KNX_KNXKEY_PASSWORD, @@ -41,16 +37,19 @@ from homeassistant.components.knx.const import ( ) from homeassistant.const import CONF_HOST, CONF_PORT from homeassistant.core import HomeAssistant -from homeassistant.data_entry_flow import FlowResultType +from homeassistant.data_entry_flow import FlowResult, FlowResultType from tests.common import MockConfigEntry def _gateway_descriptor( - ip: str, port: int, supports_tunnelling_tcp: bool = False + ip: str, + port: int, + supports_tunnelling_tcp: bool = False, + requires_secure: bool = False, ) -> GatewayDescriptor: """Get mock gw descriptor.""" - return GatewayDescriptor( + descriptor = GatewayDescriptor( name="Test", ip_addr=ip, port=port, @@ -60,6 +59,9 @@ def _gateway_descriptor( supports_tunnelling=True, supports_tunnelling_tcp=supports_tunnelling_tcp, ) + descriptor.tunnelling_requires_secure = requires_secure + descriptor.routing_requires_secure = requires_secure + return descriptor async def test_user_single_instance(hass): @@ -92,7 +94,7 @@ async def test_routing_setup(hass: HomeAssistant) -> None: await hass.async_block_till_done() assert result2["type"] == FlowResultType.FORM assert result2["step_id"] == "routing" - assert not result2["errors"] + assert result2["errors"] == {"base": "no_router_discovered"} with patch( "homeassistant.components.knx.async_setup_entry", @@ -108,7 +110,7 @@ async def test_routing_setup(hass: HomeAssistant) -> None: ) await hass.async_block_till_done() assert result3["type"] == FlowResultType.CREATE_ENTRY - assert result3["title"] == CONF_KNX_ROUTING.capitalize() + assert result3["title"] == "Routing as 1.1.110" assert result3["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, @@ -144,7 +146,7 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None: await hass.async_block_till_done() assert result2["type"] == FlowResultType.FORM assert result2["step_id"] == "routing" - assert not result2["errors"] + assert result2["errors"] == {"base": "no_router_discovered"} # invalid user input result_invalid_input = await hass.config_entries.flow.async_configure( @@ -163,6 +165,7 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None: CONF_KNX_MCAST_GRP: "invalid_ip_address", CONF_KNX_INDIVIDUAL_ADDRESS: "invalid_individual_address", CONF_KNX_LOCAL_IP: "invalid_ip_address", + "base": "no_router_discovered", } # valid user input @@ -181,7 +184,7 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None: ) await hass.async_block_till_done() assert result3["type"] == FlowResultType.CREATE_ENTRY - assert result3["title"] == CONF_KNX_ROUTING.capitalize() + assert result3["title"] == "Routing as 1.1.110" assert result3["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, @@ -199,9 +202,10 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None: [ ( { - CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP, + CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING, CONF_HOST: "192.168.0.1", CONF_PORT: 3675, + CONF_KNX_ROUTE_BACK: False, }, { **DEFAULT_ENTRY_DATA, @@ -215,9 +219,10 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None: ), ( { - CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_TCP, + CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP, CONF_HOST: "192.168.0.1", CONF_PORT: 3675, + CONF_KNX_ROUTE_BACK: False, }, { **DEFAULT_ENTRY_DATA, @@ -231,9 +236,10 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None: ), ( { - CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK, + CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING, CONF_HOST: "192.168.0.1", CONF_PORT: 3675, + CONF_KNX_ROUTE_BACK: True, }, { **DEFAULT_ENTRY_DATA, @@ -247,13 +253,12 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None: ), ], ) -async def test_tunneling_setup( +async def test_tunneling_setup_manual( hass: HomeAssistant, user_input, config_entry_data ) -> None: - """Test tunneling if only one gateway is found.""" - gateway = _gateway_descriptor("192.168.0.1", 3675, True) + """Test tunneling if no gateway was found found (or `manual` option was chosen).""" with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways: - gateways.return_value = [gateway] + gateways.return_value = [] result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER} ) @@ -269,7 +274,7 @@ async def test_tunneling_setup( await hass.async_block_till_done() assert result2["type"] == FlowResultType.FORM assert result2["step_id"] == "manual_tunnel" - assert not result2["errors"] + assert result2["errors"] == {"base": "no_tunnel_discovered"} with patch( "homeassistant.components.knx.async_setup_entry", @@ -289,9 +294,8 @@ async def test_tunneling_setup( async def test_tunneling_setup_for_local_ip(hass: HomeAssistant) -> None: """Test tunneling if only one gateway is found.""" - gateway = _gateway_descriptor("192.168.0.2", 3675) with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways: - gateways.return_value = [gateway] + gateways.return_value = [] result = await hass.config_entries.flow.async_init( DOMAIN, context={ @@ -311,13 +315,13 @@ async def test_tunneling_setup_for_local_ip(hass: HomeAssistant) -> None: await hass.async_block_till_done() assert result2["type"] == FlowResultType.FORM assert result2["step_id"] == "manual_tunnel" - assert not result2["errors"] + assert result2["errors"] == {"base": "no_tunnel_discovered"} # 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_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING, CONF_HOST: DEFAULT_MCAST_GRP, # multicast addresses are invalid CONF_PORT: 3675, CONF_KNX_LOCAL_IP: "192.168.1.112", @@ -326,12 +330,15 @@ async def test_tunneling_setup_for_local_ip(hass: HomeAssistant) -> None: await hass.async_block_till_done() assert result_invalid_host["type"] == FlowResultType.FORM assert result_invalid_host["step_id"] == "manual_tunnel" - assert result_invalid_host["errors"] == {CONF_HOST: "invalid_ip_address"} + assert result_invalid_host["errors"] == { + CONF_HOST: "invalid_ip_address", + "base": "no_tunnel_discovered", + } # 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_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING, CONF_HOST: "192.168.0.2", CONF_PORT: 3675, CONF_KNX_LOCAL_IP: "asdf", @@ -340,7 +347,10 @@ async def test_tunneling_setup_for_local_ip(hass: HomeAssistant) -> None: await hass.async_block_till_done() assert result_invalid_local["type"] == FlowResultType.FORM assert result_invalid_local["step_id"] == "manual_tunnel" - assert result_invalid_local["errors"] == {CONF_KNX_LOCAL_IP: "invalid_ip_address"} + assert result_invalid_local["errors"] == { + CONF_KNX_LOCAL_IP: "invalid_ip_address", + "base": "no_tunnel_discovered", + } # valid user input with patch( @@ -350,7 +360,7 @@ async def test_tunneling_setup_for_local_ip(hass: HomeAssistant) -> None: result3 = await hass.config_entries.flow.async_configure( result2["flow_id"], { - CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP, + CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING, CONF_HOST: "192.168.0.2", CONF_PORT: 3675, CONF_KNX_LOCAL_IP: "192.168.1.112", @@ -395,29 +405,17 @@ async def test_tunneling_setup_for_multiple_found_gateways(hass: HomeAssistant) assert tunnel_flow["step_id"] == "tunnel" assert not tunnel_flow["errors"] - manual_tunnel = await hass.config_entries.flow.async_configure( - tunnel_flow["flow_id"], - {CONF_KNX_GATEWAY: str(gateway)}, - ) - await hass.async_block_till_done() - assert manual_tunnel["type"] == FlowResultType.FORM - assert manual_tunnel["step_id"] == "manual_tunnel" - with patch( "homeassistant.components.knx.async_setup_entry", return_value=True, ) as mock_setup_entry: - manual_tunnel_flow = await hass.config_entries.flow.async_configure( - manual_tunnel["flow_id"], - { - CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP, - CONF_HOST: "192.168.0.1", - CONF_PORT: 3675, - }, + result = await hass.config_entries.flow.async_configure( + tunnel_flow["flow_id"], + {CONF_KNX_GATEWAY: str(gateway)}, ) await hass.async_block_till_done() - assert manual_tunnel_flow["type"] == FlowResultType.CREATE_ENTRY - assert manual_tunnel_flow["data"] == { + assert result["type"] == FlowResultType.CREATE_ENTRY + assert result["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_HOST: "192.168.0.1", @@ -430,10 +428,22 @@ async def test_tunneling_setup_for_multiple_found_gateways(hass: HomeAssistant) assert len(mock_setup_entry.mock_calls) == 1 -async def test_manual_tunnel_step_when_no_gateway(hass: HomeAssistant) -> None: - """Test manual tunnel if no gateway is found and tunneling is selected.""" +@pytest.mark.parametrize( + "gateway", + [ + _gateway_descriptor("192.168.0.1", 3675), + _gateway_descriptor("192.168.0.1", 3675, supports_tunnelling_tcp=True), + _gateway_descriptor( + "192.168.0.1", 3675, supports_tunnelling_tcp=True, requires_secure=True + ), + ], +) +async def test_manual_tunnel_step_with_found_gateway( + hass: HomeAssistant, gateway +) -> None: + """Test manual tunnel if gateway was found and tunneling is selected.""" with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways: - gateways.return_value = [] + gateways.return_value = [gateway] result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER} ) @@ -448,9 +458,20 @@ async def test_manual_tunnel_step_when_no_gateway(hass: HomeAssistant) -> None: ) await hass.async_block_till_done() assert tunnel_flow["type"] == FlowResultType.FORM - assert tunnel_flow["step_id"] == "manual_tunnel" + assert tunnel_flow["step_id"] == "tunnel" assert not tunnel_flow["errors"] + manual_tunnel_flow = await hass.config_entries.flow.async_configure( + tunnel_flow["flow_id"], + { + CONF_KNX_GATEWAY: OPTION_MANUAL_TUNNEL, + }, + ) + await hass.async_block_till_done() + assert manual_tunnel_flow["type"] == FlowResultType.FORM + assert manual_tunnel_flow["step_id"] == "manual_tunnel" + assert not manual_tunnel_flow["errors"] + async def test_form_with_automatic_connection_handling(hass: HomeAssistant) -> None: """Test we get the form.""" @@ -484,9 +505,14 @@ async def test_form_with_automatic_connection_handling(hass: HomeAssistant) -> N assert len(mock_setup_entry.mock_calls) == 1 -async def _get_menu_step(hass: HomeAssistant) -> None: - """Test ip secure manuel.""" - gateway = _gateway_descriptor("192.168.0.1", 3675, True) +async def _get_menu_step(hass: HomeAssistant) -> FlowResult: + """Return flow in secure_tunnellinn menu step.""" + gateway = _gateway_descriptor( + "192.168.0.1", + 3675, + supports_tunnelling_tcp=True, + requires_secure=True, + ) with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways: gateways.return_value = [gateway] result = await hass.config_entries.flow.async_init( @@ -503,16 +529,12 @@ async def _get_menu_step(hass: HomeAssistant) -> None: ) await hass.async_block_till_done() assert result2["type"] == FlowResultType.FORM - assert result2["step_id"] == "manual_tunnel" + assert result2["step_id"] == "tunnel" assert not result2["errors"] result3 = await hass.config_entries.flow.async_configure( result2["flow_id"], - { - CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_TCP_SECURE, - CONF_HOST: "192.168.0.1", - CONF_PORT: 3675, - }, + {CONF_KNX_GATEWAY: str(gateway)}, ) await hass.async_block_till_done() assert result3["type"] == FlowResultType.MENU @@ -520,23 +542,72 @@ async def _get_menu_step(hass: HomeAssistant) -> None: return result3 -async def test_configure_secure_manual(hass: HomeAssistant): - """Test configure secure manual.""" +async def test_get_secure_menu_step_manual_tunnelling( + hass: HomeAssistant, +): + """Test flow reaches secure_tunnellinn menu step from manual tunnelling configuration.""" + gateway = _gateway_descriptor( + "192.168.0.1", + 3675, + supports_tunnelling_tcp=True, + requires_secure=True, + ) + with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways: + gateways.return_value = [gateway] + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] == FlowResultType.FORM + assert not result["errors"] + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, + }, + ) + await hass.async_block_till_done() + assert result2["type"] == FlowResultType.FORM + assert result2["step_id"] == "tunnel" + assert not result2["errors"] + + manual_tunnel_flow = await hass.config_entries.flow.async_configure( + result2["flow_id"], + { + CONF_KNX_GATEWAY: OPTION_MANUAL_TUNNEL, + }, + ) + + result3 = await hass.config_entries.flow.async_configure( + manual_tunnel_flow["flow_id"], + { + CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3675, + }, + ) + await hass.async_block_till_done() + assert result3["type"] == FlowResultType.MENU + assert result3["step_id"] == "secure_tunneling" + + +async def test_configure_secure_tunnel_manual(hass: HomeAssistant): + """Test configure tunnelling secure keys manually.""" menu_step = await _get_menu_step(hass) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], - {"next_step_id": "secure_manual"}, + {"next_step_id": "secure_tunnel_manual"}, ) assert result["type"] == FlowResultType.FORM - assert result["step_id"] == "secure_manual" + assert result["step_id"] == "secure_tunnel_manual" assert not result["errors"] with patch( "homeassistant.components.knx.async_setup_entry", return_value=True, ) as mock_setup_entry: - secure_manual = await hass.config_entries.flow.async_configure( + secure_tunnel_manual = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_SECURE_USER_ID: 2, @@ -545,8 +616,8 @@ async def test_configure_secure_manual(hass: HomeAssistant): }, ) await hass.async_block_till_done() - assert secure_manual["type"] == FlowResultType.CREATE_ENTRY - assert secure_manual["data"] == { + assert secure_tunnel_manual["type"] == FlowResultType.CREATE_ENTRY + assert secure_tunnel_manual["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, CONF_KNX_SECURE_USER_ID: 2, @@ -662,265 +733,88 @@ async def test_configure_secure_knxkeys_invalid_signature(hass: HomeAssistant): assert secure_knxkeys["errors"][CONF_KNX_KNXKEY_PASSWORD] == "invalid_signature" -async def test_options_flow( +async def test_options_flow_connection_type( hass: HomeAssistant, mock_config_entry: MockConfigEntry ) -> None: - """Test options config flow.""" + """Test options flow changing interface.""" mock_config_entry.add_to_hass(hass) - gateway = _gateway_descriptor("192.168.0.1", 3675) + + menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id) + with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways: gateways.return_value = [gateway] - result = await hass.config_entries.options.async_init( - mock_config_entry.entry_id + result = await hass.config_entries.options.async_configure( + menu_step["flow_id"], + {"next_step_id": "connection_type"}, ) assert result.get("type") == FlowResultType.FORM - assert result.get("step_id") == "init" - assert "flow_id" in result - - result2 = await hass.config_entries.options.async_configure( - result["flow_id"], - user_input={ - CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, - CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255", - CONF_KNX_MCAST_PORT: 3675, - CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, - }, - ) - - await hass.async_block_till_done() - assert result2.get("type") == FlowResultType.CREATE_ENTRY - assert not result2.get("data") - - assert mock_config_entry.data == { - CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, - CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255", - CONF_HOST: "", - CONF_KNX_LOCAL_IP: None, - CONF_KNX_MCAST_PORT: 3675, - CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, - CONF_KNX_RATE_LIMIT: 20, - CONF_KNX_STATE_UPDATER: True, - } - - -@pytest.mark.parametrize( - "user_input,config_entry_data", - [ - ( - { - CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK, - CONF_HOST: "192.168.1.1", - CONF_PORT: 3675, - }, - { - CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, - CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255", - CONF_KNX_MCAST_PORT: 3675, - CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, - CONF_KNX_RATE_LIMIT: 20, - CONF_KNX_STATE_UPDATER: True, - CONF_KNX_LOCAL_IP: None, - CONF_HOST: "192.168.1.1", - CONF_PORT: 3675, - CONF_KNX_ROUTE_BACK: True, - }, - ), - ( - { - CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP, - CONF_HOST: "192.168.1.1", - CONF_PORT: 3675, - }, - { - CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, - CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255", - CONF_KNX_MCAST_PORT: 3675, - CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, - CONF_KNX_RATE_LIMIT: 20, - CONF_KNX_STATE_UPDATER: True, - CONF_KNX_LOCAL_IP: None, - CONF_HOST: "192.168.1.1", - CONF_PORT: 3675, - CONF_KNX_ROUTE_BACK: False, - }, - ), - ( - { - CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_TCP, - CONF_HOST: "192.168.1.1", - CONF_PORT: 3675, - }, - { - CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP, - CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255", - CONF_KNX_MCAST_PORT: 3675, - CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, - CONF_KNX_RATE_LIMIT: 20, - CONF_KNX_STATE_UPDATER: True, - CONF_KNX_LOCAL_IP: None, - CONF_HOST: "192.168.1.1", - CONF_PORT: 3675, - CONF_KNX_ROUTE_BACK: False, - }, - ), - ], -) -async def test_tunneling_options_flow( - hass: HomeAssistant, - mock_config_entry: MockConfigEntry, - user_input, - config_entry_data, -) -> None: - """Test options flow for tunneling.""" - mock_config_entry.add_to_hass(hass) - - gateway = _gateway_descriptor("192.168.0.1", 3675) - with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways: - gateways.return_value = [gateway] - result = await hass.config_entries.options.async_init( - mock_config_entry.entry_id - ) - - assert result.get("type") == FlowResultType.FORM - assert result.get("step_id") == "init" - assert "flow_id" in result + assert result.get("step_id") == "connection_type" result2 = await hass.config_entries.options.async_configure( result["flow_id"], user_input={ CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, - CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255", - CONF_KNX_MCAST_PORT: 3675, - CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, }, ) - assert result2.get("type") == FlowResultType.FORM - assert not result2.get("data") - assert "flow_id" in result2 + assert result2.get("step_id") == "tunnel" result3 = await hass.config_entries.options.async_configure( result2["flow_id"], - user_input=user_input, + user_input={ + CONF_KNX_GATEWAY: str(gateway), + }, ) - await hass.async_block_till_done() assert result3.get("type") == FlowResultType.CREATE_ENTRY assert not result3.get("data") - assert mock_config_entry.data == config_entry_data + assert mock_config_entry.data == { + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, + CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250", + CONF_HOST: "192.168.0.1", + CONF_PORT: 3675, + CONF_KNX_LOCAL_IP: None, + CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT, + CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, + CONF_KNX_RATE_LIMIT: 20, + CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER, + CONF_KNX_ROUTE_BACK: False, + } -@pytest.mark.parametrize( - "user_input,config_entry_data", - [ - ( - { - CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, - CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250", - CONF_KNX_MCAST_PORT: 3675, - CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, - CONF_KNX_RATE_LIMIT: 25, - CONF_KNX_STATE_UPDATER: False, - CONF_KNX_LOCAL_IP: "192.168.1.112", - }, - { - CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, - CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250", - CONF_HOST: "", - CONF_KNX_MCAST_PORT: 3675, - CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, - CONF_KNX_RATE_LIMIT: 25, - CONF_KNX_STATE_UPDATER: False, - CONF_KNX_LOCAL_IP: "192.168.1.112", - }, - ), - ( - { - CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, - CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250", - CONF_KNX_MCAST_PORT: 3675, - CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, - CONF_KNX_RATE_LIMIT: 25, - CONF_KNX_STATE_UPDATER: False, - CONF_KNX_LOCAL_IP: CONF_DEFAULT_LOCAL_IP, - }, - { - CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, - CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250", - CONF_HOST: "", - CONF_KNX_MCAST_PORT: 3675, - CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, - CONF_KNX_RATE_LIMIT: 25, - CONF_KNX_STATE_UPDATER: False, - CONF_KNX_LOCAL_IP: None, - }, - ), - ], -) -async def test_advanced_options( - hass: HomeAssistant, - mock_config_entry: MockConfigEntry, - user_input, - config_entry_data, +async def test_options_communication_settings( + hass: HomeAssistant, mock_config_entry: MockConfigEntry ) -> None: - """Test options config flow.""" + """Test options flow changing communication settings.""" mock_config_entry.add_to_hass(hass) - gateway = _gateway_descriptor("192.168.0.1", 3675) - with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways: - gateways.return_value = [gateway] - result = await hass.config_entries.options.async_init( - mock_config_entry.entry_id, context={"show_advanced_options": True} - ) + menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id) - assert result.get("type") == FlowResultType.FORM - assert result.get("step_id") == "init" - assert "flow_id" in result + result = await hass.config_entries.options.async_configure( + menu_step["flow_id"], + {"next_step_id": "communication_settings"}, + ) + assert result.get("type") == FlowResultType.FORM + assert result.get("step_id") == "communication_settings" - result2 = await hass.config_entries.options.async_configure( - result["flow_id"], - user_input=user_input, - ) + result2 = await hass.config_entries.options.async_configure( + result["flow_id"], + user_input={ + CONF_KNX_STATE_UPDATER: False, + CONF_KNX_RATE_LIMIT: 0, + }, + ) - await hass.async_block_till_done() - assert result2.get("type") == FlowResultType.CREATE_ENTRY - assert not result2.get("data") + await hass.async_block_till_done() + assert result2.get("type") == FlowResultType.CREATE_ENTRY + assert not result2.get("data") - assert mock_config_entry.data == config_entry_data - - -@pytest.mark.parametrize( - "config_entry_data,result", - [ - ( - { - CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, - CONF_KNX_ROUTE_BACK: False, - }, - CONF_KNX_LABEL_TUNNELING_UDP, - ), - ( - { - CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, - CONF_KNX_ROUTE_BACK: True, - }, - CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK, - ), - ( - { - CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP, - CONF_KNX_ROUTE_BACK: False, - }, - CONF_KNX_LABEL_TUNNELING_TCP, - ), - ], -) -async def test_get_knx_tunneling_type( - config_entry_data, - result, -) -> None: - """Test converting config entry data to tunneling type for config flow.""" - assert get_knx_tunneling_type(config_entry_data) == result + assert mock_config_entry.data == { + **DEFAULT_ENTRY_DATA, + CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, + CONF_KNX_STATE_UPDATER: False, + CONF_KNX_RATE_LIMIT: 0, + }