jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

KNX: Option to select specific tunnel endpoint on TCP connections (#131996)

Browse Source
This commit is contained in:
Matthias Alphart 2024-12-21 15:10:14 +01:00 committed by GitHub
parent a3febc4449
commit b5a7a41ebe
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 320 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
homeassistant/components/knx/__init__.py
View File

@ -401,6 +401,9 @@ class KNXModule:
)
return ConnectionConfig(
auto_reconnect=True,
individual_address=self.entry.data.get(
CONF_KNX_TUNNEL_ENDPOINT_IA, # may be configured at knxkey upload
),
secure_config=SecureConfig(
knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD),
knxkeys_file_path=_knxkeys_file,

143
homeassistant/components/knx/config_flow.py
View File

@ -4,7 +4,7 @@ from __future__ import annotations
from abc import ABC, abstractmethod
from collections.abc import AsyncGenerator
from typing import Any, Final
from typing import Any, Final, Literal
import voluptuous as vol
from xknx import XKNX
@ -121,6 +121,15 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
self._gatewayscanner: GatewayScanner | None = None
self._async_scan_gen: AsyncGenerator[GatewayDescriptor] | None = None
@property
def _xknx(self) -> XKNX:
"""Return XKNX instance."""
if isinstance(self, OptionsFlow) and (
knx_module := self.hass.data.get(KNX_MODULE_KEY)
):
return knx_module.xknx
return XKNX()
@abstractmethod
def finish_flow(self) -> ConfigFlowResult:
"""Finish the flow."""
@ -183,14 +192,8 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
CONF_KNX_ROUTING: CONF_KNX_ROUTING.capitalize(),
}
if isinstance(self, OptionsFlow) and (
knx_module := self.hass.data.get(KNX_MODULE_KEY)
):
xknx = knx_module.xknx
else:
xknx = XKNX()
self._gatewayscanner = GatewayScanner(
xknx, stop_on_found=0, timeout_in_seconds=2
self._xknx, stop_on_found=0, timeout_in_seconds=2
)
# keep a reference to the generator to scan in background until user selects a connection type
self._async_scan_gen = self._gatewayscanner.async_scan()
@ -204,8 +207,25 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
CONF_KNX_AUTOMATIC: CONF_KNX_AUTOMATIC.capitalize()
} | supported_connection_types
default_connection_type: Literal["automatic", "tunneling", "routing"]
_current_conn = self.initial_data.get(CONF_KNX_CONNECTION_TYPE)
if _current_conn in (
CONF_KNX_TUNNELING,
CONF_KNX_TUNNELING_TCP,
CONF_KNX_TUNNELING_TCP_SECURE,
):
default_connection_type = CONF_KNX_TUNNELING
elif _current_conn in (CONF_KNX_ROUTING, CONF_KNX_ROUTING_SECURE):
default_connection_type = CONF_KNX_ROUTING
elif CONF_KNX_AUTOMATIC in supported_connection_types:
default_connection_type = CONF_KNX_AUTOMATIC
else:
default_connection_type = CONF_KNX_TUNNELING
fields = {
vol.Required(CONF_KNX_CONNECTION_TYPE): vol.In(supported_connection_types)
vol.Required(
CONF_KNX_CONNECTION_TYPE, default=default_connection_type
): vol.In(supported_connection_types)
}
return self.async_show_form(
step_id="connection_type", data_schema=vol.Schema(fields)
@ -216,8 +236,7 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
) -> ConfigFlowResult:
"""Select a tunnel from a list.
Will be skipped if the gateway scan was unsuccessful
or if only one gateway was found.
Will be skipped if the gateway scan was unsuccessful.
"""
if user_input is not None:
if user_input[CONF_KNX_GATEWAY] == OPTION_MANUAL_TUNNEL:
@ -247,6 +266,8 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
user_password=None,
tunnel_endpoint_ia=None,
)
if connection_type == CONF_KNX_TUNNELING_TCP:
return await self.async_step_tcp_tunnel_endpoint()
if connection_type == CONF_KNX_TUNNELING_TCP_SECURE:
return await self.async_step_secure_key_source_menu_tunnel()
self.new_title = f"Tunneling @ {self._selected_tunnel}"
@ -255,16 +276,99 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
if not self._found_tunnels:
return await self.async_step_manual_tunnel()
errors: dict = {}
tunnel_options = {
str(tunnel): f"{tunnel}{' 🔐' if tunnel.tunnelling_requires_secure else ''}"
tunnel_options = [
selector.SelectOptionDict(
value=str(tunnel),
label=(
f"{tunnel}"
f"{' TCP' if tunnel.supports_tunnelling_tcp else ' UDP'}"
f"{' 🔐 Secure tunneling' if tunnel.tunnelling_requires_secure else ''}"
),
)
for tunnel in self._found_tunnels
]
tunnel_options.append(
selector.SelectOptionDict(
value=OPTION_MANUAL_TUNNEL, label=OPTION_MANUAL_TUNNEL
)
)
default_tunnel = next(
(
str(tunnel)
for tunnel in self._found_tunnels
if tunnel.ip_addr == self.initial_data.get(CONF_HOST)
),
vol.UNDEFINED,
)
fields = {
vol.Required(
CONF_KNX_GATEWAY, default=default_tunnel
): selector.SelectSelector(
selector.SelectSelectorConfig(
options=tunnel_options,
mode=selector.SelectSelectorMode.LIST,
)
)
}
tunnel_options |= {OPTION_MANUAL_TUNNEL: OPTION_MANUAL_TUNNEL}
fields = {vol.Required(CONF_KNX_GATEWAY): vol.In(tunnel_options)}
return self.async_show_form(step_id="tunnel", data_schema=vol.Schema(fields))
async def async_step_tcp_tunnel_endpoint(
self, user_input: dict | None = None
) -> ConfigFlowResult:
"""Select specific tunnel endpoint for plain TCP connection."""
if user_input is not None:
selected_tunnel_ia: str | None = (
None
if user_input[CONF_KNX_TUNNEL_ENDPOINT_IA] == CONF_KNX_AUTOMATIC
else user_input[CONF_KNX_TUNNEL_ENDPOINT_IA]
)
self.new_entry_data |= KNXConfigEntryData(
tunnel_endpoint_ia=selected_tunnel_ia,
)
self.new_title = (
f"{selected_tunnel_ia or 'Tunneling'} @ {self._selected_tunnel}"
)
return self.finish_flow()
# this step is only called from async_step_tunnel so self._selected_tunnel is always set
assert self._selected_tunnel
# skip if only one tunnel endpoint or no tunnelling slot infos
if len(self._selected_tunnel.tunnelling_slots) <= 1:
return self.finish_flow()
tunnel_endpoint_options = [
selector.SelectOptionDict(
value=CONF_KNX_AUTOMATIC, label=CONF_KNX_AUTOMATIC.capitalize()
)
]
_current_ia = self._xknx.current_address
tunnel_endpoint_options.extend(
selector.SelectOptionDict(
value=str(slot),
label=(
f"{slot} - {'current connection' if slot == _current_ia else 'occupied' if not slot_status.free else 'free'}"
),
)
for slot, slot_status in self._selected_tunnel.tunnelling_slots.items()
)
default_endpoint = (
self.initial_data.get(CONF_KNX_TUNNEL_ENDPOINT_IA) or CONF_KNX_AUTOMATIC
)
return self.async_show_form(
step_id="tunnel", data_schema=vol.Schema(fields), errors=errors
step_id="tcp_tunnel_endpoint",
data_schema=vol.Schema(
{
vol.Required(
CONF_KNX_TUNNEL_ENDPOINT_IA, default=default_endpoint
): selector.SelectSelector(
selector.SelectSelectorConfig(
options=tunnel_endpoint_options,
mode=selector.SelectSelectorMode.LIST,
)
),
}
),
)
async def async_step_manual_tunnel(
@ -612,12 +716,15 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
)
for endpoint in self._tunnel_endpoints
)
default_endpoint = (
self.initial_data.get(CONF_KNX_TUNNEL_ENDPOINT_IA) or CONF_KNX_AUTOMATIC
)
return self.async_show_form(
step_id="knxkeys_tunnel_select",
data_schema=vol.Schema(
{
vol.Required(
CONF_KNX_TUNNEL_ENDPOINT_IA, default=CONF_KNX_AUTOMATIC
CONF_KNX_TUNNEL_ENDPOINT_IA, default=default_endpoint
): selector.SelectSelector(
selector.SelectSelectorConfig(
options=tunnel_endpoint_options,

18
homeassistant/components/knx/strings.json
View File

@ -15,6 +15,13 @@
"gateway": "KNX Tunnel Connection"
}
},
"tcp_tunnel_endpoint": {
"title": "[%key:component::knx::config::step::knxkeys_tunnel_select::title%]",
"description": "[%key:component::knx::config::step::knxkeys_tunnel_select::description%]",
"data": {
"tunnel_endpoint_ia": "[%key:component::knx::config::step::knxkeys_tunnel_select::data::user_id%]"
}
},
"manual_tunnel": {
"title": "Tunnel settings",
"description": "Please enter the connection information of your tunneling device.",
@ -61,9 +68,9 @@
},
"knxkeys_tunnel_select": {
"title": "Tunnel endpoint",
"description": "Select the tunnel used for connection.",
"description": "Select the tunnel endpoint used for the connection.",
"data": {
"user_id": "`Automatic` will use the first free tunnel endpoint."
"user_id": "'Automatic' selects a free tunnel endpoint for you when connecting. If you're unsure, this is the best option."
}
},
"secure_tunnel_manual": {
@ -159,6 +166,13 @@
"gateway": "[%key:component::knx::config::step::tunnel::data::gateway%]"
}
},
"tcp_tunnel_endpoint": {
"title": "[%key:component::knx::config::step::knxkeys_tunnel_select::title%]",
"description": "[%key:component::knx::config::step::knxkeys_tunnel_select::description%]",
"data": {
"tunnel_endpoint_ia": "[%key:component::knx::config::step::knxkeys_tunnel_select::data::user_id%]"
}
},
"manual_tunnel": {
"title": "[%key:component::knx::config::step::manual_tunnel::title%]",
"description": "[%key:component::knx::config::step::manual_tunnel::description%]",

180
tests/components/knx/test_config_flow.py
View File

@ -7,6 +7,7 @@ import pytest
from xknx.exceptions.exception import CommunicationError, InvalidSecureConfiguration
from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
from xknx.io.gateway_scanner import GatewayDescriptor
from xknx.knxip.dib import TunnelingSlotStatus
from xknx.secure.keyring import sync_load_keyring
from xknx.telegram import IndividualAddress
@ -105,6 +106,7 @@ def _gateway_descriptor(
port: int,
supports_tunnelling_tcp: bool = False,
requires_secure: bool = False,
slots: bool = True,
) -> GatewayDescriptor:
"""Get mock gw descriptor."""
descriptor = GatewayDescriptor(
@ -120,6 +122,12 @@ def _gateway_descriptor(
)
descriptor.tunnelling_requires_secure = requires_secure
descriptor.routing_requires_secure = requires_secure
if supports_tunnelling_tcp and slots:
descriptor.tunnelling_slots = {
IndividualAddress("1.0.240"): TunnelingSlotStatus(True, True, True),
IndividualAddress("1.0.241"): TunnelingSlotStatus(True, True, False),
IndividualAddress("1.0.242"): TunnelingSlotStatus(True, True, True),
}
return descriptor
@ -791,12 +799,14 @@ async def test_tunneling_setup_for_multiple_found_gateways(
hass: HomeAssistant, knx_setup
) -> None:
"""Test tunneling if multiple gateways are found."""
gateway = _gateway_descriptor("192.168.0.1", 3675)
gateway2 = _gateway_descriptor("192.168.1.100", 3675)
gateway_udp = _gateway_descriptor("192.168.0.1", 3675)
gateway_tcp = _gateway_descriptor("192.168.1.100", 3675, True)
with patch(
"homeassistant.components.knx.config_flow.GatewayScanner"
) as gateway_scanner_mock:
gateway_scanner_mock.return_value = GatewayScannerMock([gateway, gateway2])
gateway_scanner_mock.return_value = GatewayScannerMock(
[gateway_udp, gateway_tcp]
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
@ -815,7 +825,7 @@ async def test_tunneling_setup_for_multiple_found_gateways(
result = await hass.config_entries.flow.async_configure(
tunnel_flow["flow_id"],
{CONF_KNX_GATEWAY: str(gateway)},
{CONF_KNX_GATEWAY: str(gateway_udp)},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
@ -833,6 +843,110 @@ async def test_tunneling_setup_for_multiple_found_gateways(
knx_setup.assert_called_once()
async def test_tunneling_setup_tcp_endpoint_select_skip(
hass: HomeAssistant, knx_setup
) -> None:
"""Test tunneling TCP endpoint selection skipped if no slot info found."""
gateway_udp = _gateway_descriptor("192.168.0.1", 3675)
gateway_tcp_no_slots = _gateway_descriptor("192.168.1.100", 3675, True, slots=False)
with patch(
"homeassistant.components.knx.config_flow.GatewayScanner"
) as gateway_scanner_mock:
gateway_scanner_mock.return_value = GatewayScannerMock(
[gateway_udp, gateway_tcp_no_slots]
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert not result["errors"]
tunnel_flow = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
},
)
assert tunnel_flow["type"] is FlowResultType.FORM
assert tunnel_flow["step_id"] == "tunnel"
assert not tunnel_flow["errors"]
result = await hass.config_entries.flow.async_configure(
tunnel_flow["flow_id"],
{CONF_KNX_GATEWAY: str(gateway_tcp_no_slots)},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP,
CONF_HOST: "192.168.1.100",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240",
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_TUNNEL_ENDPOINT_IA: None,
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None,
CONF_KNX_SECURE_USER_ID: None,
CONF_KNX_SECURE_USER_PASSWORD: None,
}
knx_setup.assert_called_once()
async def test_tunneling_setup_tcp_endpoint_select(
hass: HomeAssistant, knx_setup
) -> None:
"""Test tunneling TCP endpoint selection."""
gateway_tcp = _gateway_descriptor("192.168.1.100", 3675, True)
with patch(
"homeassistant.components.knx.config_flow.GatewayScanner"
) as gateway_scanner_mock:
gateway_scanner_mock.return_value = GatewayScannerMock([gateway_tcp])
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert not result["errors"]
tunnel_flow = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
},
)
assert tunnel_flow["type"] is FlowResultType.FORM
assert tunnel_flow["step_id"] == "tunnel"
assert not tunnel_flow["errors"]
endpoint_flow = await hass.config_entries.flow.async_configure(
tunnel_flow["flow_id"],
{CONF_KNX_GATEWAY: str(gateway_tcp)},
)
assert endpoint_flow["type"] is FlowResultType.FORM
assert endpoint_flow["step_id"] == "tcp_tunnel_endpoint"
assert not endpoint_flow["errors"]
result = await hass.config_entries.flow.async_configure(
endpoint_flow["flow_id"],
{CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.242"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "1.0.242 @ 1.0.0 - Test @ 192.168.1.100:3675"
assert result["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP,
CONF_HOST: "192.168.1.100",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240",
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.242",
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None,
CONF_KNX_SECURE_USER_ID: None,
CONF_KNX_SECURE_USER_PASSWORD: None,
}
knx_setup.assert_called_once()
@pytest.mark.parametrize(
"gateway",
[
@ -1319,6 +1433,64 @@ async def test_options_flow_secure_manual_to_keyfile(
knx_setup.assert_called_once()
async def test_options_flow_routing(hass: HomeAssistant, knx_setup) -> None:
"""Test options flow changing routing settings."""
mock_config_entry = MockConfigEntry(
title="KNX",
domain="knx",
data={
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
},
)
gateway = _gateway_descriptor("192.168.0.1", 3676)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
with patch(
"homeassistant.components.knx.config_flow.GatewayScanner"
) as gateway_scanner_mock:
gateway_scanner_mock.return_value = GatewayScannerMock([gateway])
result = await hass.config_entries.options.async_configure(
menu_step["flow_id"],
{"next_step_id": "connection_type"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "connection_type"
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "routing"
assert result2["errors"] == {}
result3 = await hass.config_entries.options.async_configure(
result2["flow_id"],
{
CONF_KNX_INDIVIDUAL_ADDRESS: "2.0.4",
},
)
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert mock_config_entry.data == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_LOCAL_IP: None,
CONF_KNX_INDIVIDUAL_ADDRESS: "2.0.4",
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None,
CONF_KNX_SECURE_USER_ID: None,
CONF_KNX_SECURE_USER_PASSWORD: None,
CONF_KNX_TUNNEL_ENDPOINT_IA: None,
}
knx_setup.assert_called_once()
async def test_options_communication_settings(
hass: HomeAssistant, knx_setup, mock_config_entry: MockConfigEntry
) -> None: