Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-06 21:35:13 +01:00
Code Activity
Files
9d24d8a1d9b2bc655c19b46ffff485b47fb4e79c
core/tests/components/knx/test_config_flow.py
T
Martin Hoefling 9eb29b8494 Add PostgreSQL backend and storage backend selection for KNX telegrams (#175673) 
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
2026-07-14 10:09:11 +02:00

2030 lines
71 KiB
Python
Raw Blame History

"""Test the KNX config flow."""
import asyncio
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: dict = DEFAULT_ENTRY_OPTIONS,
) -> None:
"""Assert the config entry stores connection data and options separately."""
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
View Git Blame Copy Permalink