evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add support for TCP tunneling connections to KNX (#64928)

Browse source
This commit is contained in:
Marvin Wichmann 2022-01-26 06:36:07 +01:00 • committed by GitHub
parent cdad1a9f27
commit 072f7d5611
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 344 additions and 149 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
homeassistant/components/knx/__init__.py
View file

@ -45,6 +45,7 @@ from .const import (
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_ROUTING,
CONF_KNX_TUNNELING,
CONF_KNX_TUNNELING_TCP,
DATA_HASS_CONFIG,
DATA_KNX_CONFIG,
DOMAIN,
@ -404,6 +405,13 @@ class KNXModule:
route_back=self.config.get(ConnectionSchema.CONF_KNX_ROUTE_BACK, False),
auto_reconnect=True,
)
if _conn_type == CONF_KNX_TUNNELING_TCP:
return ConnectionConfig(
connection_type=ConnectionType.TUNNELING_TCP,
gateway_ip=self.config[CONF_HOST],
gateway_port=self.config[CONF_PORT],
auto_reconnect=True,
)
return ConnectionConfig(auto_reconnect=True)
async def connection_state_changed_cb(self, state: XknxConnectionState) -> None:

263
homeassistant/components/knx/config_flow.py
View file

@ -22,6 +22,7 @@ from .const import (
CONF_KNX_INITIAL_CONNECTION_TYPES,
CONF_KNX_ROUTING,
CONF_KNX_TUNNELING,
CONF_KNX_TUNNELING_TCP,
DOMAIN,
)
from .schema import ConnectionSchema
@ -38,15 +39,19 @@ DEFAULT_ENTRY_DATA: Final = {
ConnectionSchema.CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
}
CONF_KNX_TUNNELING_TYPE: Final = "tunneling_type"
CONF_KNX_LABEL_TUNNELING_TCP: Final = "TCP"
CONF_KNX_LABEL_TUNNELING_UDP: Final = "UDP"
CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK: Final = "UDP with route back / NAT mode"
class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a KNX config flow."""
VERSION = 1
_tunnels: list[GatewayDescriptor]
_gateway_ip: str = ""
_gateway_port: int = DEFAULT_MCAST_PORT
_found_tunnels: list[GatewayDescriptor]
_selected_tunnel: GatewayDescriptor | None
@staticmethod
@callback
@ -59,7 +64,8 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
self._tunnels = []
self._found_tunnels = []
self._selected_tunnel = None
return await self.async_step_type()
async def async_step_type(self, user_input: dict | None = None) -> FlowResult:
@ -75,7 +81,7 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if connection_type == CONF_KNX_ROUTING:
return await self.async_step_routing()
if connection_type == CONF_KNX_TUNNELING and self._tunnels:
if connection_type == CONF_KNX_TUNNELING and self._found_tunnels:
return await self.async_step_tunnel()
return await self.async_step_manual_tunnel()
@ -88,7 +94,7 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if gateways:
# add automatic only if a gateway responded
supported_connection_types.insert(0, CONF_KNX_AUTOMATIC)
self._tunnels = [
self._found_tunnels = [
gateway for gateway in gateways if gateway.supports_tunnelling
]
@ -100,11 +106,35 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
step_id="type", data_schema=vol.Schema(fields), errors=errors
)
async def async_step_tunnel(self, user_input: dict | None = None) -> FlowResult:
"""Select a tunnel from a list. Will be skipped if the gateway scan was unsuccessful or if only one gateway was found."""
if user_input is not None:
self._selected_tunnel = next(
tunnel
for tunnel in self._found_tunnels
if user_input[CONF_KNX_GATEWAY] == str(tunnel)
)
return await self.async_step_manual_tunnel()
# skip this step if the user has only one unique gateway.
if len(self._found_tunnels) == 1:
self._selected_tunnel = self._found_tunnels[0]
return await self.async_step_manual_tunnel()
errors: dict = {}
tunnels_repr = {str(tunnel) for tunnel in self._found_tunnels}
fields = {vol.Required(CONF_KNX_GATEWAY): vol.In(tunnels_repr)}
return self.async_show_form(
step_id="tunnel", data_schema=vol.Schema(fields), errors=errors
)
async def async_step_manual_tunnel(
self, user_input: dict | None = None
) -> FlowResult:
"""General setup."""
"""Manually configure tunnel connection parameters. Fields default to preselected gateway if one was found."""
if user_input is not None:
connection_type = user_input[CONF_KNX_TUNNELING_TYPE]
return self.async_create_entry(
title=f"{CONF_KNX_TUNNELING.capitalize()} @ {user_input[CONF_HOST]}",
data={
@ -114,26 +144,41 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
CONF_KNX_INDIVIDUAL_ADDRESS: user_input[
CONF_KNX_INDIVIDUAL_ADDRESS
],
ConnectionSchema.CONF_KNX_ROUTE_BACK: user_input[
ConnectionSchema.CONF_KNX_ROUTE_BACK
],
ConnectionSchema.CONF_KNX_ROUTE_BACK: (
connection_type == CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK
),
ConnectionSchema.CONF_KNX_LOCAL_IP: user_input.get(
ConnectionSchema.CONF_KNX_LOCAL_IP
),
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_KNX_CONNECTION_TYPE: (
CONF_KNX_TUNNELING_TCP
if connection_type == CONF_KNX_LABEL_TUNNELING_TCP
else CONF_KNX_TUNNELING
),
},
)
errors: dict = {}
connection_methods: list[str] = [
CONF_KNX_LABEL_TUNNELING_TCP,
CONF_KNX_LABEL_TUNNELING_UDP,
CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK,
]
ip_address = ""
port = DEFAULT_MCAST_PORT
if self._selected_tunnel is not None:
ip_address = self._selected_tunnel.ip_addr
port = self._selected_tunnel.port
if not self._selected_tunnel.supports_tunnelling_tcp:
connection_methods.remove(CONF_KNX_LABEL_TUNNELING_TCP)
fields = {
vol.Required(CONF_HOST, default=self._gateway_ip): str,
vol.Required(CONF_PORT, default=self._gateway_port): vol.Coerce(int),
vol.Required(CONF_KNX_TUNNELING_TYPE): vol.In(connection_methods),
vol.Required(CONF_HOST, default=ip_address): str,
vol.Required(CONF_PORT, default=port): cv.port,
vol.Required(
CONF_KNX_INDIVIDUAL_ADDRESS, default=XKNX.DEFAULT_ADDRESS
): str,
vol.Required(
ConnectionSchema.CONF_KNX_ROUTE_BACK, default=False
): vol.Coerce(bool),
}
if self.show_advanced_options:
@ -143,38 +188,6 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
step_id="manual_tunnel", data_schema=vol.Schema(fields), errors=errors
)
async def async_step_tunnel(self, user_input: dict | None = None) -> FlowResult:
"""Select a tunnel from a list. Will be skipped if the gateway scan was unsuccessful or if only one gateway was found."""
if user_input is not None:
gateway: GatewayDescriptor = next(
gateway
for gateway in self._tunnels
if user_input[CONF_KNX_GATEWAY] == str(gateway)
)
self._gateway_ip = gateway.ip_addr
self._gateway_port = gateway.port
return await self.async_step_manual_tunnel()
errors: dict = {}
tunnel_repr = {
str(tunnel) for tunnel in self._tunnels if tunnel.supports_tunnelling
}
# skip this step if the user has only one unique gateway.
if len(tunnel_repr) == 1:
_gateway: GatewayDescriptor = self._tunnels[0]
self._gateway_ip = _gateway.ip_addr
self._gateway_port = _gateway.port
return await self.async_step_manual_tunnel()
fields = {vol.Required(CONF_KNX_GATEWAY): vol.In(tunnel_repr)}
return self.async_show_form(
step_id="tunnel", data_schema=vol.Schema(fields), errors=errors
)
async def async_step_routing(self, user_input: dict | None = None) -> FlowResult:
"""Routing setup."""
if user_input is not None:
@ -291,65 +304,6 @@ class KNXOptionsFlowHandler(OptionsFlow):
"""Initialize KNX options flow."""
self.config_entry = config_entry
async def async_step_tunnel(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage KNX tunneling options."""
if (
self.general_settings.get(CONF_KNX_CONNECTION_TYPE) == CONF_KNX_TUNNELING
and user_input is None
):
return self.async_show_form(
step_id="tunnel",
data_schema=vol.Schema(
{
vol.Required(
CONF_HOST, default=self.current_config.get(CONF_HOST)
): str,
vol.Required(
CONF_PORT, default=self.current_config.get(CONF_PORT, 3671)
): cv.port,
vol.Required(
ConnectionSchema.CONF_KNX_ROUTE_BACK,
default=self.current_config.get(
ConnectionSchema.CONF_KNX_ROUTE_BACK, False
),
): vol.Coerce(bool),
}
),
last_step=True,
)
entry_data = {
**DEFAULT_ENTRY_DATA,
**self.general_settings,
ConnectionSchema.CONF_KNX_LOCAL_IP: self.general_settings.get(
ConnectionSchema.CONF_KNX_LOCAL_IP
)
if self.general_settings.get(ConnectionSchema.CONF_KNX_LOCAL_IP)
!= CONF_DEFAULT_LOCAL_IP
else None,
CONF_HOST: self.current_config.get(CONF_HOST, ""),
}
if user_input is not None:
entry_data = {
**entry_data,
**user_input,
}
entry_title = str(entry_data[CONF_KNX_CONNECTION_TYPE]).capitalize()
if entry_data[CONF_KNX_CONNECTION_TYPE] == CONF_KNX_TUNNELING:
entry_title = f"{CONF_KNX_TUNNELING.capitalize()} @ {entry_data[CONF_HOST]}"
self.hass.config_entries.async_update_entry(
self.config_entry,
data=entry_data,
title=entry_title,
)
return self.async_create_entry(title="", data={})
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
@ -368,7 +322,12 @@ class KNXOptionsFlowHandler(OptionsFlow):
data_schema = {
vol.Required(
CONF_KNX_CONNECTION_TYPE,
default=self.current_config.get(CONF_KNX_CONNECTION_TYPE),
default=(
CONF_KNX_TUNNELING
if self.current_config.get(CONF_KNX_CONNECTION_TYPE)
== CONF_KNX_TUNNELING_TCP
else self.current_config.get(CONF_KNX_CONNECTION_TYPE)
),
): vol.In(supported_connection_types),
vol.Required(
CONF_KNX_INDIVIDUAL_ADDRESS,
@ -427,6 +386,94 @@ class KNXOptionsFlowHandler(OptionsFlow):
!= CONF_KNX_TUNNELING,
)
async def async_step_tunnel(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage KNX tunneling options."""
if (
self.general_settings.get(CONF_KNX_CONNECTION_TYPE) == CONF_KNX_TUNNELING
and user_input is None
):
connection_methods: list[str] = [
CONF_KNX_LABEL_TUNNELING_TCP,
CONF_KNX_LABEL_TUNNELING_UDP,
CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK,
]
return self.async_show_form(
step_id="tunnel",
data_schema=vol.Schema(
{
vol.Required(
CONF_KNX_TUNNELING_TYPE,
default=get_knx_tunneling_type(self.current_config),
): vol.In(connection_methods),
vol.Required(
CONF_HOST, default=self.current_config.get(CONF_HOST)
): str,
vol.Required(
CONF_PORT, default=self.current_config.get(CONF_PORT, 3671)
): cv.port,
}
),
last_step=True,
)
entry_data = {
**DEFAULT_ENTRY_DATA,
**self.general_settings,
ConnectionSchema.CONF_KNX_LOCAL_IP: self.general_settings.get(
ConnectionSchema.CONF_KNX_LOCAL_IP
)
if self.general_settings.get(ConnectionSchema.CONF_KNX_LOCAL_IP)
!= CONF_DEFAULT_LOCAL_IP
else None,
CONF_HOST: self.current_config.get(CONF_HOST, ""),
}
if user_input is not None:
connection_type = user_input[CONF_KNX_TUNNELING_TYPE]
entry_data = {
**entry_data,
CONF_HOST: user_input[CONF_HOST],
CONF_PORT: user_input[CONF_PORT],
ConnectionSchema.CONF_KNX_ROUTE_BACK: (
connection_type == CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK
),
CONF_KNX_CONNECTION_TYPE: (
CONF_KNX_TUNNELING_TCP
if connection_type == CONF_KNX_LABEL_TUNNELING_TCP
else CONF_KNX_TUNNELING
),
}
entry_title = str(entry_data[CONF_KNX_CONNECTION_TYPE]).capitalize()
if entry_data[CONF_KNX_CONNECTION_TYPE] == CONF_KNX_TUNNELING:
entry_title = f"{CONF_KNX_TUNNELING.capitalize()} @ {entry_data[CONF_HOST]}"
if entry_data[CONF_KNX_CONNECTION_TYPE] == CONF_KNX_TUNNELING_TCP:
entry_title = (
f"{CONF_KNX_TUNNELING.capitalize()} (TCP) @ {entry_data[CONF_HOST]}"
)
self.hass.config_entries.async_update_entry(
self.config_entry,
data=entry_data,
title=entry_title,
)
return self.async_create_entry(title="", data={})
def get_knx_tunneling_type(config_entry_data: dict) -> str:
"""Obtain the knx tunneling type based on the data in the config entry data."""
connection_type = config_entry_data[CONF_KNX_CONNECTION_TYPE]
route_back = config_entry_data.get(ConnectionSchema.CONF_KNX_ROUTE_BACK, False)
if route_back and connection_type == CONF_KNX_TUNNELING:
return CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK
if connection_type == CONF_KNX_TUNNELING_TCP:
return CONF_KNX_LABEL_TUNNELING_TCP
return CONF_KNX_LABEL_TUNNELING_UDP
async def scan_for_gateways(stop_on_found: int = 0) -> list[GatewayDescriptor]:
"""Scan for gateways within the network."""

1
homeassistant/components/knx/const.py
View file

@ -34,6 +34,7 @@ CONF_KNX_CONNECTION_TYPE: Final = "connection_type"
CONF_KNX_AUTOMATIC: Final = "automatic"
CONF_KNX_ROUTING: Final = "routing"
CONF_KNX_TUNNELING: Final = "tunneling"
CONF_KNX_TUNNELING_TCP: Final = "tunneling_tcp"
CONF_PAYLOAD: Final = "payload"
CONF_PAYLOAD_LENGTH: Final = "payload_length"
CONF_RESET_AFTER: Final = "reset_after"

6
homeassistant/components/knx/strings.json
View file

@ -16,10 +16,10 @@
"manual_tunnel": {
"description": "Please enter the connection information of your tunneling device.",
"data": {
"tunneling_type": "KNX Tunneling Type",
"port": "[%key:common::config_flow::data::port%]",
"host": "[%key:common::config_flow::data::host%]",
"individual_address": "Individual address for the connection",
"route_back": "Route Back / NAT Mode",
"local_ip": "Local IP of Home Assistant (leave empty for automatic detection)"
}
},
@ -56,9 +56,9 @@
},
"tunnel": {
"data": {
"tunneling_type": "KNX Tunneling Type",
"port": "[%key:common::config_flow::data::port%]",
"host": "[%key:common::config_flow::data::host%]",
"route_back": "Route Back / NAT Mode"
"host": "[%key:common::config_flow::data::host%]"
}
}
}

8
homeassistant/components/knx/translations/en.json
View file

@ -10,11 +10,11 @@
"step": {
"manual_tunnel": {
"data": {
"tunneling_type": "KNX Tunneling Type",
"host": "Host",
"individual_address": "Individual address for the connection",
"local_ip": "Local IP of Home Assistant (leave empty for automatic detection)",
"port": "Port",
"route_back": "Route Back / NAT Mode"
"port": "Port"
},
"description": "Please enter the connection information of your tunneling device."
},
@ -56,10 +56,10 @@
},
"tunnel": {
"data": {
"tunneling_type": "KNX Tunneling Type",
"host": "Host",
"local_ip": "Local IP (leave empty if unsure)",
"port": "Port",
"route_back": "Route Back / NAT Mode"
"port": "Port"
}
}
}

207
tests/components/knx/test_config_flow.py
View file

@ -11,7 +11,12 @@ from homeassistant.components.knx import ConnectionSchema
from homeassistant.components.knx.config_flow import (
CONF_DEFAULT_LOCAL_IP,
CONF_KNX_GATEWAY,
CONF_KNX_LABEL_TUNNELING_TCP,
CONF_KNX_LABEL_TUNNELING_UDP,
CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK,
CONF_KNX_TUNNELING_TYPE,
DEFAULT_ENTRY_DATA,
get_knx_tunneling_type,
)
from homeassistant.components.knx.const import (
CONF_KNX_AUTOMATIC,
@ -19,6 +24,7 @@ from homeassistant.components.knx.const import (
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_ROUTING,
CONF_KNX_TUNNELING,
CONF_KNX_TUNNELING_TCP,
DOMAIN,
)
from homeassistant.const import CONF_HOST, CONF_PORT
@ -28,7 +34,9 @@ from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY, RESULT_TYPE_
from tests.common import MockConfigEntry
def _gateway_descriptor(ip: str, port: int) -> GatewayDescriptor:
def _gateway_descriptor(
ip: str, port: int, supports_tunnelling_tcp: bool = False
) -> GatewayDescriptor:
"""Get mock gw descriptor."""
return GatewayDescriptor(
"Test",
@ -38,6 +46,7 @@ def _gateway_descriptor(ip: str, port: int) -> GatewayDescriptor:
"127.0.0.1",
supports_routing=True,
supports_tunnelling=True,
supports_tunnelling_tcp=supports_tunnelling_tcp,
)
@ -153,9 +162,64 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None:
assert len(mock_setup_entry.mock_calls) == 1
async def test_tunneling_setup(hass: HomeAssistant) -> None:
@pytest.mark.parametrize(
"user_input,config_entry_data",
[
(
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
},
{
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
},
),
(
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_TCP,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
},
{
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
},
),
(
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
},
{
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
},
),
],
)
async def test_tunneling_setup(
hass: HomeAssistant, user_input, config_entry_data
) -> None:
"""Test tunneling if only one gateway is found."""
gateway = _gateway_descriptor("192.168.0.1", 3675)
gateway = _gateway_descriptor("192.168.0.1", 3675, True)
with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways:
gateways.return_value = [gateway]
result = await hass.config_entries.flow.async_init(
@ -181,23 +245,12 @@ async def test_tunneling_setup(hass: HomeAssistant) -> None:
) as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
},
user_input,
)
await hass.async_block_till_done()
assert result3["type"] == RESULT_TYPE_CREATE_ENTRY
assert result3["title"] == "Tunneling @ 192.168.0.1"
assert result3["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
}
assert result3["data"] == config_entry_data
assert len(mock_setup_entry.mock_calls) == 1
@ -235,6 +288,7 @@ async def test_tunneling_setup_for_local_ip(hass: HomeAssistant) -> None:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP,
CONF_HOST: "192.168.0.2",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_LOCAL_IP: "192.168.1.112",
@ -294,6 +348,7 @@ async def test_tunneling_setup_for_multiple_found_gateways(hass: HomeAssistant)
manual_tunnel_flow = await hass.config_entries.flow.async_configure(
manual_tunnel["flow_id"],
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
},
@ -595,8 +650,73 @@ async def test_options_flow(
}
@pytest.mark.parametrize(
"user_input,config_entry_data",
[
(
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK,
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
},
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
},
),
(
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP,
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
},
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
},
),
(
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_TCP,
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
},
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
},
),
],
)
async def test_tunneling_options_flow(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
user_input,
config_entry_data,
) -> None:
"""Test options flow for tunneling."""
mock_config_entry.add_to_hass(hass)
@ -628,29 +748,14 @@ async def test_tunneling_options_flow(
result3 = await hass.config_entries.options.async_configure(
result2["flow_id"],
user_input={
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
},
user_input=user_input,
)
await hass.async_block_till_done()
assert result3.get("type") == RESULT_TYPE_CREATE_ENTRY
assert not result3.get("data")
assert mock_config_entry.data == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
}
assert mock_config_entry.data == config_entry_data
@pytest.mark.parametrize(
@ -730,3 +835,37 @@ async def test_advanced_options(
assert not result2.get("data")
assert mock_config_entry.data == config_entry_data
@pytest.mark.parametrize(
"config_entry_data,result",
[
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
},
CONF_KNX_LABEL_TUNNELING_UDP,
),
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
},
CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK,
),
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP,
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
},
CONF_KNX_LABEL_TUNNELING_TCP,
),
],
)
async def test_get_knx_tunneling_type(
config_entry_data,
result,
) -> None:
"""Test converting config entry data to tunneling type for config flow."""
assert get_knx_tunneling_type(config_entry_data) == result