"""Test the KNX config flow.""" import asyncio from collections.abc import Mapping from contextlib import contextmanager from unittest.mock import AsyncMock, MagicMock, Mock, patch from knx_telegram_store.connection import ConnectionCheckResult, ConnectionErrorKind import pytest from xknx.exceptions import XKNXException 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 from homeassistant import config_entries from homeassistant.components.knx.config_flow import ( CONF_KEYRING_FILE, CONF_KNX_GATEWAY, CONF_KNX_TELEGRAM_STORE_SECTION, CONF_KNX_TUNNELING_TYPE, DEFAULT_ENTRY_DATA, DEFAULT_ENTRY_OPTIONS, OPTION_MANUAL_TUNNEL, _build_dsn, _parse_dsn, ) from homeassistant.components.knx.const import ( CONF_KNX_AUTOMATIC, CONF_KNX_CONNECTION_TYPE, CONF_KNX_INDIVIDUAL_ADDRESS, CONF_KNX_KNXKEY_FILENAME, CONF_KNX_KNXKEY_PASSWORD, CONF_KNX_LOCAL_IP, CONF_KNX_MCAST_GRP, CONF_KNX_MCAST_PORT, CONF_KNX_RATE_LIMIT, CONF_KNX_ROUTE_BACK, CONF_KNX_ROUTING, CONF_KNX_ROUTING_BACKBONE_KEY, CONF_KNX_ROUTING_SECURE, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE, CONF_KNX_SECURE_DEVICE_AUTHENTICATION, CONF_KNX_SECURE_USER_ID, CONF_KNX_SECURE_USER_PASSWORD, CONF_KNX_STATE_UPDATER, CONF_KNX_TELEGRAM_DB_BACKEND, CONF_KNX_TELEGRAM_DB_LOAD_HOURS, CONF_KNX_TELEGRAM_DB_POSTGRES_DSN, CONF_KNX_TELEGRAM_DB_RETENTION_DAYS, CONF_KNX_TUNNEL_ENDPOINT_IA, CONF_KNX_TUNNELING, CONF_KNX_TUNNELING_TCP, CONF_KNX_TUNNELING_TCP_SECURE, DOMAIN, KNX_TELEGRAM_BACKEND_POSTGRES, KNX_TELEGRAM_BACKEND_SQLITE, KNX_TELEGRAM_DB_RETENTION_DEFAULT, KNX_TELEGRAM_LOAD_HOURS_DEFAULT, ) from homeassistant.const import CONF_HOST, CONF_PORT from homeassistant.core import HomeAssistant from homeassistant.data_entry_flow import FlowResultType from tests.common import MockConfigEntry, get_fixture_path FIXTURE_KNXKEYS_PASSWORD = "test" FIXTURE_KEYRING = sync_load_keyring( get_fixture_path("fixture.knxkeys", DOMAIN), FIXTURE_KNXKEYS_PASSWORD ) FIXTURE_UPLOAD_UUID = "0123456789abcdef0123456789abcdef" GATEWAY_INDIVIDUAL_ADDRESS = IndividualAddress("1.0.0") async def _mock_validate_ip_for_invalid_local(ip_address: str) -> str: if ip_address in {"no_local_ip", "asdf"}: raise XKNXException return ip_address def _assert_mock_entry_data( mock_entry: MockConfigEntry, expected_data: dict, expected_options: Mapping | None = None, ) -> None: """Assert the config entry stores connection data and options separately.""" if expected_options is None: expected_options = DEFAULT_ENTRY_OPTIONS assert dict(mock_entry.data) == expected_data assert dict(mock_entry.options) == expected_options @pytest.fixture(name="knx_setup") async def fixture_knx_setup(hass: HomeAssistant): """Mock KNX entry setup.""" with ( patch("homeassistant.components.knx.async_setup", return_value=True), patch( "homeassistant.components.knx.async_setup_entry", return_value=True ) as mock_async_setup_entry, ): yield mock_async_setup_entry @contextmanager def patch_file_upload(return_value=FIXTURE_KEYRING, side_effect=None): """Patch file upload. Yields the Keyring instance (return_value).""" with ( patch( "homeassistant.components.knx.storage.keyring.process_uploaded_file" ) as file_upload_mock, patch( "homeassistant.components.knx.storage.keyring.sync_load_keyring", return_value=return_value, side_effect=side_effect, ), patch( "pathlib.Path.mkdir", ) as mkdir_mock, patch( "shutil.move", ) as shutil_move_mock, ): file_upload_mock.return_value.__enter__.return_value = Mock() yield return_value if side_effect: mkdir_mock.assert_not_called() shutil_move_mock.assert_not_called() else: mkdir_mock.assert_called_once() shutil_move_mock.assert_called_once() def _gateway_descriptor( ip: str, port: int, supports_tunnelling_tcp: bool = False, requires_secure: bool = False, slots: bool = True, ) -> GatewayDescriptor: """Get mock gw descriptor.""" descriptor = GatewayDescriptor( name="Test", individual_address=GATEWAY_INDIVIDUAL_ADDRESS, ip_addr=ip, port=port, local_interface="eth0", local_ip="127.0.0.1", supports_routing=True, supports_tunnelling=True, supports_tunnelling_tcp=supports_tunnelling_tcp, ) 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 class GatewayScannerMock: """Mock GatewayScanner.""" def __init__(self, gateways=None) -> None: """Initialize GatewayScannerMock.""" # Key is a HPAI instance in xknx, but not used in HA anyway. self.found_gateways = ( {f"{gateway.ip_addr}:{gateway.port}": gateway for gateway in gateways} if gateways else {} ) async def async_scan(self): """Mock async generator.""" for gateway in self.found_gateways: yield gateway async def test_user_single_instance(hass: HomeAssistant) -> None: """Test we only allow a single config flow.""" MockConfigEntry(domain=DOMAIN).add_to_hass(hass) result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER} ) assert result["type"] is FlowResultType.ABORT assert result["reason"] == "single_instance_allowed" @patch( "homeassistant.components.knx.config_flow.GatewayScanner", return_value=GatewayScannerMock(), ) async def test_routing_setup( gateway_scanner_mock, hass: HomeAssistant, knx_setup ) -> None: """Test routing setup.""" 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"] result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "routing" assert result["errors"] == {"base": "no_router_discovered"} result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110", }, ) assert result["type"] is FlowResultType.CREATE_ENTRY assert result["title"] == "Routing as 1.1.110" assert result["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_PORT: 3675, CONF_KNX_LOCAL_IP: None, CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110", CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, CONF_KNX_TUNNEL_ENDPOINT_IA: None, } assert result["options"] == DEFAULT_ENTRY_OPTIONS knx_setup.assert_called_once() @patch( "homeassistant.components.knx.config_flow.GatewayScanner", return_value=GatewayScannerMock(), ) async def test_routing_setup_with_local_ip( gateway_scanner_mock, hass: HomeAssistant, knx_setup ) -> None: """Test routing setup where user specifies a local IP.""" 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"] result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "routing" assert result["errors"] == {"base": "no_router_discovered"} # invalid user input with patch( "homeassistant.components.knx.config_flow.xknx_validate_ip", new=AsyncMock(side_effect=_mock_validate_ip_for_invalid_local), ): result_invalid_input = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_MCAST_GRP: "10.1.2.3", # no valid multicast group CONF_KNX_MCAST_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "not_a_valid_address", CONF_KNX_LOCAL_IP: "no_local_ip", }, ) assert result_invalid_input["type"] is FlowResultType.FORM assert result_invalid_input["step_id"] == "routing" assert result_invalid_input["errors"] == { CONF_KNX_MCAST_GRP: "invalid_ip_address", CONF_KNX_INDIVIDUAL_ADDRESS: "invalid_individual_address", CONF_KNX_LOCAL_IP: "invalid_ip_address", "base": "no_router_discovered", } # valid user input result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110", CONF_KNX_LOCAL_IP: "192.168.1.112", }, ) assert result["type"] is FlowResultType.CREATE_ENTRY assert result["title"] == "Routing as 1.1.110" assert result["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_PORT: 3675, CONF_KNX_LOCAL_IP: "192.168.1.112", CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110", CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, CONF_KNX_TUNNEL_ENDPOINT_IA: None, } assert result["options"] == DEFAULT_ENTRY_OPTIONS knx_setup.assert_called_once() @patch( "homeassistant.components.knx.config_flow.GatewayScanner", return_value=GatewayScannerMock(), ) async def test_routing_secure_manual_setup( gateway_scanner_mock, hass: HomeAssistant, knx_setup ) -> None: """Test routing secure setup with manual key config.""" 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"] result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "routing" assert result["errors"] == {"base": "no_router_discovered"} result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_PORT: 3671, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.123", CONF_KNX_ROUTING_SECURE: True, }, ) assert result["type"] is FlowResultType.MENU assert result["step_id"] == "secure_key_source_menu_routing" result = await hass.config_entries.flow.async_configure( result["flow_id"], {"next_step_id": "secure_routing_manual"}, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "secure_routing_manual" assert not result["errors"] result_invalid_key1 = await hass.config_entries.flow.async_configure( result["flow_id"], { # invalid hex string CONF_KNX_ROUTING_BACKBONE_KEY: "xxaacc44bbaacc44bbaacc44bbaaccyy", CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000, }, ) assert result_invalid_key1["type"] is FlowResultType.FORM assert result_invalid_key1["step_id"] == "secure_routing_manual" assert result_invalid_key1["errors"] == {"backbone_key": "invalid_backbone_key"} result_invalid_key2 = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44", # invalid length CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000, }, ) assert result_invalid_key2["type"] is FlowResultType.FORM assert result_invalid_key2["step_id"] == "secure_routing_manual" assert result_invalid_key2["errors"] == {"backbone_key": "invalid_backbone_key"} secure_routing_manual = await hass.config_entries.flow.async_configure( result_invalid_key2["flow_id"], { CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44bbaacc44bbaacc44", CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000, }, ) assert secure_routing_manual["type"] is FlowResultType.CREATE_ENTRY assert secure_routing_manual["title"] == "Secure Routing as 0.0.123" assert secure_routing_manual["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING_SECURE, CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44bbaacc44bbaacc44", CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.123", CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, CONF_KNX_TUNNEL_ENDPOINT_IA: None, } assert secure_routing_manual["options"] == DEFAULT_ENTRY_OPTIONS knx_setup.assert_called_once() @patch( "homeassistant.components.knx.config_flow.GatewayScanner", return_value=GatewayScannerMock(), ) async def test_routing_secure_keyfile( gateway_scanner_mock, hass: HomeAssistant, knx_setup ) -> None: """Test routing secure setup with keyfile.""" 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"] result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "routing" assert result["errors"] == {"base": "no_router_discovered"} result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_PORT: 3671, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.123", CONF_KNX_ROUTING_SECURE: True, }, ) assert result["type"] is FlowResultType.MENU assert result["step_id"] == "secure_key_source_menu_routing" result = await hass.config_entries.flow.async_configure( result["flow_id"], {"next_step_id": "secure_knxkeys"}, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "secure_knxkeys" assert not result["errors"] with patch_file_upload(): routing_secure_knxkeys = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KNX_KNXKEY_PASSWORD: "password", }, ) assert routing_secure_knxkeys["type"] is FlowResultType.CREATE_ENTRY assert routing_secure_knxkeys["title"] == "Secure Routing as 0.0.123" assert routing_secure_knxkeys["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING_SECURE, CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys", CONF_KNX_KNXKEY_PASSWORD: "password", CONF_KNX_ROUTING_BACKBONE_KEY: None, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None, CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, CONF_KNX_TUNNEL_ENDPOINT_IA: None, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.123", } assert routing_secure_knxkeys["options"] == DEFAULT_ENTRY_OPTIONS knx_setup.assert_called_once() @pytest.mark.parametrize( ("user_input", "title", "config_entry_data"), [ ( { CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING, CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_ROUTE_BACK: False, }, "Tunneling UDP @ 192.168.0.1", { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: None, CONF_KNX_TUNNEL_ENDPOINT_IA: None, CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, }, ), ( { CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP, CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_ROUTE_BACK: False, }, "Tunneling TCP @ 192.168.0.1", { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP, CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: None, CONF_KNX_TUNNEL_ENDPOINT_IA: None, CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, }, ), ( { CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING, CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_ROUTE_BACK: True, }, "Tunneling UDP @ 192.168.0.1", { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: True, CONF_KNX_LOCAL_IP: None, CONF_KNX_TUNNEL_ENDPOINT_IA: None, CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, }, ), ], ) @patch( "homeassistant.components.knx.config_flow.GatewayScanner", return_value=GatewayScannerMock(), ) async def test_tunneling_setup_manual( gateway_scanner_mock: MagicMock, hass: HomeAssistant, knx_setup, user_input, title, 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( DOMAIN, context={"source": config_entries.SOURCE_USER} ) assert result["type"] is FlowResultType.FORM assert not result["errors"] result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "manual_tunnel" assert result["errors"] == {"base": "no_tunnel_discovered"} 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 ), ), ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input, ) assert result["type"] is FlowResultType.CREATE_ENTRY assert result["title"] == title assert result["data"] == config_entry_data assert result["options"] == DEFAULT_ENTRY_OPTIONS knx_setup.assert_called_once() @patch( "homeassistant.components.knx.config_flow.GatewayScanner", return_value=GatewayScannerMock(), ) async def test_tunneling_setup_manual_request_description_error( gateway_scanner_mock: MagicMock, 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, }, ) assert result["type"] is FlowResultType.CREATE_ENTRY assert result["title"] == "Tunneling TCP @ 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, CONF_KNX_TUNNEL_ENDPOINT_IA: None, CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, } assert result["options"] == DEFAULT_ENTRY_OPTIONS 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( request_description_mock: MagicMock, gateway_scanner_mock: MagicMock, hass: HomeAssistant, knx_setup, ) -> None: """Test tunneling if only one gateway is found.""" 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"] result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "manual_tunnel" assert result["errors"] == {"base": "no_tunnel_discovered"} # invalid host ip address result_invalid_host = await hass.config_entries.flow.async_configure( result["flow_id"], { 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", }, ) assert result_invalid_host["type"] is FlowResultType.FORM assert result_invalid_host["step_id"] == "manual_tunnel" assert result_invalid_host["errors"] == { CONF_HOST: "invalid_ip_address", "base": "no_tunnel_discovered", } # invalid local ip address with patch( "homeassistant.components.knx.config_flow.xknx_validate_ip", new=AsyncMock(side_effect=_mock_validate_ip_for_invalid_local), ): result_invalid_local = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING, CONF_HOST: "192.168.0.2", CONF_PORT: 3675, CONF_KNX_LOCAL_IP: "asdf", }, ) assert result_invalid_local["type"] is FlowResultType.FORM assert result_invalid_local["step_id"] == "manual_tunnel" assert result_invalid_local["errors"] == { CONF_KNX_LOCAL_IP: "invalid_ip_address", "base": "no_tunnel_discovered", } # valid user input result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING, CONF_HOST: "192.168.0.2", CONF_PORT: 3675, CONF_KNX_LOCAL_IP: "192.168.1.112", }, ) assert result["type"] is FlowResultType.CREATE_ENTRY assert result["title"] == "Tunneling UDP @ 192.168.0.2" assert result["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_HOST: "192.168.0.2", CONF_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: "192.168.1.112", CONF_KNX_TUNNEL_ENDPOINT_IA: None, CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, } assert result["options"] == DEFAULT_ENTRY_OPTIONS knx_setup.assert_called_once() async def test_tunneling_setup_for_multiple_found_gateways( hass: HomeAssistant, knx_setup ) -> None: """Test tunneling if multiple gateways are found.""" 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_udp, 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"] result = await hass.config_entries.flow.async_configure( tunnel_flow["flow_id"], {CONF_KNX_GATEWAY: str(gateway_udp)}, ) assert result["type"] is FlowResultType.CREATE_ENTRY assert result["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_HOST: "192.168.0.1", 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, } assert result["options"] == DEFAULT_ENTRY_OPTIONS 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, } assert result["options"] == DEFAULT_ENTRY_OPTIONS 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, } assert result["options"] == DEFAULT_ENTRY_OPTIONS knx_setup.assert_called_once() @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( "homeassistant.components.knx.config_flow.GatewayScanner" ) as gateway_scanner_mock: gateway_scanner_mock.return_value = GatewayScannerMock([gateway]) 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"] manual_tunnel_flow = await hass.config_entries.flow.async_configure( tunnel_flow["flow_id"], { CONF_KNX_GATEWAY: OPTION_MANUAL_TUNNEL, }, ) assert manual_tunnel_flow["type"] is 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, knx_setup ) -> None: """Test we get the form.""" with patch( "homeassistant.components.knx.config_flow.GatewayScanner" ) as gateway_scanner_mock: gateway_scanner_mock.return_value = GatewayScannerMock( [_gateway_descriptor("192.168.0.1", 3675)] ) 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"] result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, }, ) assert result["type"] is FlowResultType.CREATE_ENTRY assert result["title"] == CONF_KNX_AUTOMATIC.capitalize() # don't use the DEFAULT_ENTRY_* constants here to check for correct usage of defaults assert result["data"] == { CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_LOCAL_IP: None, CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_ROUTE_BACK: False, CONF_KNX_TUNNEL_ENDPOINT_IA: None, } assert result["options"] == { CONF_KNX_RATE_LIMIT: 0, CONF_KNX_STATE_UPDATER: True, CONF_KNX_TELEGRAM_DB_RETENTION_DAYS: KNX_TELEGRAM_DB_RETENTION_DEFAULT, CONF_KNX_TELEGRAM_DB_LOAD_HOURS: KNX_TELEGRAM_LOAD_HOURS_DEFAULT, CONF_KNX_TELEGRAM_DB_BACKEND: KNX_TELEGRAM_BACKEND_SQLITE, } knx_setup.assert_called_once() async def _get_menu_step_secure_tunnel( hass: HomeAssistant, ) -> config_entries.ConfigFlowResult: """Return flow in secure_tunnel menu step.""" gateway = _gateway_descriptor( "192.168.0.1", 3675, supports_tunnelling_tcp=True, requires_secure=True, ) with patch( "homeassistant.components.knx.config_flow.GatewayScanner" ) as gateway_scanner_mock: gateway_scanner_mock.return_value = GatewayScannerMock([gateway]) 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"] result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "tunnel" assert not result["errors"] result = await hass.config_entries.flow.async_configure( result["flow_id"], {CONF_KNX_GATEWAY: str(gateway)}, ) assert result["type"] is FlowResultType.MENU assert result["step_id"] == "secure_key_source_menu_tunnel" return result @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: MagicMock, hass: HomeAssistant, ) -> None: """Test flow reaches secure_tunnelling menu from manual config.""" gateway = _gateway_descriptor( "192.168.0.1", 3675, supports_tunnelling_tcp=True, requires_secure=True, ) with patch( "homeassistant.components.knx.config_flow.GatewayScanner" ) as gateway_scanner_mock: gateway_scanner_mock.return_value = GatewayScannerMock([gateway]) 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"] result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "tunnel" assert not result["errors"] manual_tunnel_flow = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_GATEWAY: OPTION_MANUAL_TUNNEL, }, ) result = 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, }, ) assert result["type"] is FlowResultType.MENU assert result["step_id"] == "secure_key_source_menu_tunnel" async def test_configure_secure_tunnel_manual(hass: HomeAssistant, knx_setup) -> None: """Test configure tunneling secure keys manually.""" menu_step = await _get_menu_step_secure_tunnel(hass) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], {"next_step_id": "secure_tunnel_manual"}, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "secure_tunnel_manual" assert not result["errors"] secure_tunnel_manual = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_SECURE_USER_ID: 2, CONF_KNX_SECURE_USER_PASSWORD: "password", CONF_KNX_SECURE_DEVICE_AUTHENTICATION: "device_auth", }, ) assert secure_tunnel_manual["type"] is 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, CONF_KNX_SECURE_USER_PASSWORD: "password", CONF_KNX_SECURE_DEVICE_AUTHENTICATION: "device_auth", CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: None, CONF_KNX_TUNNEL_ENDPOINT_IA: None, } assert secure_tunnel_manual["options"] == DEFAULT_ENTRY_OPTIONS knx_setup.assert_called_once() async def test_configure_secure_knxkeys(hass: HomeAssistant, knx_setup) -> None: """Test configure secure knxkeys.""" menu_step = await _get_menu_step_secure_tunnel(hass) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], {"next_step_id": "secure_knxkeys"}, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "secure_knxkeys" assert not result["errors"] with patch_file_upload(): secure_knxkeys = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KNX_KNXKEY_PASSWORD: "test", }, ) assert result["type"] is FlowResultType.FORM assert secure_knxkeys["step_id"] == "knxkeys_tunnel_select" assert not result["errors"] secure_knxkeys = await hass.config_entries.flow.async_configure( secure_knxkeys["flow_id"], {CONF_KNX_TUNNEL_ENDPOINT_IA: CONF_KNX_AUTOMATIC}, ) assert secure_knxkeys["type"] is FlowResultType.CREATE_ENTRY assert secure_knxkeys["data"] == { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys", CONF_KNX_KNXKEY_PASSWORD: "test", CONF_KNX_ROUTING_BACKBONE_KEY: None, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None, CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, CONF_KNX_TUNNEL_ENDPOINT_IA: None, CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: None, } assert secure_knxkeys["options"] == DEFAULT_ENTRY_OPTIONS knx_setup.assert_called_once() async def test_configure_secure_knxkeys_invalid_signature(hass: HomeAssistant) -> None: """Test configure secure knxkeys but file was not found.""" menu_step = await _get_menu_step_secure_tunnel(hass) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], {"next_step_id": "secure_knxkeys"}, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "secure_knxkeys" assert not result["errors"] with patch_file_upload( side_effect=InvalidSecureConfiguration(), ): secure_knxkeys = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KNX_KNXKEY_PASSWORD: "password", }, ) assert secure_knxkeys["type"] is FlowResultType.FORM assert secure_knxkeys["errors"] assert ( secure_knxkeys["errors"][CONF_KNX_KNXKEY_PASSWORD] == "keyfile_invalid_signature" ) async def test_configure_secure_knxkeys_no_tunnel_for_host(hass: HomeAssistant) -> None: """Test configure secure knxkeys but file was not found.""" menu_step = await _get_menu_step_secure_tunnel(hass) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], {"next_step_id": "secure_knxkeys"}, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "secure_knxkeys" assert not result["errors"] with patch_file_upload(return_value=Mock()) as mock_keyring: mock_keyring.get_tunnel_interfaces_by_host.return_value = [] secure_knxkeys = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KNX_KNXKEY_PASSWORD: "password", }, ) assert secure_knxkeys["type"] is FlowResultType.FORM assert secure_knxkeys["errors"] == {"base": "keyfile_no_tunnel_for_host"} async def test_reconfigure_flow_connection_type( hass: HomeAssistant, knx, mock_config_entry: MockConfigEntry ) -> None: """Test reconfigure flow changing interface.""" # run one flow test with a set up integration (knx fixture) # instead of mocking async_setup_entry (knx_setup fixture) to test # usage of the already running XKNX instance for gateway scanner gateway = _gateway_descriptor("192.168.0.1", 3675) await knx.setup_integration() menu_step = await knx.mock_config_entry.start_reconfigure_flow(hass) with patch( "homeassistant.components.knx.config_flow.GatewayScanner" ) as gateway_scanner_mock: gateway_scanner_mock.return_value = GatewayScannerMock([gateway]) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], {"next_step_id": "connection_type"}, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "connection_type" result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={ CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "tunnel" result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={ CONF_KNX_GATEWAY: str(gateway), }, ) assert result["type"] is FlowResultType.ABORT assert result["reason"] == "reconfigure_successful" _assert_mock_entry_data( mock_config_entry, { CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, 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, }, ) async def test_reconfigure_flow_secure_manual_to_keyfile( hass: HomeAssistant, knx_setup ) -> None: """Test reconfigure flow changing secure credential source.""" mock_config_entry = MockConfigEntry( title="KNX", domain=DOMAIN, version=2, data={ **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, CONF_KNX_SECURE_USER_ID: 2, CONF_KNX_SECURE_USER_PASSWORD: "password", CONF_KNX_SECURE_DEVICE_AUTHENTICATION: "device_auth", CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys", CONF_KNX_KNXKEY_PASSWORD: "invalid_password", CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: None, }, options=DEFAULT_ENTRY_OPTIONS, ) gateway = _gateway_descriptor( "192.168.0.1", 3675, supports_tunnelling_tcp=True, requires_secure=True, ) mock_config_entry.add_to_hass(hass) await hass.config_entries.async_setup(mock_config_entry.entry_id) knx_setup.reset_mock() menu_step = await mock_config_entry.start_reconfigure_flow(hass) with patch( "homeassistant.components.knx.config_flow.GatewayScanner" ) as gateway_scanner_mock: gateway_scanner_mock.return_value = GatewayScannerMock([gateway]) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], {"next_step_id": "connection_type"}, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "connection_type" result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={ CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "tunnel" assert not result["errors"] result = await hass.config_entries.flow.async_configure( result["flow_id"], {CONF_KNX_GATEWAY: str(gateway)}, ) assert result["type"] is FlowResultType.MENU assert result["step_id"] == "secure_key_source_menu_tunnel" result = await hass.config_entries.flow.async_configure( result["flow_id"], {"next_step_id": "secure_knxkeys"}, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "secure_knxkeys" assert not result["errors"] with patch_file_upload(): secure_knxkeys = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KNX_KNXKEY_PASSWORD: "test", }, ) assert result["type"] is FlowResultType.FORM assert secure_knxkeys["step_id"] == "knxkeys_tunnel_select" assert not result["errors"] secure_knxkeys = await hass.config_entries.flow.async_configure( secure_knxkeys["flow_id"], {CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1"}, ) assert secure_knxkeys["type"] is FlowResultType.ABORT assert secure_knxkeys["reason"] == "reconfigure_successful" _assert_mock_entry_data( mock_config_entry, { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys", CONF_KNX_KNXKEY_PASSWORD: "test", CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1", CONF_KNX_ROUTING_BACKBONE_KEY: None, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None, CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: None, }, ) knx_setup.assert_called_once() async def test_reconfigure_flow_routing(hass: HomeAssistant, knx_setup) -> None: """Test reconfigure flow changing routing settings.""" mock_config_entry = MockConfigEntry( title="KNX", domain=DOMAIN, version=2, data={ **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, }, options=DEFAULT_ENTRY_OPTIONS, ) 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) knx_setup.reset_mock() menu_step = await mock_config_entry.start_reconfigure_flow(hass) with patch( "homeassistant.components.knx.config_flow.GatewayScanner" ) as gateway_scanner_mock: gateway_scanner_mock.return_value = GatewayScannerMock([gateway]) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], {"next_step_id": "connection_type"}, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "connection_type" result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={ CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "routing" assert result["errors"] == {} result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KNX_INDIVIDUAL_ADDRESS: "2.0.4", }, ) assert result["type"] is FlowResultType.ABORT assert result["reason"] == "reconfigure_successful" _assert_mock_entry_data( mock_config_entry, { **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_reconfigure_update_keyfile(hass: HomeAssistant, knx_setup) -> None: """Test reconfigure flow updating keyfile when tunnel is configured.""" start_data = { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, CONF_KNX_SECURE_USER_ID: 2, CONF_KNX_SECURE_USER_PASSWORD: "password", CONF_KNX_SECURE_DEVICE_AUTHENTICATION: "device_auth", CONF_KNX_KNXKEY_PASSWORD: "old_password", CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: None, CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1", } mock_config_entry = MockConfigEntry( title="KNX", domain=DOMAIN, version=2, data=start_data, options=DEFAULT_ENTRY_OPTIONS, ) mock_config_entry.add_to_hass(hass) await hass.config_entries.async_setup(mock_config_entry.entry_id) knx_setup.reset_mock() menu_step = await mock_config_entry.start_reconfigure_flow(hass) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], {"next_step_id": "secure_knxkeys"}, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "secure_knxkeys" with patch_file_upload(): result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KNX_KNXKEY_PASSWORD: "password", }, ) assert result["type"] is FlowResultType.ABORT assert result["reason"] == "reconfigure_successful" _assert_mock_entry_data( mock_config_entry, { **start_data, CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys", CONF_KNX_KNXKEY_PASSWORD: "password", CONF_KNX_ROUTING_BACKBONE_KEY: None, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None, }, ) knx_setup.assert_called_once() async def test_reconfigure_keyfile_upload(hass: HomeAssistant, knx_setup) -> None: """Test reconfigure flow uploading a keyfile for the first time.""" start_data = { **DEFAULT_ENTRY_DATA, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP, CONF_HOST: "192.168.0.1", CONF_PORT: 3675, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_ROUTE_BACK: False, CONF_KNX_LOCAL_IP: None, } mock_config_entry = MockConfigEntry( title="KNX", domain=DOMAIN, version=2, data=start_data, options=DEFAULT_ENTRY_OPTIONS, ) mock_config_entry.add_to_hass(hass) await hass.config_entries.async_setup(mock_config_entry.entry_id) knx_setup.reset_mock() menu_step = await mock_config_entry.start_reconfigure_flow(hass) result = await hass.config_entries.flow.async_configure( menu_step["flow_id"], {"next_step_id": "secure_knxkeys"}, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "secure_knxkeys" with patch_file_upload(): result = await hass.config_entries.flow.async_configure( result["flow_id"], { CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KNX_KNXKEY_PASSWORD: "password", }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "knxkeys_tunnel_select" result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={ CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1", }, ) assert result["type"] is FlowResultType.ABORT assert result["reason"] == "reconfigure_successful" _assert_mock_entry_data( mock_config_entry, { **start_data, CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys", CONF_KNX_KNXKEY_PASSWORD: "password", CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1", CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_PASSWORD: None, CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_ROUTING_BACKBONE_KEY: None, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None, }, ) knx_setup.assert_called_once() async def test_options_communication_settings( hass: HomeAssistant, knx_setup, mock_config_entry: MockConfigEntry ) -> None: """Test options flow changing communication settings.""" initial_data = dict(mock_config_entry.data) mock_config_entry.add_to_hass(hass) await hass.config_entries.async_setup(mock_config_entry.entry_id) result = await hass.config_entries.options.async_init(mock_config_entry.entry_id) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "communication_settings" result = await hass.config_entries.options.async_configure( result["flow_id"], user_input={ CONF_KNX_STATE_UPDATER: False, CONF_KNX_RATE_LIMIT: 40, CONF_KNX_TELEGRAM_STORE_SECTION: { CONF_KNX_TELEGRAM_DB_LOAD_HOURS: KNX_TELEGRAM_LOAD_HOURS_DEFAULT, CONF_KNX_TELEGRAM_DB_RETENTION_DAYS: 30, CONF_KNX_TELEGRAM_DB_BACKEND: KNX_TELEGRAM_BACKEND_SQLITE, }, }, ) assert result["type"] is FlowResultType.CREATE_ENTRY assert result["data"] == { CONF_KNX_STATE_UPDATER: False, CONF_KNX_RATE_LIMIT: 40, CONF_KNX_TELEGRAM_DB_RETENTION_DAYS: 30, CONF_KNX_TELEGRAM_DB_LOAD_HOURS: KNX_TELEGRAM_LOAD_HOURS_DEFAULT, CONF_KNX_TELEGRAM_DB_BACKEND: KNX_TELEGRAM_BACKEND_SQLITE, } assert mock_config_entry.data == initial_data assert mock_config_entry.options == { CONF_KNX_STATE_UPDATER: False, CONF_KNX_RATE_LIMIT: 40, CONF_KNX_TELEGRAM_DB_RETENTION_DAYS: 30, CONF_KNX_TELEGRAM_DB_LOAD_HOURS: KNX_TELEGRAM_LOAD_HOURS_DEFAULT, CONF_KNX_TELEGRAM_DB_BACKEND: KNX_TELEGRAM_BACKEND_SQLITE, } assert len(knx_setup.mock_calls) == 2 async def _advance_to_postgres_step( hass: HomeAssistant, flow_id: str, *, retention_days: int = 14 ) -> config_entries.ConfigFlowResult: """Select the PostgreSQL backend and land on its connection step.""" result = await hass.config_entries.options.async_configure( flow_id, user_input={ CONF_KNX_STATE_UPDATER: False, CONF_KNX_RATE_LIMIT: 40, CONF_KNX_TELEGRAM_STORE_SECTION: { CONF_KNX_TELEGRAM_DB_LOAD_HOURS: KNX_TELEGRAM_LOAD_HOURS_DEFAULT, CONF_KNX_TELEGRAM_DB_RETENTION_DAYS: retention_days, CONF_KNX_TELEGRAM_DB_BACKEND: KNX_TELEGRAM_BACKEND_POSTGRES, }, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "telegram_store_postgres" assert not result["errors"] return result async def test_options_telegram_store_postgres( hass: HomeAssistant, knx_setup: AsyncMock, mock_config_entry: MockConfigEntry ) -> None: """Test options flow selecting the PostgreSQL telegram store backend.""" mock_config_entry.add_to_hass(hass) await hass.config_entries.async_setup(mock_config_entry.entry_id) result = await hass.config_entries.options.async_init(mock_config_entry.entry_id) result = await _advance_to_postgres_step(hass, result["flow_id"]) with patch( "knx_telegram_store.backends.postgres.PostgresStore.check_config", return_value=ConnectionCheckResult.success(), ): result = await hass.config_entries.options.async_configure( result["flow_id"], user_input={ "host": "db.local", "port": 5432, "user": "knx", "password": "s3cret", "database": "knx_telegrams", "tls": True, }, ) assert result["type"] is FlowResultType.CREATE_ENTRY assert ( mock_config_entry.options[CONF_KNX_TELEGRAM_DB_BACKEND] == KNX_TELEGRAM_BACKEND_POSTGRES ) assert mock_config_entry.options[CONF_KNX_TELEGRAM_DB_RETENTION_DAYS] == 14 assert ( mock_config_entry.options[CONF_KNX_TELEGRAM_DB_POSTGRES_DSN] == "postgresql://knx:s3cret@db.local:5432/knx_telegrams?sslmode=require" ) assert len(knx_setup.mock_calls) == 2 async def test_options_telegram_store_postgres_reuses_password( hass: HomeAssistant, knx_setup: AsyncMock, mock_config_entry: MockConfigEntry ) -> None: """Test the PostgreSQL store reuses the stored password when left blank.""" existing_dsn = "postgresql://olduser:oldpass@old.host:6543/olddb?sslmode=require" mock_config_entry.add_to_hass(hass) hass.config_entries.async_update_entry( mock_config_entry, options={ **mock_config_entry.options, CONF_KNX_TELEGRAM_DB_POSTGRES_DSN: existing_dsn, }, ) await hass.config_entries.async_setup(mock_config_entry.entry_id) result = await hass.config_entries.options.async_init(mock_config_entry.entry_id) result = await _advance_to_postgres_step(hass, result["flow_id"], retention_days=7) # Submit with an empty password - the existing one (parsed from the DSN) # must be reused. with patch( "knx_telegram_store.backends.postgres.PostgresStore.check_config", return_value=ConnectionCheckResult.success(), ): result = await hass.config_entries.options.async_configure( result["flow_id"], user_input={ "host": "new.host", "port": 5432, "user": "newuser", "password": "", "database": "newdb", "tls": False, }, ) assert result["type"] is FlowResultType.CREATE_ENTRY assert ( mock_config_entry.options[CONF_KNX_TELEGRAM_DB_POSTGRES_DSN] == "postgresql://newuser:oldpass@new.host:5432/newdb" ) assert len(knx_setup.mock_calls) == 2 @pytest.mark.parametrize( ("error_kind", "expected_error"), [ pytest.param(ConnectionErrorKind.AUTH, "invalid_auth", id="invalid_auth"), pytest.param( ConnectionErrorKind.HOST_UNREACHABLE, "host_unreachable", id="host_unreachable", ), ], ) async def test_options_telegram_store_postgres_connection_failure( hass: HomeAssistant, knx_setup: AsyncMock, mock_config_entry: MockConfigEntry, error_kind: ConnectionErrorKind, expected_error: str, ) -> None: """Test the PostgreSQL step maps connection check failures to form errors.""" mock_config_entry.add_to_hass(hass) await hass.config_entries.async_setup(mock_config_entry.entry_id) result = await hass.config_entries.options.async_init(mock_config_entry.entry_id) result = await _advance_to_postgres_step(hass, result["flow_id"]) with patch( "knx_telegram_store.backends.postgres.PostgresStore.check_config", return_value=ConnectionCheckResult.failure(error_kind, "check failed"), ): result = await hass.config_entries.options.async_configure( result["flow_id"], user_input={ "host": "db.local", "port": 5432, "user": "knx", "password": "wrong_password", "database": "knx_telegrams", "tls": True, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "telegram_store_postgres" assert result["errors"] == {"base": expected_error} async def test_options_telegram_store_postgres_timeout( hass: HomeAssistant, knx_setup: AsyncMock, mock_config_entry: MockConfigEntry ) -> None: """Test options flow surfaces a timeout when the connection check hangs.""" async def hanging_check(dsn: str) -> None: await asyncio.Event().wait() mock_config_entry.add_to_hass(hass) await hass.config_entries.async_setup(mock_config_entry.entry_id) result = await hass.config_entries.options.async_init(mock_config_entry.entry_id) result = await _advance_to_postgres_step(hass, result["flow_id"]) with ( patch("homeassistant.components.knx.config_flow.DSN_CHECK_TIMEOUT", 0.05), patch( "knx_telegram_store.backends.postgres.PostgresStore.check_config", side_effect=hanging_check, ), ): result = await hass.config_entries.options.async_configure( result["flow_id"], user_input={ "host": "db.local", "port": 5432, "user": "knx", "password": "s3cret", "database": "knx_telegrams", "tls": True, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "telegram_store_postgres" assert result["errors"] == {"base": "timeout"} async def test_options_telegram_store_postgres_malformed_dsn( hass: HomeAssistant, knx_setup: AsyncMock, mock_config_entry: MockConfigEntry ) -> None: """Test the PostgreSQL step maps a DSN the driver rejects to a form error.""" mock_config_entry.add_to_hass(hass) await hass.config_entries.async_setup(mock_config_entry.entry_id) result = await hass.config_entries.options.async_init(mock_config_entry.entry_id) result = await _advance_to_postgres_step(hass, result["flow_id"]) # An unterminated bracketed IPv6 address makes engine creation # raise ValueError before any connection attempt. result = await hass.config_entries.options.async_configure( result["flow_id"], user_input={ "host": "[::1", "port": 5432, "user": "knx", "password": "s3cret", "database": "knx_telegrams", "tls": False, }, ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "telegram_store_postgres" assert result["errors"] == {"base": "cannot_connect"} @pytest.mark.parametrize( ("dsn", "expected"), [ pytest.param("", {}, id="empty"), # Invalid port makes urlparse.port raise ValueError -> {} pytest.param("postgresql://host:notaport/db", {}, id="invalid_port"), pytest.param( "postgresql://u:p@h:5432/db?sslmode=require", { "user": "u", "password": "p", "host": "h", "port": 5432, "database": "db", "tls": True, }, id="full", ), pytest.param( "postgresql://user%40domain:p%40ss%25word@h:5432/db", { "user": "user@domain", "password": "p@ss%word", "host": "h", "port": 5432, "database": "db", "tls": False, }, id="percent_encoded_credentials", ), pytest.param( "postgresql://u:p@[2001:db8::1]:5432/db", { "user": "u", "password": "p", "host": "2001:db8::1", "port": 5432, "database": "db", "tls": False, }, id="ipv6_host", ), pytest.param( "postgresql://u:p@h:5432/db%3Fquery%23hash", { "user": "u", "password": "p", "host": "h", "port": 5432, "database": "db?query#hash", "tls": False, }, id="percent_encoded_database", ), ], ) def test_parse_dsn(dsn: str, expected: dict) -> None: """Test PostgreSQL DSN parsing, including malformed input.""" assert _parse_dsn(dsn) == expected @pytest.mark.parametrize( ("user", "password", "host", "database"), [ pytest.param("simple", "plain", "localhost", "knx", id="plain"), pytest.param("user@domain", "p@ss", "localhost", "knx", id="at_sign"), pytest.param("user", "p@ss%word", "localhost", "knx", id="percent_sign"), pytest.param( "us:er", "p/a:s@s", "localhost", "knx", id="multiple_special_chars" ), pytest.param("user", "pass", "2001:db8::1", "knx", id="ipv6_host"), pytest.param( "user", "pass", "localhost", "knx?query#hash", id="database_special_chars" ), ], ) def test_dsn_round_trip(user: str, password: str, host: str, database: str) -> None: """Test _build_dsn -> _parse_dsn -> _build_dsn produces identical DSNs. Catches double percent-encoding: urlparse returns percent-encoded values, so _parse_dsn must decode them before they are fed back into _build_dsn. IPv6 hosts must be bracketed in the netloc for the DSN to stay parseable. Database names with URL delimiters are percent-encoded to prevent truncation. """ params = { "user": user, "password": password, "host": host, "port": 5432, "database": database, "tls": False, } dsn1 = _build_dsn(params) parsed = _parse_dsn(dsn1) dsn2 = _build_dsn(parsed) assert dsn1 == dsn2