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

KNX Config/OptionsFlow: Test connection to manually configured tunnel (#82872)

Browse source
This commit is contained in:
Matthias Alphart 2022-12-03 12:53:12 +01:00 • committed by GitHub
parent 949ebeeb97
commit 6cef37641c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 208 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -7,9 +7,10 @@ from typing import Any, Final
import voluptuous as vol
from xknx import XKNX
from xknx.exceptions.exception import InvalidSecureConfiguration
from xknx.exceptions.exception import CommunicationError, InvalidSecureConfiguration
from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
from xknx.io.gateway_scanner import GatewayDescriptor, GatewayScanner
from xknx.io.self_description import request_description
from xknx.secure import load_keyring
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
@ -204,8 +205,11 @@ class KNXCommonFlow(ABC, FlowHandler):
return await self.async_step_manual_tunnel()
errors: dict = {}
tunnel_options = [str(tunnel) for tunnel in self._found_tunnels]
tunnel_options.append(OPTION_MANUAL_TUNNEL)
tunnel_options = {
str(tunnel): f"{tunnel}{' 🔐' if tunnel.tunnelling_requires_secure else ''}"
for tunnel in self._found_tunnels
}
tunnel_options |= {OPTION_MANUAL_TUNNEL: OPTION_MANUAL_TUNNEL}
fields = {vol.Required(CONF_KNX_GATEWAY): vol.In(tunnel_options)}
return self.async_show_form(
@ -230,17 +234,38 @@ class KNXCommonFlow(ABC, FlowHandler):
except vol.Invalid:
errors[CONF_KNX_LOCAL_IP] = "invalid_ip_address"
selected_tunnelling_type = user_input[CONF_KNX_TUNNELING_TYPE]
if not errors:
try:
self._selected_tunnel = await request_description(
gateway_ip=_host,
gateway_port=user_input[CONF_PORT],
local_ip=_local_ip,
route_back=user_input[CONF_KNX_ROUTE_BACK],
)
except CommunicationError:
errors["base"] = "cannot_connect"
else:
if bool(self._selected_tunnel.tunnelling_requires_secure) is not (
selected_tunnelling_type == CONF_KNX_TUNNELING_TCP_SECURE
):
errors[CONF_KNX_TUNNELING_TYPE] = "unsupported_tunnel_type"
elif (
selected_tunnelling_type == CONF_KNX_TUNNELING_TCP
and not self._selected_tunnel.supports_tunnelling_tcp
):
errors[CONF_KNX_TUNNELING_TYPE] = "unsupported_tunnel_type"
if not errors:
connection_type = user_input[CONF_KNX_TUNNELING_TYPE]
self.new_entry_data = KNXConfigEntryData(
connection_type=selected_tunnelling_type,
host=_host,
port=user_input[CONF_PORT],
route_back=user_input[CONF_KNX_ROUTE_BACK],
local_ip=_local_ip,
connection_type=connection_type,
)
if connection_type == CONF_KNX_TUNNELING_TCP_SECURE:
if selected_tunnelling_type == CONF_KNX_TUNNELING_TCP_SECURE:
return self.async_show_menu(
step_id="secure_key_source",
menu_options=["secure_knxkeys", "secure_routing_manual"],
@ -299,7 +324,7 @@ class KNXCommonFlow(ABC, FlowHandler):
if self.show_advanced_options:
fields[vol.Optional(CONF_KNX_LOCAL_IP)] = _IP_SELECTOR
if not self._found_tunnels:
if not self._found_tunnels and not errors.get("base"):
errors["base"] = "no_tunnel_discovered"
return self.async_show_form(
step_id="manual_tunnel", data_schema=vol.Schema(fields), errors=errors

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

@ -99,7 +99,8 @@
"invalid_signature": "The password to decrypt the `.knxkeys` file is wrong.",
"file_not_found": "The specified `.knxkeys` file was not found in the path config/.storage/knx/",
"no_router_discovered": "No KNXnet/IP router was discovered on the network.",
"no_tunnel_discovered": "Could not find a KNX tunneling server on your network."
"no_tunnel_discovered": "Could not find a KNX tunneling server on your network.",
"unsupported_tunnel_type": "Selected tunnelling type not supported by gateway."
}
},
"options": {
@ -214,7 +215,8 @@
"invalid_signature": "[%key:component::knx::config::error::invalid_signature%]",
"file_not_found": "[%key:component::knx::config::error::file_not_found%]",
"no_router_discovered": "[%key:component::knx::config::error::no_router_discovered%]",
"no_tunnel_discovered": "[%key:component::knx::config::error::no_tunnel_discovered%]"
"no_tunnel_discovered": "[%key:component::knx::config::error::no_tunnel_discovered%]",
"unsupported_tunnel_type": "[%key:component::knx::config::error::unsupported_tunnel_type%]"
}
}
}

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

