diff --git a/homeassistant/components/knx/__init__.py b/homeassistant/components/knx/__init__.py index ea654c358e7..edb9cc62008 100644 --- a/homeassistant/components/knx/__init__.py +++ b/homeassistant/components/knx/__init__.py @@ -401,6 +401,9 @@ class KNXModule: ) return ConnectionConfig( auto_reconnect=True, + individual_address=self.entry.data.get( + CONF_KNX_TUNNEL_ENDPOINT_IA, # may be configured at knxkey upload + ), secure_config=SecureConfig( knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD), knxkeys_file_path=_knxkeys_file, diff --git a/homeassistant/components/knx/config_flow.py b/homeassistant/components/knx/config_flow.py index feeb7626577..eda160cd1a6 100644 --- a/homeassistant/components/knx/config_flow.py +++ b/homeassistant/components/knx/config_flow.py @@ -4,7 +4,7 @@ from __future__ import annotations from abc import ABC, abstractmethod from collections.abc import AsyncGenerator -from typing import Any, Final +from typing import Any, Final, Literal import voluptuous as vol from xknx import XKNX @@ -121,6 +121,15 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow): self._gatewayscanner: GatewayScanner | None = None self._async_scan_gen: AsyncGenerator[GatewayDescriptor] | None = None + @property + def _xknx(self) -> XKNX: + """Return XKNX instance.""" + if isinstance(self, OptionsFlow) and ( + knx_module := self.hass.data.get(KNX_MODULE_KEY) + ): + return knx_module.xknx + return XKNX() + @abstractmethod def finish_flow(self) -> ConfigFlowResult: """Finish the flow.""" @@ -183,14 +192,8 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow): CONF_KNX_ROUTING: CONF_KNX_ROUTING.capitalize(), } - if isinstance(self, OptionsFlow) and ( - knx_module := self.hass.data.get(KNX_MODULE_KEY) - ): - xknx = knx_module.xknx - else: - xknx = XKNX() self._gatewayscanner = GatewayScanner( - xknx, stop_on_found=0, timeout_in_seconds=2 + self._xknx, stop_on_found=0, timeout_in_seconds=2 ) # keep a reference to the generator to scan in background until user selects a connection type self._async_scan_gen = self._gatewayscanner.async_scan() @@ -204,8 +207,25 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow): CONF_KNX_AUTOMATIC: CONF_KNX_AUTOMATIC.capitalize() } | supported_connection_types + default_connection_type: Literal["automatic", "tunneling", "routing"] + _current_conn = self.initial_data.get(CONF_KNX_CONNECTION_TYPE) + if _current_conn in ( + CONF_KNX_TUNNELING, + CONF_KNX_TUNNELING_TCP, + CONF_KNX_TUNNELING_TCP_SECURE, + ): + default_connection_type = CONF_KNX_TUNNELING + elif _current_conn in (CONF_KNX_ROUTING, CONF_KNX_ROUTING_SECURE): + default_connection_type = CONF_KNX_ROUTING + elif CONF_KNX_AUTOMATIC in supported_connection_types: + default_connection_type = CONF_KNX_AUTOMATIC + else: + default_connection_type = CONF_KNX_TUNNELING + fields = { - vol.Required(CONF_KNX_CONNECTION_TYPE): vol.In(supported_connection_types) + vol.Required( + CONF_KNX_CONNECTION_TYPE, default=default_connection_type + ): vol.In(supported_connection_types) } return self.async_show_form( step_id="connection_type", data_schema=vol.Schema(fields) @@ -216,8 +236,7 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow): ) -> ConfigFlowResult: """Select a tunnel from a list. - Will be skipped if the gateway scan was unsuccessful - or if only one gateway was found. + Will be skipped if the gateway scan was unsuccessful. """ if user_input is not None: if user_input[CONF_KNX_GATEWAY] == OPTION_MANUAL_TUNNEL: @@ -247,6 +266,8 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow): user_password=None, tunnel_endpoint_ia=None, ) + if connection_type == CONF_KNX_TUNNELING_TCP: + return await self.async_step_tcp_tunnel_endpoint() if connection_type == CONF_KNX_TUNNELING_TCP_SECURE: return await self.async_step_secure_key_source_menu_tunnel() self.new_title = f"Tunneling @ {self._selected_tunnel}" @@ -255,16 +276,99 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow): if not self._found_tunnels: return await self.async_step_manual_tunnel() - errors: dict = {} - tunnel_options = { - str(tunnel): f"{tunnel}{' 🔐' if tunnel.tunnelling_requires_secure else ''}" + tunnel_options = [ + selector.SelectOptionDict( + value=str(tunnel), + label=( + f"{tunnel}" + f"{' TCP' if tunnel.supports_tunnelling_tcp else ' UDP'}" + f"{' 🔐 Secure tunneling' if tunnel.tunnelling_requires_secure else ''}" + ), + ) for tunnel in self._found_tunnels + ] + tunnel_options.append( + selector.SelectOptionDict( + value=OPTION_MANUAL_TUNNEL, label=OPTION_MANUAL_TUNNEL + ) + ) + default_tunnel = next( + ( + str(tunnel) + for tunnel in self._found_tunnels + if tunnel.ip_addr == self.initial_data.get(CONF_HOST) + ), + vol.UNDEFINED, + ) + fields = { + vol.Required( + CONF_KNX_GATEWAY, default=default_tunnel + ): selector.SelectSelector( + selector.SelectSelectorConfig( + options=tunnel_options, + mode=selector.SelectSelectorMode.LIST, + ) + ) } - tunnel_options |= {OPTION_MANUAL_TUNNEL: 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)) + + async def async_step_tcp_tunnel_endpoint( + self, user_input: dict | None = None + ) -> ConfigFlowResult: + """Select specific tunnel endpoint for plain TCP connection.""" + if user_input is not None: + selected_tunnel_ia: str | None = ( + None + if user_input[CONF_KNX_TUNNEL_ENDPOINT_IA] == CONF_KNX_AUTOMATIC + else user_input[CONF_KNX_TUNNEL_ENDPOINT_IA] + ) + self.new_entry_data |= KNXConfigEntryData( + tunnel_endpoint_ia=selected_tunnel_ia, + ) + self.new_title = ( + f"{selected_tunnel_ia or 'Tunneling'} @ {self._selected_tunnel}" + ) + return self.finish_flow() + + # this step is only called from async_step_tunnel so self._selected_tunnel is always set + assert self._selected_tunnel + # skip if only one tunnel endpoint or no tunnelling slot infos + if len(self._selected_tunnel.tunnelling_slots) <= 1: + return self.finish_flow() + + tunnel_endpoint_options = [ + selector.SelectOptionDict( + value=CONF_KNX_AUTOMATIC, label=CONF_KNX_AUTOMATIC.capitalize() + ) + ] + _current_ia = self._xknx.current_address + tunnel_endpoint_options.extend( + selector.SelectOptionDict( + value=str(slot), + label=( + f"{slot} - {'current connection' if slot == _current_ia else 'occupied' if not slot_status.free else 'free'}" + ), + ) + for slot, slot_status in self._selected_tunnel.tunnelling_slots.items() + ) + default_endpoint = ( + self.initial_data.get(CONF_KNX_TUNNEL_ENDPOINT_IA) or CONF_KNX_AUTOMATIC + ) return self.async_show_form( - step_id="tunnel", data_schema=vol.Schema(fields), errors=errors + step_id="tcp_tunnel_endpoint", + data_schema=vol.Schema( + { + vol.Required( + CONF_KNX_TUNNEL_ENDPOINT_IA, default=default_endpoint + ): selector.SelectSelector( + selector.SelectSelectorConfig( + options=tunnel_endpoint_options, + mode=selector.SelectSelectorMode.LIST, + ) + ), + } + ), ) async def async_step_manual_tunnel( @@ -612,12 +716,15 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow): ) for endpoint in self._tunnel_endpoints ) + default_endpoint = ( + self.initial_data.get(CONF_KNX_TUNNEL_ENDPOINT_IA) or CONF_KNX_AUTOMATIC + ) return self.async_show_form( step_id="knxkeys_tunnel_select", data_schema=vol.Schema( { vol.Required( - CONF_KNX_TUNNEL_ENDPOINT_IA, default=CONF_KNX_AUTOMATIC + CONF_KNX_TUNNEL_ENDPOINT_IA, default=default_endpoint ): selector.SelectSelector( selector.SelectSelectorConfig( options=tunnel_endpoint_options, diff --git a/homeassistant/components/knx/strings.json b/homeassistant/components/knx/strings.json index d697fa79e78..cde697007aa 100644 --- a/homeassistant/components/knx/strings.json +++ b/homeassistant/components/knx/strings.json @@ -15,6 +15,13 @@ "gateway": "KNX Tunnel Connection" } }, + "tcp_tunnel_endpoint": { + "title": "[%key:component::knx::config::step::knxkeys_tunnel_select::title%]", + "description": "[%key:component::knx::config::step::knxkeys_tunnel_select::description%]", + "data": { + "tunnel_endpoint_ia": "[%key:component::knx::config::step::knxkeys_tunnel_select::data::user_id%]" + } + }, "manual_tunnel": { "title": "Tunnel settings", "description": "Please enter the connection information of your tunneling device.", @@ -61,9 +68,9 @@ }, "knxkeys_tunnel_select": { "title": "Tunnel endpoint", - "description": "Select the tunnel used for connection.", + "description": "Select the tunnel endpoint used for the connection.", "data": { - "user_id": "`Automatic` will use the first free tunnel endpoint." + "user_id": "'Automatic' selects a free tunnel endpoint for you when connecting. If you're unsure, this is the best option." } }, "secure_tunnel_manual": { @@ -159,6 +166,13 @@ "gateway": "[%key:component::knx::config::step::tunnel::data::gateway%]" } }, + "tcp_tunnel_endpoint": { + "title": "[%key:component::knx::config::step::knxkeys_tunnel_select::title%]", + "description": "[%key:component::knx::config::step::knxkeys_tunnel_select::description%]", + "data": { + "tunnel_endpoint_ia": "[%key:component::knx::config::step::knxkeys_tunnel_select::data::user_id%]" + } + }, "manual_tunnel": { "title": "[%key:component::knx::config::step::manual_tunnel::title%]", "description": "[%key:component::knx::config::step::manual_tunnel::description%]", diff --git a/tests/components/knx/test_config_flow.py b/tests/components/knx/test_config_flow.py index 2187721a518..8ed79f837bb 100644 --- a/tests/components/knx/test_config_flow.py +++ b/tests/components/knx/test_config_flow.py @@ -7,6 +7,7 @@ import pytest from xknx.exceptions.exception import CommunicationError, InvalidSecureConfiguration from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT from xknx.io.gateway_scanner import GatewayDescriptor +from xknx.knxip.dib import TunnelingSlotStatus from xknx.secure.keyring import sync_load_keyring from xknx.telegram import IndividualAddress @@ -105,6 +106,7 @@ def _gateway_descriptor( port: int, supports_tunnelling_tcp: bool = False, requires_secure: bool = False, + slots: bool = True, ) -> GatewayDescriptor: """Get mock gw descriptor.""" descriptor = GatewayDescriptor( @@ -120,6 +122,12 @@ def _gateway_descriptor( ) descriptor.tunnelling_requires_secure = requires_secure descriptor.routing_requires_secure = requires_secure + if supports_tunnelling_tcp and slots: + descriptor.tunnelling_slots = { + IndividualAddress("1.0.240"): TunnelingSlotStatus(True, True, True), + IndividualAddress("1.0.241"): TunnelingSlotStatus(True, True, False), + IndividualAddress("1.0.242"): TunnelingSlotStatus(True, True, True), + } return descriptor @@ -791,12 +799,14 @@ async def test_tunneling_setup_for_multiple_found_gateways( hass: HomeAssistant, knx_setup ) -> None: """Test tunneling if multiple gateways are found.""" - gateway = _gateway_descriptor("192.168.0.1", 3675) - gateway2 = _gateway_descriptor("192.168.1.100", 3675) + gateway_udp = _gateway_descriptor("192.168.0.1", 3675) + gateway_tcp = _gateway_descriptor("192.168.1.100", 3675, True) with patch( "homeassistant.components.knx.config_flow.GatewayScanner" ) as gateway_scanner_mock: - gateway_scanner_mock.return_value = GatewayScannerMock([gateway, gateway2]) + gateway_scanner_mock.return_value = GatewayScannerMock( + [gateway_udp, gateway_tcp] + ) result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER} ) @@ -815,7 +825,7 @@ async def test_tunneling_setup_for_multiple_found_gateways( result = await hass.config_entries.flow.async_configure( tunnel_flow["flow_id"], - {CONF_KNX_GATEWAY: str(gateway)}, + {CONF_KNX_GATEWAY: str(gateway_udp)}, ) assert result["type"] is FlowResultType.CREATE_ENTRY assert result["data"] == { @@ -833,6 +843,110 @@ async def test_tunneling_setup_for_multiple_found_gateways( knx_setup.assert_called_once() +async def test_tunneling_setup_tcp_endpoint_select_skip( + hass: HomeAssistant, knx_setup +) -> None: + """Test tunneling TCP endpoint selection skipped if no slot info found.""" + gateway_udp = _gateway_descriptor("192.168.0.1", 3675) + gateway_tcp_no_slots = _gateway_descriptor("192.168.1.100", 3675, True, slots=False) + with patch( + "homeassistant.components.knx.config_flow.GatewayScanner" + ) as gateway_scanner_mock: + gateway_scanner_mock.return_value = GatewayScannerMock( + [gateway_udp, gateway_tcp_no_slots] + ) + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] is FlowResultType.FORM + assert not result["errors"] + + tunnel_flow = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, + }, + ) + assert tunnel_flow["type"] is FlowResultType.FORM + assert tunnel_flow["step_id"] == "tunnel" + assert not tunnel_flow["errors"] + + result = await hass.config_entries.flow.async_configure( + tunnel_flow["flow_id"], + {CONF_KNX_GATEWAY: str(gateway_tcp_no_slots)}, + ) + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"] == { + **DEFAULT_ENTRY_DATA, + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP, + CONF_HOST: "192.168.1.100", + CONF_PORT: 3675, + CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", + CONF_KNX_ROUTE_BACK: False, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, + } + knx_setup.assert_called_once() + + +async def test_tunneling_setup_tcp_endpoint_select( + hass: HomeAssistant, knx_setup +) -> None: + """Test tunneling TCP endpoint selection.""" + gateway_tcp = _gateway_descriptor("192.168.1.100", 3675, True) + with patch( + "homeassistant.components.knx.config_flow.GatewayScanner" + ) as gateway_scanner_mock: + gateway_scanner_mock.return_value = GatewayScannerMock([gateway_tcp]) + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] is FlowResultType.FORM + assert not result["errors"] + + tunnel_flow = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, + }, + ) + assert tunnel_flow["type"] is FlowResultType.FORM + assert tunnel_flow["step_id"] == "tunnel" + assert not tunnel_flow["errors"] + + endpoint_flow = await hass.config_entries.flow.async_configure( + tunnel_flow["flow_id"], + {CONF_KNX_GATEWAY: str(gateway_tcp)}, + ) + + assert endpoint_flow["type"] is FlowResultType.FORM + assert endpoint_flow["step_id"] == "tcp_tunnel_endpoint" + assert not endpoint_flow["errors"] + + result = await hass.config_entries.flow.async_configure( + endpoint_flow["flow_id"], + {CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.242"}, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == "1.0.242 @ 1.0.0 - Test @ 192.168.1.100:3675" + assert result["data"] == { + **DEFAULT_ENTRY_DATA, + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP, + CONF_HOST: "192.168.1.100", + CONF_PORT: 3675, + CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", + CONF_KNX_ROUTE_BACK: False, + CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.242", + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, + } + knx_setup.assert_called_once() + + @pytest.mark.parametrize( "gateway", [ @@ -1319,6 +1433,64 @@ async def test_options_flow_secure_manual_to_keyfile( knx_setup.assert_called_once() +async def test_options_flow_routing(hass: HomeAssistant, knx_setup) -> None: + """Test options flow changing routing settings.""" + mock_config_entry = MockConfigEntry( + title="KNX", + domain="knx", + data={ + **DEFAULT_ENTRY_DATA, + CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, + }, + ) + gateway = _gateway_descriptor("192.168.0.1", 3676) + mock_config_entry.add_to_hass(hass) + await hass.config_entries.async_setup(mock_config_entry.entry_id) + menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id) + + with patch( + "homeassistant.components.knx.config_flow.GatewayScanner" + ) as gateway_scanner_mock: + gateway_scanner_mock.return_value = GatewayScannerMock([gateway]) + result = await hass.config_entries.options.async_configure( + menu_step["flow_id"], + {"next_step_id": "connection_type"}, + ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "connection_type" + + result2 = await hass.config_entries.options.async_configure( + result["flow_id"], + user_input={ + CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, + }, + ) + assert result2["type"] is FlowResultType.FORM + assert result2["step_id"] == "routing" + assert result2["errors"] == {} + + result3 = await hass.config_entries.options.async_configure( + result2["flow_id"], + { + CONF_KNX_INDIVIDUAL_ADDRESS: "2.0.4", + }, + ) + assert result3["type"] is FlowResultType.CREATE_ENTRY + assert mock_config_entry.data == { + **DEFAULT_ENTRY_DATA, + CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, + CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, + CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT, + CONF_KNX_LOCAL_IP: None, + CONF_KNX_INDIVIDUAL_ADDRESS: "2.0.4", + CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, + CONF_KNX_SECURE_USER_ID: None, + CONF_KNX_SECURE_USER_PASSWORD: None, + CONF_KNX_TUNNEL_ENDPOINT_IA: None, + } + knx_setup.assert_called_once() + + async def test_options_communication_settings( hass: HomeAssistant, knx_setup, mock_config_entry: MockConfigEntry ) -> None: