diff --git a/homeassistant/components/knx/config_flow.py b/homeassistant/components/knx/config_flow.py index 30e48cd4d6a32..f7ef0d943a9ff 100644 --- a/homeassistant/components/knx/config_flow.py +++ b/homeassistant/components/knx/config_flow.py @@ -7,9 +7,10 @@ import voluptuous as vol from xknx import XKNX -from xknx.exceptions.exception import InvalidSecureConfiguration +from xknx.exceptions.exception import CommunicationError, InvalidSecureConfiguration from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT from xknx.io.gateway_scanner import GatewayDescriptor, GatewayScanner +from xknx.io.self_description import request_description from xknx.secure import load_keyring from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow @@ -204,8 +205,11 @@ async def async_step_tunnel(self, user_input: dict | None = None) -> FlowResult: return await self.async_step_manual_tunnel() errors: dict = {} - tunnel_options = [str(tunnel) for tunnel in self._found_tunnels] - tunnel_options.append(OPTION_MANUAL_TUNNEL) + tunnel_options = { + str(tunnel): f"{tunnel}{' 🔐' if tunnel.tunnelling_requires_secure else ''}" + for tunnel in self._found_tunnels + } + tunnel_options |= {OPTION_MANUAL_TUNNEL: OPTION_MANUAL_TUNNEL} fields = {vol.Required(CONF_KNX_GATEWAY): vol.In(tunnel_options)} return self.async_show_form( @@ -230,17 +234,38 @@ async def async_step_manual_tunnel( except vol.Invalid: errors[CONF_KNX_LOCAL_IP] = "invalid_ip_address" + selected_tunnelling_type = user_input[CONF_KNX_TUNNELING_TYPE] + if not errors: + try: + self._selected_tunnel = await request_description( + gateway_ip=_host, + gateway_port=user_input[CONF_PORT], + local_ip=_local_ip, + route_back=user_input[CONF_KNX_ROUTE_BACK], + ) + except CommunicationError: + errors["base"] = "cannot_connect" + else: + if bool(self._selected_tunnel.tunnelling_requires_secure) is not ( + selected_tunnelling_type == CONF_KNX_TUNNELING_TCP_SECURE + ): + errors[CONF_KNX_TUNNELING_TYPE] = "unsupported_tunnel_type" + elif ( + selected_tunnelling_type == CONF_KNX_TUNNELING_TCP + and not self._selected_tunnel.supports_tunnelling_tcp + ): + errors[CONF_KNX_TUNNELING_TYPE] = "unsupported_tunnel_type" + if not errors: - connection_type = user_input[CONF_KNX_TUNNELING_TYPE] self.new_entry_data = KNXConfigEntryData( + connection_type=selected_tunnelling_type, host=_host, port=user_input[CONF_PORT], route_back=user_input[CONF_KNX_ROUTE_BACK], local_ip=_local_ip, - connection_type=connection_type, ) - if connection_type == CONF_KNX_TUNNELING_TCP_SECURE: + if selected_tunnelling_type == CONF_KNX_TUNNELING_TCP_SECURE: return self.async_show_menu( step_id="secure_key_source", menu_options=["secure_knxkeys", "secure_routing_manual"], @@ -299,7 +324,7 @@ async def async_step_manual_tunnel( if self.show_advanced_options: fields[vol.Optional(CONF_KNX_LOCAL_IP)] = _IP_SELECTOR - if not self._found_tunnels: + if not self._found_tunnels and not errors.get("base"): errors["base"] = "no_tunnel_discovered" return self.async_show_form( step_id="manual_tunnel", data_schema=vol.Schema(fields), errors=errors diff --git a/homeassistant/components/knx/strings.json b/homeassistant/components/knx/strings.json index 0fbb7fc3ae6f3..632af9961dcd2 100644 --- a/homeassistant/components/knx/strings.json +++ b/homeassistant/components/knx/strings.json @@ -99,7 +99,8 @@ "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/", "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." + "no_tunnel_discovered": "Could not find a KNX tunneling server on your network.", + "unsupported_tunnel_type": "Selected tunnelling type not supported by gateway." } }, "options": { @@ -214,7 +215,8 @@ "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%]" + "no_tunnel_discovered": "[%key:component::knx::config::error::no_tunnel_discovered%]", + "unsupported_tunnel_type": "[%key:component::knx::config::error::unsupported_tunnel_type%]" } } } diff --git a/tests/components/knx/test_config_flow.py b/tests/components/knx/test_config_flow.py index 9d6eaf68c5431..55f7b2a889171 100644 --- a/tests/components/knx/test_config_flow.py +++ b/tests/components/knx/test_config_flow.py @@ -1,8 +1,8 @@ """Test the KNX config flow.""" -from unittest.mock import patch +from unittest.mock import Mock, patch import pytest -from xknx.exceptions.exception import InvalidSecureConfiguration +from xknx.exceptions.exception import CommunicationError, InvalidSecureConfiguration from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT from xknx.io.gateway_scanner import GatewayDescriptor @@ -441,7 +441,11 @@ async def test_routing_secure_keyfile( return_value=GatewayScannerMock(), ) async def test_tunneling_setup_manual( - gateway_scanner_mock, hass: HomeAssistant, knx_setup, user_input, config_entry_data + _gateway_scanner_mock, + hass: HomeAssistant, + knx_setup, + user_input, + config_entry_data, ) -> None: """Test tunneling if no gateway was found found (or `manual` option was chosen).""" result = await hass.config_entries.flow.async_init( @@ -460,11 +464,21 @@ async def test_tunneling_setup_manual( assert result2["step_id"] == "manual_tunnel" assert result2["errors"] == {"base": "no_tunnel_discovered"} - result3 = await hass.config_entries.flow.async_configure( - result2["flow_id"], - user_input, - ) - await hass.async_block_till_done() + with patch( + "homeassistant.components.knx.config_flow.request_description", + return_value=_gateway_descriptor( + user_input[CONF_HOST], + user_input[CONF_PORT], + supports_tunnelling_tcp=( + user_input[CONF_KNX_TUNNELING_TYPE] == CONF_KNX_TUNNELING_TCP + ), + ), + ): + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + user_input, + ) + await hass.async_block_till_done() assert result3["type"] == FlowResultType.CREATE_ENTRY assert result3["title"] == "Tunneling @ 192.168.0.1" assert result3["data"] == config_entry_data @@ -475,8 +489,146 @@ async def test_tunneling_setup_manual( "homeassistant.components.knx.config_flow.GatewayScanner", return_value=GatewayScannerMock(), ) +async def test_tunneling_setup_manual_request_description_error( + _gateway_scanner_mock, + hass: HomeAssistant, + knx_setup, +) -> None: + """Test tunneling if no gateway was found found (or `manual` option was chosen).""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, + }, + ) + assert result["step_id"] == "manual_tunnel" + assert result["errors"] == {"base": "no_tunnel_discovered"} + + # TCP configured but not supported by gateway + with patch( + "homeassistant.components.knx.config_flow.request_description", + return_value=_gateway_descriptor( + "192.168.0.1", + 3671, + supports_tunnelling_tcp=False, + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3671, + }, + ) + assert result["step_id"] == "manual_tunnel" + assert result["errors"] == { + "base": "no_tunnel_discovered", + "tunneling_type": "unsupported_tunnel_type", + } + # TCP configured but Secure required by gateway + with patch( + "homeassistant.components.knx.config_flow.request_description", + return_value=_gateway_descriptor( + "192.168.0.1", + 3671, + supports_tunnelling_tcp=True, + requires_secure=True, + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3671, + }, + ) + assert result["step_id"] == "manual_tunnel" + assert result["errors"] == { + "base": "no_tunnel_discovered", + "tunneling_type": "unsupported_tunnel_type", + } + # Secure configured but not enabled on gateway + with patch( + "homeassistant.components.knx.config_flow.request_description", + return_value=_gateway_descriptor( + "192.168.0.1", + 3671, + supports_tunnelling_tcp=True, + requires_secure=False, + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3671, + }, + ) + assert result["step_id"] == "manual_tunnel" + assert result["errors"] == { + "base": "no_tunnel_discovered", + "tunneling_type": "unsupported_tunnel_type", + } + # No connection to gateway + with patch( + "homeassistant.components.knx.config_flow.request_description", + side_effect=CommunicationError(""), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3671, + }, + ) + assert result["step_id"] == "manual_tunnel" + assert result["errors"] == {"base": "cannot_connect"} + # OK configuration + with patch( + "homeassistant.components.knx.config_flow.request_description", + return_value=_gateway_descriptor( + "192.168.0.1", + 3671, + supports_tunnelling_tcp=True, + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3671, + }, + ) + await hass.async_block_till_done() + assert result["type"] == FlowResultType.CREATE_ENTRY + assert result["title"] == "Tunneling @ 192.168.0.1" + assert result["data"] == { + **DEFAULT_ENTRY_DATA, + CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP, + CONF_HOST: "192.168.0.1", + CONF_PORT: 3671, + } + knx_setup.assert_called_once() + + +@patch( + "homeassistant.components.knx.config_flow.GatewayScanner", + return_value=GatewayScannerMock(), +) +@patch( + "homeassistant.components.knx.config_flow.request_description", + return_value=_gateway_descriptor("192.168.0.2", 3675), +) async def test_tunneling_setup_for_local_ip( - gateway_scanner_mock, hass: HomeAssistant, knx_setup + _request_description_mock, _gateway_scanner_mock, hass: HomeAssistant, knx_setup ) -> None: """Test tunneling if only one gateway is found.""" result = await hass.config_entries.flow.async_init( @@ -715,7 +867,17 @@ async def _get_menu_step(hass: HomeAssistant) -> FlowResult: return result3 +@patch( + "homeassistant.components.knx.config_flow.request_description", + return_value=_gateway_descriptor( + "192.168.0.1", + 3675, + supports_tunnelling_tcp=True, + requires_secure=True, + ), +) async def test_get_secure_menu_step_manual_tunnelling( + _request_description_mock, hass: HomeAssistant, ): """Test flow reaches secure_tunnellinn menu step from manual tunnelling configuration.""" @@ -908,6 +1070,7 @@ async def test_options_flow_connection_type( gateway = _gateway_descriptor("192.168.0.1", 3675) await hass.config_entries.async_setup(mock_config_entry.entry_id) + hass.data[DOMAIN] = Mock() # GatewayScanner uses running XKNX() in options flow menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id) with patch(