@ -1,8 +1,8 @@
"""Test the KNX config flow."""
from unittest.mock import patch
from unittest.mock import Mock, patch
import pytest
from xknx.exceptions.exception import InvalidSecureConfiguration
from xknx.exceptions.exception import CommunicationError, InvalidSecureConfiguration
from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
from xknx.io.gateway_scanner import GatewayDescriptor
@ -441,7 +441,11 @@ async def test_routing_secure_keyfile(
return_value=GatewayScannerMock(),
)
async def test_tunneling_setup_manual(
gateway_scanner_mock, hass: HomeAssistant, knx_setup, user_input, config_entry_data
_gateway_scanner_mock,
hass: HomeAssistant,
knx_setup,
user_input,
config_entry_data,
) -> None:
"""Test tunneling if no gateway was found found (or `manual` option was chosen)."""
result = await hass.config_entries.flow.async_init(
@ -460,11 +464,21 @@ async def test_tunneling_setup_manual(
assert result2["step_id"] == "manual_tunnel"
assert result2["errors"] == {"base": "no_tunnel_discovered"}
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
user_input,
)
await hass.async_block_till_done()
with patch(
"homeassistant.components.knx.config_flow.request_description",
return_value=_gateway_descriptor(
user_input[CONF_HOST],
user_input[CONF_PORT],
supports_tunnelling_tcp=(
user_input[CONF_KNX_TUNNELING_TYPE] == CONF_KNX_TUNNELING_TCP
),
),
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
user_input,
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "Tunneling @ 192.168.0.1"
assert result3["data"] == config_entry_data
@ -475,8 +489,146 @@ async def test_tunneling_setup_manual(
"homeassistant.components.knx.config_flow.GatewayScanner",
return_value=GatewayScannerMock(),
)
async def test_tunneling_setup_manual_request_description_error(
_gateway_scanner_mock,
hass: HomeAssistant,
knx_setup,
) -> None:
"""Test tunneling if no gateway was found found (or `manual` option was chosen)."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
},
)
assert result["step_id"] == "manual_tunnel"
assert result["errors"] == {"base": "no_tunnel_discovered"}
# TCP configured but not supported by gateway
with patch(
"homeassistant.components.knx.config_flow.request_description",
return_value=_gateway_descriptor(
"192.168.0.1",
3671,
supports_tunnelling_tcp=False,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3671,
},
)
assert result["step_id"] == "manual_tunnel"
assert result["errors"] == {
"base": "no_tunnel_discovered",
"tunneling_type": "unsupported_tunnel_type",
}
# TCP configured but Secure required by gateway
with patch(
"homeassistant.components.knx.config_flow.request_description",
return_value=_gateway_descriptor(
"192.168.0.1",
3671,
supports_tunnelling_tcp=True,
requires_secure=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3671,
},
)
assert result["step_id"] == "manual_tunnel"
assert result["errors"] == {
"base": "no_tunnel_discovered",
"tunneling_type": "unsupported_tunnel_type",
}
# Secure configured but not enabled on gateway
with patch(
"homeassistant.components.knx.config_flow.request_description",
return_value=_gateway_descriptor(
"192.168.0.1",
3671,
supports_tunnelling_tcp=True,
requires_secure=False,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3671,
},
)
assert result["step_id"] == "manual_tunnel"
assert result["errors"] == {
"base": "no_tunnel_discovered",
"tunneling_type": "unsupported_tunnel_type",
}
# No connection to gateway
with patch(
"homeassistant.components.knx.config_flow.request_description",
side_effect=CommunicationError(""),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3671,
},
)
assert result["step_id"] == "manual_tunnel"
assert result["errors"] == {"base": "cannot_connect"}
# OK configuration
with patch(
"homeassistant.components.knx.config_flow.request_description",
return_value=_gateway_descriptor(
"192.168.0.1",
3671,
supports_tunnelling_tcp=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3671,
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "Tunneling @ 192.168.0.1"
assert result["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3671,
}
knx_setup.assert_called_once()
@patch(
"homeassistant.components.knx.config_flow.GatewayScanner",
return_value=GatewayScannerMock(),
)
@patch(
"homeassistant.components.knx.config_flow.request_description",
return_value=_gateway_descriptor("192.168.0.2", 3675),
)
async def test_tunneling_setup_for_local_ip(
gateway_scanner_mock, hass: HomeAssistant, knx_setup
_request_description_mock, _gateway_scanner_mock, hass: HomeAssistant, knx_setup
) -> None:
"""Test tunneling if only one gateway is found."""
result = await hass.config_entries.flow.async_init(
@ -715,7 +867,17 @@ async def _get_menu_step(hass: HomeAssistant) -> FlowResult:
return result3
@patch(
"homeassistant.components.knx.config_flow.request_description",
return_value=_gateway_descriptor(
"192.168.0.1",
3675,
supports_tunnelling_tcp=True,
requires_secure=True,
),
)
async def test_get_secure_menu_step_manual_tunnelling(
_request_description_mock,
hass: HomeAssistant,
):
"""Test flow reaches secure_tunnellinn menu step from manual tunnelling configuration."""
@ -908,6 +1070,7 @@ async def test_options_flow_connection_type(
gateway = _gateway_descriptor("192.168.0.1", 3675)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
hass.data[DOMAIN] = Mock() # GatewayScanner uses running XKNX() in options flow
menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
with patch(