diff --git a/homeassistant/components/knx/__init__.py b/homeassistant/components/knx/__init__.py index b9b7918f766..cdaf5c73e74 100644 --- a/homeassistant/components/knx/__init__.py +++ b/homeassistant/components/knx/__init__.py @@ -45,6 +45,7 @@ from .const import ( CONF_KNX_INDIVIDUAL_ADDRESS, CONF_KNX_ROUTING, CONF_KNX_TUNNELING, + CONF_KNX_TUNNELING_TCP, DATA_HASS_CONFIG, DATA_KNX_CONFIG, DOMAIN, @@ -404,6 +405,13 @@ class KNXModule: route_back=self.config.get(ConnectionSchema.CONF_KNX_ROUTE_BACK, False), auto_reconnect=True, ) + if _conn_type == CONF_KNX_TUNNELING_TCP: + return ConnectionConfig( + connection_type=ConnectionType.TUNNELING_TCP, + gateway_ip=self.config[CONF_HOST], + gateway_port=self.config[CONF_PORT], + auto_reconnect=True, + ) return ConnectionConfig(auto_reconnect=True) async def connection_state_changed_cb(self, state: XknxConnectionState) -> None: diff --git a/homeassistant/components/knx/config_flow.py b/homeassistant/components/knx/config_flow.py index 99cdc4807c6..4f7a9d6723c 100644 --- a/homeassistant/components/knx/config_flow.py +++ b/homeassistant/components/knx/config_flow.py @@ -22,6 +22,7 @@ from .const import ( CONF_KNX_INITIAL_CONNECTION_TYPES, CONF_KNX_ROUTING, CONF_KNX_TUNNELING, + CONF_KNX_TUNNELING_TCP, DOMAIN, ) from .schema import ConnectionSchema @@ -38,15 +39,19 @@ DEFAULT_ENTRY_DATA: Final = { ConnectionSchema.CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT, } +CONF_KNX_TUNNELING_TYPE: Final = "tunneling_type" +CONF_KNX_LABEL_TUNNELING_TCP: Final = "TCP" +CONF_KNX_LABEL_TUNNELING_UDP: Final = "UDP" +CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK: Final = "UDP with route back / NAT mode" + class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): """Handle a KNX config flow.""" VERSION = 1 - _tunnels: list[GatewayDescriptor] - _gateway_ip: str = "" - _gateway_port: int = DEFAULT_MCAST_PORT + _found_tunnels: list[GatewayDescriptor] + _selected_tunnel: GatewayDescriptor | None @staticmethod @callback @@ -59,7 +64,8 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): if self._async_current_entries(): return self.async_abort(reason="single_instance_allowed") - self._tunnels = [] + self._found_tunnels = [] + self._selected_tunnel = None return await self.async_step_type() async def async_step_type(self, user_input: dict | None = None) -> FlowResult: @@ -75,7 +81,7 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): if connection_type == CONF_KNX_ROUTING: return await self.async_step_routing() - if connection_type == CONF_KNX_TUNNELING and self._tunnels: + if connection_type == CONF_KNX_TUNNELING and self._found_tunnels: return await self.async_step_tunnel() return await self.async_step_manual_tunnel() @@ -88,7 +94,7 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): if gateways: # add automatic only if a gateway responded supported_connection_types.insert(0, CONF_KNX_AUTOMATIC) - self._tunnels = [ + self._found_tunnels = [ gateway for gateway in gateways if gateway.supports_tunnelling ] @@ -100,11 +106,35 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): step_id="type", data_schema=vol.Schema(fields), errors=errors ) + 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: + 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() + + # skip this step if the user has only one unique gateway. + if len(self._found_tunnels) == 1: + self._selected_tunnel = self._found_tunnels[0] + 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)} + + return self.async_show_form( + step_id="tunnel", data_schema=vol.Schema(fields), errors=errors + ) + async def async_step_manual_tunnel( self, user_input: dict | None = None ) -> FlowResult: - """General setup.""" + """Manually configure tunnel connection parameters. Fields default to preselected gateway if one was found.""" if user_input is not None: + connection_type = user_input[CONF_KNX_TUNNELING_TYPE] return self.async_create_entry( title=f"{CONF_KNX_TUNNELING.capitalize()} @ {user_input[CONF_HOST]}", data={ @@ -114,26 +144,41 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): CONF_KNX_INDIVIDUAL_ADDRESS: user_input[ CONF_KNX_INDIVIDUAL_ADDRESS ], - ConnectionSchema.CONF_KNX_ROUTE_BACK: user_input[ - ConnectionSchema.CONF_KNX_ROUTE_BACK - ], + ConnectionSchema.CONF_KNX_ROUTE_BACK: ( + connection_type == CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK + ), ConnectionSchema.CONF_KNX_LOCAL_IP: user_input.get( ConnectionSchema.CONF_KNX_LOCAL_IP ), - CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, + CONF_KNX_CONNECTION_TYPE: ( + CONF_KNX_TUNNELING_TCP + if connection_type == CONF_KNX_LABEL_TUNNELING_TCP + else CONF_KNX_TUNNELING + ), }, ) errors: dict = {} + connection_methods: list[str] = [ + CONF_KNX_LABEL_TUNNELING_TCP, + CONF_KNX_LABEL_TUNNELING_UDP, + CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK, + ] + ip_address = "" + port = DEFAULT_MCAST_PORT + if self._selected_tunnel is not None: + 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) + fields = { - vol.Required(CONF_HOST, default=self._gateway_ip): str, - vol.Required(CONF_PORT, default=self._gateway_port): vol.Coerce(int), + vol.Required(CONF_KNX_TUNNELING_TYPE): vol.In(connection_methods), + vol.Required(CONF_HOST, default=ip_address): str, + vol.Required(CONF_PORT, default=port): cv.port, vol.Required( CONF_KNX_INDIVIDUAL_ADDRESS, default=XKNX.DEFAULT_ADDRESS ): str, - vol.Required( - ConnectionSchema.CONF_KNX_ROUTE_BACK, default=False - ): vol.Coerce(bool), } if self.show_advanced_options: @@ -143,38 +188,6 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN): step_id="manual_tunnel", data_schema=vol.Schema(fields), errors=errors ) - 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: - gateway: GatewayDescriptor = next( - gateway - for gateway in self._tunnels - if user_input[CONF_KNX_GATEWAY] == str(gateway) - ) - - self._gateway_ip = gateway.ip_addr - self._gateway_port = gateway.port - - return await self.async_step_manual_tunnel() - - errors: dict = {} - tunnel_repr = { - str(tunnel) for tunnel in self._tunnels if tunnel.supports_tunnelling - } - - # skip this step if the user has only one unique gateway. - if len(tunnel_repr) == 1: - _gateway: GatewayDescriptor = self._tunnels[0] - self._gateway_ip = _gateway.ip_addr - self._gateway_port = _gateway.port - return await self.async_step_manual_tunnel() - - fields = {vol.Required(CONF_KNX_GATEWAY): vol.In(tunnel_repr)} - - return self.async_show_form( - step_id="tunnel", data_schema=vol.Schema(fields), errors=errors - ) - async def async_step_routing(self, user_input: dict | None = None) -> FlowResult: """Routing setup.""" if user_input is not None: @@ -291,65 +304,6 @@ class KNXOptionsFlowHandler(OptionsFlow): """Initialize KNX options flow.""" self.config_entry = config_entry - 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 - ): - return self.async_show_form( - step_id="tunnel", - data_schema=vol.Schema( - { - vol.Required( - CONF_HOST, default=self.current_config.get(CONF_HOST) - ): str, - vol.Required( - CONF_PORT, default=self.current_config.get(CONF_PORT, 3671) - ): cv.port, - vol.Required( - ConnectionSchema.CONF_KNX_ROUTE_BACK, - default=self.current_config.get( - ConnectionSchema.CONF_KNX_ROUTE_BACK, False - ), - ): vol.Coerce(bool), - } - ), - last_step=True, - ) - - entry_data = { - **DEFAULT_ENTRY_DATA, - **self.general_settings, - ConnectionSchema.CONF_KNX_LOCAL_IP: self.general_settings.get( - ConnectionSchema.CONF_KNX_LOCAL_IP - ) - if self.general_settings.get(ConnectionSchema.CONF_KNX_LOCAL_IP) - != CONF_DEFAULT_LOCAL_IP - else None, - CONF_HOST: self.current_config.get(CONF_HOST, ""), - } - - if user_input is not None: - entry_data = { - **entry_data, - **user_input, - } - - 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]}" - - self.hass.config_entries.async_update_entry( - self.config_entry, - data=entry_data, - title=entry_title, - ) - - return self.async_create_entry(title="", data={}) - async def async_step_init( self, user_input: dict[str, Any] | None = None ) -> FlowResult: @@ -368,7 +322,12 @@ class KNXOptionsFlowHandler(OptionsFlow): data_schema = { vol.Required( CONF_KNX_CONNECTION_TYPE, - default=self.current_config.get(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, @@ -427,6 +386,94 @@ class KNXOptionsFlowHandler(OptionsFlow): != CONF_KNX_TUNNELING, ) + 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) + ): str, + vol.Required( + CONF_PORT, default=self.current_config.get(CONF_PORT, 3671) + ): cv.port, + } + ), + last_step=True, + ) + + entry_data = { + **DEFAULT_ENTRY_DATA, + **self.general_settings, + ConnectionSchema.CONF_KNX_LOCAL_IP: self.general_settings.get( + ConnectionSchema.CONF_KNX_LOCAL_IP + ) + if self.general_settings.get(ConnectionSchema.CONF_KNX_LOCAL_IP) + != CONF_DEFAULT_LOCAL_IP + else None, + CONF_HOST: self.current_config.get(CONF_HOST, ""), + } + + 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], + ConnectionSchema.CONF_KNX_ROUTE_BACK: ( + connection_type == CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK + ), + CONF_KNX_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]}" + if entry_data[CONF_KNX_CONNECTION_TYPE] == CONF_KNX_TUNNELING_TCP: + entry_title = ( + f"{CONF_KNX_TUNNELING.capitalize()} (TCP) @ {entry_data[CONF_HOST]}" + ) + + 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(ConnectionSchema.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/const.py b/homeassistant/components/knx/const.py index f50460de173..ac6a156a90a 100644 --- a/homeassistant/components/knx/const.py +++ b/homeassistant/components/knx/const.py @@ -34,6 +34,7 @@ CONF_KNX_CONNECTION_TYPE: Final = "connection_type" CONF_KNX_AUTOMATIC: Final = "automatic" CONF_KNX_ROUTING: Final = "routing" CONF_KNX_TUNNELING: Final = "tunneling" +CONF_KNX_TUNNELING_TCP: Final = "tunneling_tcp" CONF_PAYLOAD: Final = "payload" CONF_PAYLOAD_LENGTH: Final = "payload_length" CONF_RESET_AFTER: Final = "reset_after" diff --git a/homeassistant/components/knx/strings.json b/homeassistant/components/knx/strings.json index d219880be5c..77e18cad808 100644 --- a/homeassistant/components/knx/strings.json +++ b/homeassistant/components/knx/strings.json @@ -16,10 +16,10 @@ "manual_tunnel": { "description": "Please enter the connection information of your tunneling device.", "data": { + "tunneling_type": "KNX Tunneling Type", "port": "[%key:common::config_flow::data::port%]", "host": "[%key:common::config_flow::data::host%]", "individual_address": "Individual address for the connection", - "route_back": "Route Back / NAT Mode", "local_ip": "Local IP of Home Assistant (leave empty for automatic detection)" } }, @@ -56,9 +56,9 @@ }, "tunnel": { "data": { + "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" + "host": "[%key:common::config_flow::data::host%]" } } } diff --git a/homeassistant/components/knx/translations/en.json b/homeassistant/components/knx/translations/en.json index a7c77e28c0d..93ba7c006f0 100644 --- a/homeassistant/components/knx/translations/en.json +++ b/homeassistant/components/knx/translations/en.json @@ -10,11 +10,11 @@ "step": { "manual_tunnel": { "data": { + "tunneling_type": "KNX Tunneling Type", "host": "Host", "individual_address": "Individual address for the connection", "local_ip": "Local IP of Home Assistant (leave empty for automatic detection)", - "port": "Port", - "route_back": "Route Back / NAT Mode" + "port": "Port" }, "description": "Please enter the connection information of your tunneling device." }, @@ -56,10 +56,10 @@ }, "tunnel": { "data": { + "tunneling_type": "KNX Tunneling Type", "host": "Host", "local_ip": "Local IP (leave empty if unsure)", - "port": "Port", - "route_back": "Route Back / NAT Mode" + "port": "Port" } } } diff --git a/tests/components/knx/test_config_flow.py b/tests/components/knx/test_config_flow.py index aec757a1086..d6c2ec98b20 100644 --- a/tests/components/knx/test_config_flow.py +++ b/tests/components/knx/test_config_flow.py @@ -11,7 +11,12 @@ from homeassistant.components.knx import ConnectionSchema from homeassistant.components.knx.config_flow import ( CONF_DEFAULT_LOCAL_IP, CONF_KNX_GATEWAY, + CONF_KNX_LABEL_TUNNELING_TCP, + CONF_KNX_LABEL_TUNNELING_UDP, + CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK, + CONF_KNX_TUNNELING_TYPE, DEFAULT_ENTRY_DATA, + get_knx_tunneling_type, ) from homeassistant.components.knx.const import ( CONF_KNX_AUTOMATIC, @@ -19,6 +24,7 @@ from homeassistant.components.knx.const import ( CONF_KNX_INDIVIDUAL_ADDRESS, CONF_KNX_ROUTING, CONF_KNX_TUNNELING, + CONF_KNX_TUNNELING_TCP, DOMAIN, ) from homeassistant.const import CONF_HOST, CONF_PORT @@ -28,7 +34,9 @@ from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY, RESULT_TYPE_ from tests.common import MockConfigEntry -def _gateway_descriptor(ip: str, port: int) -> GatewayDescriptor: +def _gateway_descriptor( + ip: str, port: int, supports_tunnelling_tcp: bool = False +) -> GatewayDescriptor: """Get mock gw descriptor.""" return GatewayDescriptor( "Test", @@ -38,6 +46,7 @@ def _gateway_descriptor(ip: str, port: int) -> GatewayDescriptor: "127.0.0.1", supports_routing=True, supports_tunnelling=True, + supports_tunnelling_tcp=supports_tunnelling_tcp, ) @@ -153,9 +162,64 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None: assert len(mock_setup_entry.mock_calls) == 1 -async def test_tunneling_setup(hass: HomeAssistant) -> None: +@pytest.mark.parametrize( + "user_input,config_entry_data", + [ + ( + { + CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3675, + }, + { + **DEFAULT_ENTRY_DATA, + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3675, + CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250", + ConnectionSchema.CONF_KNX_ROUTE_BACK: False, + ConnectionSchema.CONF_KNX_LOCAL_IP: None, + }, + ), + ( + { + CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_TCP, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3675, + }, + { + **DEFAULT_ENTRY_DATA, + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3675, + CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250", + ConnectionSchema.CONF_KNX_ROUTE_BACK: False, + ConnectionSchema.CONF_KNX_LOCAL_IP: None, + }, + ), + ( + { + CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3675, + }, + { + **DEFAULT_ENTRY_DATA, + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3675, + CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250", + ConnectionSchema.CONF_KNX_ROUTE_BACK: True, + ConnectionSchema.CONF_KNX_LOCAL_IP: None, + }, + ), + ], +) +async def test_tunneling_setup( + hass: HomeAssistant, user_input, config_entry_data +) -> None: """Test tunneling if only one gateway is found.""" - gateway = _gateway_descriptor("192.168.0.1", 3675) + gateway = _gateway_descriptor("192.168.0.1", 3675, True) with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways: gateways.return_value = [gateway] result = await hass.config_entries.flow.async_init( @@ -181,23 +245,12 @@ async def test_tunneling_setup(hass: HomeAssistant) -> None: ) as mock_setup_entry: result3 = await hass.config_entries.flow.async_configure( result2["flow_id"], - { - CONF_HOST: "192.168.0.1", - CONF_PORT: 3675, - }, + user_input, ) await hass.async_block_till_done() assert result3["type"] == RESULT_TYPE_CREATE_ENTRY assert result3["title"] == "Tunneling @ 192.168.0.1" - assert result3["data"] == { - **DEFAULT_ENTRY_DATA, - CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, - CONF_HOST: "192.168.0.1", - CONF_PORT: 3675, - CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250", - ConnectionSchema.CONF_KNX_ROUTE_BACK: False, - ConnectionSchema.CONF_KNX_LOCAL_IP: None, - } + assert result3["data"] == config_entry_data assert len(mock_setup_entry.mock_calls) == 1 @@ -235,6 +288,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_HOST: "192.168.0.2", CONF_PORT: 3675, ConnectionSchema.CONF_KNX_LOCAL_IP: "192.168.1.112", @@ -294,6 +348,7 @@ async def test_tunneling_setup_for_multiple_found_gateways(hass: HomeAssistant) 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, }, @@ -595,8 +650,73 @@ async def test_options_flow( } +@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", + ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, + ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, + ConnectionSchema.CONF_KNX_RATE_LIMIT: 20, + ConnectionSchema.CONF_KNX_STATE_UPDATER: True, + ConnectionSchema.CONF_KNX_LOCAL_IP: None, + CONF_HOST: "192.168.1.1", + CONF_PORT: 3675, + ConnectionSchema.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", + ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, + ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, + ConnectionSchema.CONF_KNX_RATE_LIMIT: 20, + ConnectionSchema.CONF_KNX_STATE_UPDATER: True, + ConnectionSchema.CONF_KNX_LOCAL_IP: None, + CONF_HOST: "192.168.1.1", + CONF_PORT: 3675, + ConnectionSchema.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", + ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, + ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, + ConnectionSchema.CONF_KNX_RATE_LIMIT: 20, + ConnectionSchema.CONF_KNX_STATE_UPDATER: True, + ConnectionSchema.CONF_KNX_LOCAL_IP: None, + CONF_HOST: "192.168.1.1", + CONF_PORT: 3675, + ConnectionSchema.CONF_KNX_ROUTE_BACK: False, + }, + ), + ], +) async def test_tunneling_options_flow( - hass: HomeAssistant, mock_config_entry: MockConfigEntry + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + user_input, + config_entry_data, ) -> None: """Test options flow for tunneling.""" mock_config_entry.add_to_hass(hass) @@ -628,29 +748,14 @@ async def test_tunneling_options_flow( result3 = await hass.config_entries.options.async_configure( result2["flow_id"], - user_input={ - CONF_HOST: "192.168.1.1", - CONF_PORT: 3675, - ConnectionSchema.CONF_KNX_ROUTE_BACK: True, - }, + user_input=user_input, ) await hass.async_block_till_done() assert result3.get("type") == RESULT_TYPE_CREATE_ENTRY assert not result3.get("data") - assert mock_config_entry.data == { - CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, - CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255", - ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, - ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, - ConnectionSchema.CONF_KNX_RATE_LIMIT: 20, - ConnectionSchema.CONF_KNX_STATE_UPDATER: True, - ConnectionSchema.CONF_KNX_LOCAL_IP: None, - CONF_HOST: "192.168.1.1", - CONF_PORT: 3675, - ConnectionSchema.CONF_KNX_ROUTE_BACK: True, - } + assert mock_config_entry.data == config_entry_data @pytest.mark.parametrize( @@ -730,3 +835,37 @@ async def test_advanced_options( 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, + ConnectionSchema.CONF_KNX_ROUTE_BACK: False, + }, + CONF_KNX_LABEL_TUNNELING_UDP, + ), + ( + { + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, + ConnectionSchema.CONF_KNX_ROUTE_BACK: True, + }, + CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK, + ), + ( + { + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP, + ConnectionSchema.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