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

Upload KNX Keyfile from Config/Options Flow directly (#88097)

Browse source 
* Manage KNX Keyfile from UI

* migrate config entry to use new keyfile

* Revert "migrate config entry to use new keyfile"

use same config style as before instead of entry version migration

* clean up uploaded file when integration is removed

* change default filename

* revert to previous step name

* remove empty directory on unload
This commit is contained in:
Matthias Alphart 2023-02-20 14:48:56 +01:00 • committed by GitHub
parent 9876dd804e
commit 6a0ea09f29
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 480 additions and 190 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -2,7 +2,9 @@
from __future__ import annotations
import asyncio
import contextlib
import logging
from pathlib import Path
from typing import Final
import voluptuous as vol
@ -335,6 +337,21 @@ async def async_update_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
await hass.config_entries.async_reload(entry.entry_id)
async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Remove a config entry."""
def remove_keyring_files(file_path: Path) -> None:
"""Remove keyring files."""
with contextlib.suppress(FileNotFoundError):
file_path.unlink()
with contextlib.suppress(FileNotFoundError, OSError):
file_path.parent.rmdir()
if (_knxkeys_file := entry.data.get(CONF_KNX_KNXKEY_FILENAME)) is not None:
file_path = Path(hass.config.path(STORAGE_DIR)) / _knxkeys_file
await hass.async_add_executor_job(remove_keyring_files, file_path)
class KNXModule:
"""Representation of KNX Object."""
@ -384,6 +401,14 @@ class KNXModule:
def connection_config(self) -> ConnectionConfig:
"""Return the connection_config."""
_conn_type: str = self.entry.data[CONF_KNX_CONNECTION_TYPE]
_knxkeys_file: str | None = (
self.hass.config.path(
STORAGE_DIR,
self.entry.data[CONF_KNX_KNXKEY_FILENAME],
)
if self.entry.data.get(CONF_KNX_KNXKEY_FILENAME) is not None
else None
)
if _conn_type == CONF_KNX_ROUTING:
return ConnectionConfig(
connection_type=ConnectionType.ROUTING,
@ -392,6 +417,10 @@ class KNXModule:
multicast_port=self.entry.data[CONF_KNX_MCAST_PORT],
local_ip=self.entry.data.get(CONF_KNX_LOCAL_IP),
auto_reconnect=True,
secure_config=SecureConfig(
knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD),
knxkeys_file_path=_knxkeys_file,
),
threaded=True,
)
if _conn_type == CONF_KNX_TUNNELING:
@ -402,6 +431,10 @@ class KNXModule:
local_ip=self.entry.data.get(CONF_KNX_LOCAL_IP),
route_back=self.entry.data.get(CONF_KNX_ROUTE_BACK, False),
auto_reconnect=True,
secure_config=SecureConfig(
knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD),
knxkeys_file_path=_knxkeys_file,
),
threaded=True,
)
if _conn_type == CONF_KNX_TUNNELING_TCP:
@ -410,16 +443,12 @@ class KNXModule:
gateway_ip=self.entry.data[CONF_HOST],
gateway_port=self.entry.data[CONF_PORT],
auto_reconnect=True,
secure_config=SecureConfig(
knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD),
knxkeys_file_path=_knxkeys_file,
),
threaded=True,
)
knxkeys_file: str | None = (
self.hass.config.path(
STORAGE_DIR,
self.entry.data[CONF_KNX_KNXKEY_FILENAME],
)
if self.entry.data.get(CONF_KNX_KNXKEY_FILENAME) is not None
else None
)
if _conn_type == CONF_KNX_TUNNELING_TCP_SECURE:
return ConnectionConfig(
connection_type=ConnectionType.TUNNELING_TCP_SECURE,
@ -432,7 +461,7 @@ class KNXModule:
CONF_KNX_SECURE_DEVICE_AUTHENTICATION
),
knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD),
knxkeys_file_path=knxkeys_file,
knxkeys_file_path=_knxkeys_file,
),
auto_reconnect=True,
threaded=True,
@ -450,13 +479,17 @@ class KNXModule:
CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE
),
knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD),
knxkeys_file_path=knxkeys_file,
knxkeys_file_path=_knxkeys_file,
),
auto_reconnect=True,
threaded=True,
)
return ConnectionConfig(
auto_reconnect=True,
secure_config=SecureConfig(
knxkeys_password=self.entry.data.get(CONF_KNX_KNXKEY_PASSWORD),
knxkeys_file_path=_knxkeys_file,
),
threaded=True,
)

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

@ -3,6 +3,7 @@ from __future__ import annotations
from abc import ABC, abstractmethod
from collections.abc import AsyncGenerator
from pathlib import Path
from typing import Any, Final
import voluptuous as vol
@ -11,8 +12,9 @@ from xknx.exceptions.exception import CommunicationError, InvalidSecureConfigura
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.keyring import XMLInterface, load_keyring
from xknx.secure.keyring import Keyring, XMLInterface, sync_load_keyring
from homeassistant.components.file_upload import process_uploaded_file
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import callback
@ -27,7 +29,6 @@ from .const import (
CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_KNXKEY_FILENAME,
CONF_KNX_KNXKEY_PASSWORD,
CONF_KNX_LOCAL_IP,
CONF_KNX_MCAST_GRP,
@ -42,10 +43,10 @@ from .const import (
CONF_KNX_SECURE_USER_ID,
CONF_KNX_SECURE_USER_PASSWORD,
CONF_KNX_STATE_UPDATER,
CONF_KNX_TUNNEL_ENDPOINT_IA,
CONF_KNX_TUNNELING,
CONF_KNX_TUNNELING_TCP,
CONF_KNX_TUNNELING_TCP_SECURE,
CONST_KNX_STORAGE_KEY,
DEFAULT_ROUTING_IA,
DOMAIN,
KNXConfigEntryData,
@ -65,6 +66,9 @@ DEFAULT_ENTRY_DATA = KNXConfigEntryData(
state_updater=CONF_KNX_DEFAULT_STATE_UPDATER,
)
CONF_KEYRING_FILE: Final = "knxkeys_file"
DEFAULT_KNX_KEYRING_FILENAME: Final = "keyring.knxkeys"
CONF_KNX_TUNNELING_TYPE: Final = "tunneling_type"
CONF_KNX_TUNNELING_TYPE_LABELS: Final = {
CONF_KNX_TUNNELING: "UDP (Tunnelling v1)",
@ -93,6 +97,9 @@ class KNXCommonFlow(ABC, FlowHandler):
"""Initialize KNXCommonFlow."""
self.initial_data = initial_data
self.new_entry_data = KNXConfigEntryData()
self.new_title: str | None = None
self._keyring: Keyring | None = None
self._found_gateways: list[GatewayDescriptor] = []
self._found_tunnels: list[GatewayDescriptor] = []
self._selected_tunnel: GatewayDescriptor | None = None
@ -102,9 +109,25 @@ class KNXCommonFlow(ABC, FlowHandler):
self._async_scan_gen: AsyncGenerator[GatewayDescriptor, None] | None = None
@abstractmethod
def finish_flow(self, title: str) -> FlowResult:
def finish_flow(self) -> FlowResult:
"""Finish the flow."""
@property
def connection_type(self) -> str:
"""Return the configured connection type."""
_new_type = self.new_entry_data.get(CONF_KNX_CONNECTION_TYPE)
if _new_type is None:
return self.initial_data[CONF_KNX_CONNECTION_TYPE]
return _new_type
@property
def tunnel_endpoint_ia(self) -> str | None:
"""Return the configured tunnel endpoint individual address."""
return self.new_entry_data.get(
CONF_KNX_TUNNEL_ENDPOINT_IA,
self.initial_data.get(CONF_KNX_TUNNEL_ENDPOINT_IA),
)
async def async_step_connection_type(
self, user_input: dict | None = None
) -> FlowResult:
@ -135,8 +158,12 @@ class KNXCommonFlow(ABC, FlowHandler):
return await self.async_step_tunnel()
# Automatic connection type
self.new_entry_data = KNXConfigEntryData(connection_type=CONF_KNX_AUTOMATIC)
return self.finish_flow(title=CONF_KNX_AUTOMATIC.capitalize())
self.new_entry_data = KNXConfigEntryData(
connection_type=CONF_KNX_AUTOMATIC,
tunnel_endpoint_ia=None,
)
self.new_title = CONF_KNX_AUTOMATIC.capitalize()
return self.finish_flow()
supported_connection_types = {
CONF_KNX_TUNNELING: CONF_KNX_TUNNELING.capitalize(),
@ -194,13 +221,18 @@ class KNXCommonFlow(ABC, FlowHandler):
port=self._selected_tunnel.port,
route_back=False,
connection_type=connection_type,
device_authentication=None,
user_id=None,
user_password=None,
tunnel_endpoint_ia=None,
)
if connection_type == CONF_KNX_TUNNELING_TCP_SECURE:
return self.async_show_menu(
step_id="secure_key_source",
menu_options=["secure_knxkeys", "secure_tunnel_manual"],
)
return self.finish_flow(title=f"Tunneling @ {self._selected_tunnel}")
self.new_title = f"Tunneling @ {self._selected_tunnel}"
return self.finish_flow()
if not self._found_tunnels:
return await self.async_step_manual_tunnel()
@ -264,6 +296,10 @@ class KNXCommonFlow(ABC, FlowHandler):
port=user_input[CONF_PORT],
route_back=user_input[CONF_KNX_ROUTE_BACK],
local_ip=_local_ip,
device_authentication=None,
user_id=None,
user_password=None,
tunnel_endpoint_ia=None,
)
if selected_tunnelling_type == CONF_KNX_TUNNELING_TCP_SECURE:
@ -271,7 +307,12 @@ class KNXCommonFlow(ABC, FlowHandler):
step_id="secure_key_source",
menu_options=["secure_knxkeys", "secure_tunnel_manual"],
)
return self.finish_flow(title=f"Tunneling @ {_host}")
self.new_title = (
"Tunneling "
f"{'UDP' if selected_tunnelling_type == CONF_KNX_TUNNELING else 'TCP'} "
f"@ {_host}"
)
return self.finish_flow()
_reconfiguring_existing_tunnel = (
self.initial_data.get(CONF_KNX_CONNECTION_TYPE)
@ -342,10 +383,10 @@ class KNXCommonFlow(ABC, FlowHandler):
device_authentication=user_input[CONF_KNX_SECURE_DEVICE_AUTHENTICATION],
user_id=user_input[CONF_KNX_SECURE_USER_ID],
user_password=user_input[CONF_KNX_SECURE_USER_PASSWORD],
tunnel_endpoint_ia=None,
)
return self.finish_flow(
title=f"Secure Tunneling @ {self.new_entry_data[CONF_HOST]}"
)
self.new_title = f"Secure Tunneling @ {self.new_entry_data[CONF_HOST]}"
return self.finish_flow()
fields = {
vol.Required(
@ -399,12 +440,8 @@ class KNXCommonFlow(ABC, FlowHandler):
CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE
],
)
return self.finish_flow(
title=(
"Secure Routing as"
f" {self.new_entry_data[CONF_KNX_INDIVIDUAL_ADDRESS]}"
)
)
self.new_title = f"Secure Routing as {self.new_entry_data[CONF_KNX_INDIVIDUAL_ADDRESS]}"
return self.finish_flow()
fields = {
vol.Required(
@ -437,92 +474,101 @@ class KNXCommonFlow(ABC, FlowHandler):
)
async def async_step_secure_knxkeys(
self, user_input: dict | None = None
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Configure secure knxkeys used to authenticate."""
errors = {}
description_placeholders = {}
"""Manage upload of new KNX Keyring file."""
errors: dict[str, str] = {}
if user_input is not None:
connection_type = self.new_entry_data[CONF_KNX_CONNECTION_TYPE]
storage_key = CONST_KNX_STORAGE_KEY + user_input[CONF_KNX_KNXKEY_FILENAME]
try:
keyring = await load_keyring(
path=self.hass.config.path(STORAGE_DIR, storage_key),
password=user_input[CONF_KNX_KNXKEY_PASSWORD],
)
except FileNotFoundError:
errors[CONF_KNX_KNXKEY_FILENAME] = "keyfile_not_found"
except InvalidSecureConfiguration:
errors[CONF_KNX_KNXKEY_PASSWORD] = "keyfile_invalid_signature"
else:
if (
connection_type == CONF_KNX_TUNNELING_TCP_SECURE
and self._selected_tunnel is not None
):
if host_ia := self._selected_tunnel.individual_address:
self._tunnel_endpoints = keyring.get_tunnel_interfaces_by_host(
host=host_ia
)
if not self._tunnel_endpoints:
errors["base"] = "keyfile_no_tunnel_for_host"
description_placeholders = {CONF_HOST: str(host_ia)}
if connection_type == CONF_KNX_ROUTING_SECURE:
if not (keyring.backbone is not None and keyring.backbone.key):
errors["base"] = "keyfile_no_backbone_key"
if not errors:
password = user_input[CONF_KNX_KNXKEY_PASSWORD]
errors = await self._save_uploaded_knxkeys_file(
uploaded_file_id=user_input[CONF_KEYRING_FILE],
password=password,
)
if not errors and self._keyring:
self.new_entry_data |= KNXConfigEntryData(
knxkeys_filename=storage_key,
knxkeys_password=user_input[CONF_KNX_KNXKEY_PASSWORD],
knxkeys_filename=f"{DOMAIN}/{DEFAULT_KNX_KEYRING_FILENAME}",
knxkeys_password=password,
backbone_key=None,
sync_latency_tolerance=None,
device_authentication=None,
user_id=None,
user_password=None,
)
if connection_type == CONF_KNX_ROUTING_SECURE:
return self.finish_flow(
title=(
"Secure Routing as"
f" {self.new_entry_data[CONF_KNX_INDIVIDUAL_ADDRESS]}"
)
)
return await self.async_step_knxkeys_tunnel_select()
# Routing
if self.connection_type in (CONF_KNX_ROUTING, CONF_KNX_ROUTING_SECURE):
return self.finish_flow()
# Tunneling / Automatic
# skip selection step if we have a keyfile update that includes a configured tunnel
if self.tunnel_endpoint_ia is not None and self.tunnel_endpoint_ia in [
str(_if.individual_address) for _if in self._keyring.interfaces
]:
return self.finish_flow()
if not errors:
return await self.async_step_knxkeys_tunnel_select()
if _default_filename := self.initial_data.get(CONF_KNX_KNXKEY_FILENAME):
_default_filename = _default_filename.lstrip(CONST_KNX_STORAGE_KEY)
fields = {
vol.Required(
CONF_KNX_KNXKEY_FILENAME, default=_default_filename
): selector.TextSelector(),
vol.Required(CONF_KEYRING_FILE): selector.FileSelector(
config=selector.FileSelectorConfig(accept=".knxkeys")
),
vol.Required(
CONF_KNX_KNXKEY_PASSWORD,
default=self.initial_data.get(CONF_KNX_KNXKEY_PASSWORD),
): selector.TextSelector(),
}
return self.async_show_form(
step_id="secure_knxkeys",
data_schema=vol.Schema(fields),
errors=errors,
description_placeholders=description_placeholders,
)
async def async_step_knxkeys_tunnel_select(
self, user_input: dict | None = None
) -> FlowResult:
"""Select if a specific tunnel should be used from knxkeys file."""
errors = {}
description_placeholders = {}
if user_input is not None:
if user_input[CONF_KNX_SECURE_USER_ID] == CONF_KNX_AUTOMATIC:
selected_user_id = None
selected_tunnel_ia: str | None = None
_if_user_id: int | None = None
if user_input[CONF_KNX_TUNNEL_ENDPOINT_IA] == CONF_KNX_AUTOMATIC:
self.new_entry_data |= KNXConfigEntryData(
tunnel_endpoint_ia=None,
)
else:
selected_user_id = int(user_input[CONF_KNX_SECURE_USER_ID])
self.new_entry_data |= KNXConfigEntryData(user_id=selected_user_id)
return self.finish_flow(
title=f"Secure Tunneling @ {self.new_entry_data[CONF_HOST]}"
selected_tunnel_ia = user_input[CONF_KNX_TUNNEL_ENDPOINT_IA]
self.new_entry_data |= KNXConfigEntryData(
tunnel_endpoint_ia=selected_tunnel_ia,
user_id=None,
user_password=None,
device_authentication=None,
)
_if_user_id = next(
(
_if.user_id
for _if in self._tunnel_endpoints
if str(_if.individual_address) == selected_tunnel_ia
),
None,
)
self.new_title = (
f"{'Secure ' if _if_user_id else ''}"
f"Tunneling @ {selected_tunnel_ia or self.new_entry_data[CONF_HOST]}"
)
return self.finish_flow()
# this step is only called from async_step_secure_knxkeys so self._keyring is always set
assert self._keyring
# Filter for selected tunnel
if self._selected_tunnel is not None:
if host_ia := self._selected_tunnel.individual_address:
self._tunnel_endpoints = self._keyring.get_tunnel_interfaces_by_host(
host=host_ia
)
if not self._tunnel_endpoints:
errors["base"] = "keyfile_no_tunnel_for_host"
description_placeholders = {CONF_HOST: str(host_ia)}
else:
self._tunnel_endpoints = self._keyring.interfaces
tunnel_endpoint_options = [
selector.SelectOptionDict(
@ -532,8 +578,12 @@ class KNXCommonFlow(ABC, FlowHandler):
for endpoint in self._tunnel_endpoints:
tunnel_endpoint_options.append(
selector.SelectOptionDict(
value=str(endpoint.user_id),
label=f"{endpoint.individual_address} (User ID: {endpoint.user_id})",
value=str(endpoint.individual_address),
label=(
f"{endpoint.individual_address} "
f"{'🔐 ' if endpoint.user_id else ''}"
f"(Data Secure GAs: {len(endpoint.group_addresses)})"
),
)
)
return self.async_show_form(
@ -541,7 +591,7 @@ class KNXCommonFlow(ABC, FlowHandler):
data_schema=vol.Schema(
{
vol.Required(
CONF_KNX_SECURE_USER_ID, default=CONF_KNX_AUTOMATIC
CONF_KNX_TUNNEL_ENDPOINT_IA, default=CONF_KNX_AUTOMATIC
): selector.SelectSelector(
selector.SelectSelectorConfig(
options=tunnel_endpoint_options,
@ -550,6 +600,8 @@ class KNXCommonFlow(ABC, FlowHandler):
),
}
),
errors=errors,
description_placeholders=description_placeholders,
)
async def async_step_routing(self, user_input: dict | None = None) -> FlowResult:
@ -598,13 +650,19 @@ class KNXCommonFlow(ABC, FlowHandler):
multicast_group=_multicast_group,
multicast_port=_multicast_port,
local_ip=_local_ip,
device_authentication=None,
user_id=None,
user_password=None,
tunnel_endpoint_ia=None,
)
if connection_type == CONF_KNX_ROUTING_SECURE:
self.new_title = f"Secure Routing as {_individual_address}"
return self.async_show_menu(
step_id="secure_key_source",
menu_options=["secure_knxkeys", "secure_routing_manual"],
)
return self.finish_flow(title=f"Routing as {_individual_address}")
self.new_title = f"Routing as {_individual_address}"
return self.finish_flow()
routers = [router for router in self._found_gateways if router.supports_routing]
if not routers:
@ -631,6 +689,32 @@ class KNXCommonFlow(ABC, FlowHandler):
step_id="routing", data_schema=vol.Schema(fields), errors=errors
)
async def _save_uploaded_knxkeys_file(
self, uploaded_file_id: str, password: str
) -> dict[str, str]:
"""Validate the uploaded file and move it to the storage directory. Return errors."""
def _process_upload() -> tuple[Keyring | None, dict[str, str]]:
keyring: Keyring | None = None
errors = {}
with process_uploaded_file(self.hass, uploaded_file_id) as file_path:
try:
keyring = sync_load_keyring(
path=file_path,
password=password,
)
except InvalidSecureConfiguration:
errors[CONF_KNX_KNXKEY_PASSWORD] = "keyfile_invalid_signature"
else:
dest_path = Path(self.hass.config.path(STORAGE_DIR, DOMAIN))
dest_path.mkdir(exist_ok=True)
file_path.rename(dest_path / DEFAULT_KNX_KEYRING_FILENAME)
return keyring, errors
keyring, errors = await self.hass.async_add_executor_job(_process_upload)
self._keyring = keyring
return errors
class KNXConfigFlow(KNXCommonFlow, ConfigFlow, domain=DOMAIN):
"""Handle a KNX config flow."""
@ -648,8 +732,9 @@ class KNXConfigFlow(KNXCommonFlow, ConfigFlow, domain=DOMAIN):
return KNXOptionsFlow(config_entry)
@callback
def finish_flow(self, title: str) -> FlowResult:
def finish_flow(self) -> FlowResult:
"""Create the ConfigEntry."""
title = self.new_title or f"KNX {self.new_entry_data[CONF_KNX_CONNECTION_TYPE]}"
return self.async_create_entry(
title=title,
data=DEFAULT_ENTRY_DATA | self.new_entry_data,
@ -673,13 +758,13 @@ class KNXOptionsFlow(KNXCommonFlow, OptionsFlow):
super().__init__(initial_data=config_entry.data) # type: ignore[arg-type]
@callback
def finish_flow(self, title: str | None) -> FlowResult:
def finish_flow(self) -> FlowResult:
"""Update the ConfigEntry and finish the flow."""
new_data = DEFAULT_ENTRY_DATA | self.initial_data | self.new_entry_data
self.hass.config_entries.async_update_entry(
self.config_entry,
data=new_data,
title=title or UNDEFINED,
title=self.new_title or UNDEFINED,
)
return self.async_create_entry(title="", data={})
@ -689,7 +774,11 @@ class KNXOptionsFlow(KNXCommonFlow, OptionsFlow):
"""Manage KNX options."""
return self.async_show_menu(
step_id="options_init",
menu_options=["connection_type", "communication_settings"],
menu_options=[
"connection_type",
"communication_settings",
"secure_knxkeys",
],
)
async def async_step_communication_settings(
@ -701,7 +790,7 @@ class KNXOptionsFlow(KNXCommonFlow, OptionsFlow):
state_updater=user_input[CONF_KNX_STATE_UPDATER],
rate_limit=user_input[CONF_KNX_RATE_LIMIT],
)
return self.finish_flow(title=None)
return self.finish_flow()
data_schema = {
vol.Required(

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

@ -39,6 +39,7 @@ CONF_KNX_TUNNELING_TCP_SECURE: Final = "tunneling_tcp_secure"
CONF_KNX_LOCAL_IP: Final = "local_ip"
CONF_KNX_MCAST_GRP: Final = "multicast_group"
CONF_KNX_MCAST_PORT: Final = "multicast_port"
CONF_KNX_TUNNEL_ENDPOINT_IA: Final = "tunnel_endpoint_ia"
CONF_KNX_RATE_LIMIT: Final = "rate_limit"
CONF_KNX_ROUTE_BACK: Final = "route_back"
@ -89,6 +90,7 @@ class KNXConfigEntryData(TypedDict, total=False):
rate_limit: int
host: str
port: int
tunnel_endpoint_ia: str | None
user_id: int | None
user_password: str | None

1
homeassistant/components/knx/manifest.json
View file

@ -3,6 +3,7 @@
"name": "KNX",
"codeowners": ["@Julius2342", "@farmio", "@marvin-w"],
"config_flow": true,
"dependencies": ["file_upload"],
"documentation": "https://www.home-assistant.io/integrations/knx",
"integration_type": "hub",
"iot_class": "local_push",

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

@ -42,14 +42,13 @@
}
},
"secure_knxkeys": {
"title": "Keyfile",
"description": "Please enter the information for your `.knxkeys` file.",
"title": "Import KNX Keyring",
"description": "Please select a `.knxkeys` file to import.",
"data": {
"knxkeys_filename": "The filename of your `.knxkeys` file (including extension)",
"knxkeys_file": "Keyring file",
"knxkeys_password": "The password to decrypt the `.knxkeys` file"
},
"data_description": {
"knxkeys_filename": "The file is expected to be found in your config directory in `.storage/knx/`.\nIn Home Assistant OS this would be `/config/.storage/knx/`\nExample: `my_project.knxkeys`",
"knxkeys_password": "This was set when exporting the file from ETS."
}
},
@ -126,7 +125,8 @@
"title": "KNX Settings",
"menu_options": {
"connection_type": "Configure KNX interface",
"communication_settings": "Communication settings"
"communication_settings": "Communication settings",
"secure_knxkeys": "Import a `.knxkeys` file"
}
},
"communication_settings": {
@ -184,11 +184,10 @@
"title": "[%key:component::knx::config::step::secure_knxkeys::title%]",
"description": "[%key:component::knx::config::step::secure_knxkeys::description%]",
"data": {
"knxkeys_filename": "[%key:component::knx::config::step::secure_knxkeys::data::knxkeys_filename%]",
"knxkeys_file": "[%key:component::knx::config::step::secure_knxkeys::data::knxkeys_file%]",
"knxkeys_password": "[%key:component::knx::config::step::secure_knxkeys::data::knxkeys_password%]"
},
"data_description": {
"knxkeys_filename": "[%key:component::knx::config::step::secure_knxkeys::data_description::knxkeys_filename%]",
"knxkeys_password": "[%key:component::knx::config::step::secure_knxkeys::data_description::knxkeys_password%]"
}
},

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

@ -1,4 +1,5 @@
"""Test the KNX config flow."""
from contextlib import contextmanager
from unittest.mock import Mock, patch
import pytest
@ -10,6 +11,7 @@ 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_TUNNELING_TYPE,
DEFAULT_ENTRY_DATA,
@ -35,6 +37,7 @@ from homeassistant.components.knx.const import (
CONF_KNX_SECURE_USER_ID,
CONF_KNX_SECURE_USER_PASSWORD,
CONF_KNX_STATE_UPDATER,
CONF_KNX_TUNNEL_ENDPOINT_IA,
CONF_KNX_TUNNELING,
CONF_KNX_TUNNELING_TCP,
CONF_KNX_TUNNELING_TCP_SECURE,
@ -47,6 +50,10 @@ from homeassistant.data_entry_flow import FlowResult, 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")
@ -59,6 +66,29 @@ def fixture_knx_setup():
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.config_flow.process_uploaded_file"
) as file_upload_mock, patch(
"homeassistant.components.knx.config_flow.sync_load_keyring",
return_value=return_value,
side_effect=side_effect,
), patch(
"pathlib.Path.mkdir"
) as mkdir_mock:
file_path_mock = Mock()
file_upload_mock.return_value.__enter__.return_value = file_path_mock
yield return_value
if side_effect:
mkdir_mock.assert_not_called()
file_path_mock.rename.assert_not_called()
else:
mkdir_mock.assert_called_once()
file_path_mock.rename.assert_called_once()
def _gateway_descriptor(
ip: str,
port: int,
@ -153,6 +183,10 @@ async def test_routing_setup(
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,
}
knx_setup.assert_called_once()
@ -224,6 +258,10 @@ async def test_routing_setup_advanced(
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,
}
knx_setup.assert_called_once()
@ -310,6 +348,10 @@ async def test_routing_secure_manual_setup(
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,
}
knx_setup.assert_called_once()
@ -358,29 +400,11 @@ async def test_routing_secure_keyfile(
assert result4["step_id"] == "secure_knxkeys"
assert not result4["errors"]
# test file without backbone key
with patch(
"homeassistant.components.knx.config_flow.load_keyring"
) as mock_load_keyring:
mock_keyring = Mock()
mock_keyring.backbone.key = None
mock_load_keyring.return_value = mock_keyring
secure_knxkeys = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys",
CONF_KNX_KNXKEY_PASSWORD: "password",
},
)
assert secure_knxkeys["type"] == FlowResultType.FORM
assert secure_knxkeys["errors"] == {"base": "keyfile_no_backbone_key"}
# test valid file
with patch("homeassistant.components.knx.config_flow.load_keyring"):
with patch_file_upload():
routing_secure_knxkeys = await hass.config_entries.flow.async_configure(
result4["flow_id"],
{
CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys",
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
CONF_KNX_KNXKEY_PASSWORD: "password",
},
)
@ -390,20 +414,21 @@ async def test_routing_secure_keyfile(
assert routing_secure_knxkeys["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING_SECURE,
CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys",
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",
}
knx_setup.assert_called_once()
@pytest.mark.parametrize(
("user_input", "config_entry_data"),
("user_input", "title", "config_entry_data"),
[
(
{
@ -412,6 +437,7 @@ async def test_routing_secure_keyfile(
CONF_PORT: 3675,
CONF_KNX_ROUTE_BACK: False,
},
"Tunneling UDP @ 192.168.0.1",
{
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
@ -420,6 +446,10 @@ async def test_routing_secure_keyfile(
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,
},
),
(
@ -429,6 +459,7 @@ async def test_routing_secure_keyfile(
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,
@ -437,6 +468,10 @@ async def test_routing_secure_keyfile(
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,
},
),
(
@ -446,6 +481,7 @@ async def test_routing_secure_keyfile(
CONF_PORT: 3675,
CONF_KNX_ROUTE_BACK: True,
},
"Tunneling UDP @ 192.168.0.1",
{
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
@ -454,6 +490,10 @@ async def test_routing_secure_keyfile(
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,
},
),
],
@ -467,6 +507,7 @@ async def test_tunneling_setup_manual(
hass: HomeAssistant,
knx_setup,
user_input,
title,
config_entry_data,
) -> None:
"""Test tunneling if no gateway was found found (or `manual` option was chosen)."""
@ -502,7 +543,7 @@ async def test_tunneling_setup_manual(
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "Tunneling @ 192.168.0.1"
assert result3["title"] == title
assert result3["data"] == config_entry_data
knx_setup.assert_called_once()
@ -631,12 +672,16 @@ async def test_tunneling_setup_manual_request_description_error(
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "Tunneling @ 192.168.0.1"
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,
}
knx_setup.assert_called_once()
@ -718,7 +763,7 @@ async def test_tunneling_setup_for_local_ip(
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "Tunneling @ 192.168.0.2"
assert result3["title"] == "Tunneling UDP @ 192.168.0.2"
assert result3["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
@ -727,6 +772,10 @@ async def test_tunneling_setup_for_local_ip(
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,
}
knx_setup.assert_called_once()
@ -771,6 +820,10 @@ async def test_tunneling_setup_for_multiple_found_gateways(
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,
}
knx_setup.assert_called_once()
@ -848,11 +901,12 @@ async def test_form_with_automatic_connection_handling(
assert result2["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_TUNNEL_ENDPOINT_IA: None,
}
knx_setup.assert_called_once()
async def _get_menu_step(hass: HomeAssistant) -> FlowResult:
async def _get_menu_step_secure_tunnel(hass: HomeAssistant) -> FlowResult:
"""Return flow in secure_tunnelling menu step."""
gateway = _gateway_descriptor(
"192.168.0.1",
@ -950,7 +1004,7 @@ async def test_get_secure_menu_step_manual_tunnelling(
async def test_configure_secure_tunnel_manual(hass: HomeAssistant, knx_setup) -> None:
"""Test configure tunnelling secure keys manually."""
menu_step = await _get_menu_step(hass)
menu_step = await _get_menu_step_secure_tunnel(hass)
result = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
@ -981,13 +1035,14 @@ async def test_configure_secure_tunnel_manual(hass: HomeAssistant, knx_setup) ->
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240",
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: None,
CONF_KNX_TUNNEL_ENDPOINT_IA: None,
}
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(hass)
menu_step = await _get_menu_step_secure_tunnel(hass)
result = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
@ -997,17 +1052,11 @@ async def test_configure_secure_knxkeys(hass: HomeAssistant, knx_setup) -> None:
assert result["step_id"] == "secure_knxkeys"
assert not result["errors"]
with patch(
"xknx.secure.keyring.sync_load_keyring",
return_value=sync_load_keyring(
str(get_fixture_path("fixture.knxkeys", DOMAIN).absolute()),
FIXTURE_KNXKEYS_PASSWORD,
),
):
with patch_file_upload():
secure_knxkeys = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys",
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
CONF_KNX_KNXKEY_PASSWORD: "test",
},
)
@ -1016,7 +1065,7 @@ async def test_configure_secure_knxkeys(hass: HomeAssistant, knx_setup) -> None:
assert not result["errors"]
secure_knxkeys = await hass.config_entries.flow.async_configure(
secure_knxkeys["flow_id"],
{CONF_KNX_SECURE_USER_ID: CONF_KNX_AUTOMATIC},
{CONF_KNX_TUNNEL_ENDPOINT_IA: CONF_KNX_AUTOMATIC},
)
await hass.async_block_till_done()
@ -1024,13 +1073,14 @@ async def test_configure_secure_knxkeys(hass: HomeAssistant, knx_setup) -> None:
assert secure_knxkeys["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys",
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",
@ -1040,37 +1090,9 @@ async def test_configure_secure_knxkeys(hass: HomeAssistant, knx_setup) -> None:
knx_setup.assert_called_once()
async def test_configure_secure_knxkeys_file_not_found(hass: HomeAssistant) -> None:
"""Test configure secure knxkeys but file was not found."""
menu_step = await _get_menu_step(hass)
result = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
{"next_step_id": "secure_knxkeys"},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "secure_knxkeys"
assert not result["errors"]
with patch(
"homeassistant.components.knx.config_flow.load_keyring",
side_effect=FileNotFoundError(),
):
secure_knxkeys = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys",
CONF_KNX_KNXKEY_PASSWORD: "password",
},
)
assert secure_knxkeys["type"] == FlowResultType.FORM
assert secure_knxkeys["errors"]
assert secure_knxkeys["errors"][CONF_KNX_KNXKEY_FILENAME] == "keyfile_not_found"
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(hass)
menu_step = await _get_menu_step_secure_tunnel(hass)
result = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
@ -1080,14 +1102,13 @@ async def test_configure_secure_knxkeys_invalid_signature(hass: HomeAssistant) -
assert result["step_id"] == "secure_knxkeys"
assert not result["errors"]
with patch(
"homeassistant.components.knx.config_flow.load_keyring",
with patch_file_upload(
side_effect=InvalidSecureConfiguration(),
):
secure_knxkeys = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys",
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
CONF_KNX_KNXKEY_PASSWORD: "password",
},
)
@ -1101,7 +1122,7 @@ async def test_configure_secure_knxkeys_invalid_signature(hass: HomeAssistant) -
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(hass)
menu_step = await _get_menu_step_secure_tunnel(hass)
result = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
@ -1111,16 +1132,12 @@ async def test_configure_secure_knxkeys_no_tunnel_for_host(hass: HomeAssistant)
assert result["step_id"] == "secure_knxkeys"
assert not result["errors"]
with patch(
"homeassistant.components.knx.config_flow.load_keyring"
) as mock_load_keyring:
mock_keyring = Mock()
with patch_file_upload(return_value=Mock()) as mock_keyring:
mock_keyring.get_tunnel_interfaces_by_host.return_value = []
mock_load_keyring.return_value = mock_keyring
secure_knxkeys = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys",
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
CONF_KNX_KNXKEY_PASSWORD: "password",
},
)
@ -1180,6 +1197,10 @@ async def test_options_flow_connection_type(
CONF_KNX_RATE_LIMIT: 0,
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,
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,
}
@ -1251,17 +1272,11 @@ async def test_options_flow_secure_manual_to_keyfile(
assert result4["step_id"] == "secure_knxkeys"
assert not result4["errors"]
with patch(
"xknx.secure.keyring.sync_load_keyring",
return_value=sync_load_keyring(
str(get_fixture_path("fixture.knxkeys", DOMAIN).absolute()),
FIXTURE_KNXKEYS_PASSWORD,
),
):
with patch_file_upload():
secure_knxkeys = await hass.config_entries.options.async_configure(
result4["flow_id"],
{
CONF_KNX_KNXKEY_FILENAME: "testcase.knxkeys",
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
CONF_KNX_KNXKEY_PASSWORD: "test",
},
)
@ -1270,7 +1285,7 @@ async def test_options_flow_secure_manual_to_keyfile(
assert not result["errors"]
secure_knxkeys = await hass.config_entries.options.async_configure(
secure_knxkeys["flow_id"],
{CONF_KNX_SECURE_USER_ID: "2"},
{CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1"},
)
await hass.async_block_till_done()
@ -1278,11 +1293,12 @@ async def test_options_flow_secure_manual_to_keyfile(
assert mock_config_entry.data == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys",
CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys",
CONF_KNX_KNXKEY_PASSWORD: "test",
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None,
CONF_KNX_SECURE_USER_ID: 2,
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",
@ -1326,3 +1342,118 @@ async def test_options_communication_settings(
CONF_KNX_RATE_LIMIT: 40,
}
knx_setup.assert_called_once()
async def test_options_update_keyfile(hass: HomeAssistant, knx_setup) -> None:
"""Test options flow updating keyfile when tunnel endpoint is already 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="knx",
data=start_data,
)
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)
result = await hass.config_entries.options.async_configure(
menu_step["flow_id"],
{"next_step_id": "secure_knxkeys"},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "secure_knxkeys"
with patch_file_upload():
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
{
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
CONF_KNX_KNXKEY_PASSWORD: "password",
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert not result2.get("data")
assert mock_config_entry.data == {
**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_options_keyfile_upload(hass: HomeAssistant, knx_setup) -> None:
"""Test options 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="knx",
data=start_data,
)
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)
result = await hass.config_entries.options.async_configure(
menu_step["flow_id"],
{"next_step_id": "secure_knxkeys"},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "secure_knxkeys"
with patch_file_upload():
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
{
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
CONF_KNX_KNXKEY_PASSWORD: "password",
},
)
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "knxkeys_tunnel_select"
result3 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1",
},
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert not result3.get("data")
assert mock_config_entry.data == {
**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()

35
tests/components/knx/test_init.py
View file

@ -1,4 +1,6 @@
"""Test KNX init."""
from unittest.mock import patch
import pytest
from xknx.io import (
DEFAULT_MCAST_GRP,
@ -8,6 +10,7 @@ from xknx.io import (
SecureConfig,
)
from homeassistant import config_entries
from homeassistant.components.knx.config_flow import DEFAULT_ROUTING_IA
from homeassistant.components.knx.const import (
CONF_KNX_AUTOMATIC,
@ -110,12 +113,17 @@ from tests.common import MockConfigEntry
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_INDIVIDUAL_ADDRESS: DEFAULT_ROUTING_IA,
CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys",
CONF_KNX_KNXKEY_PASSWORD: "password",
},
ConnectionConfig(
connection_type=ConnectionType.TUNNELING_TCP,
gateway_ip="192.168.0.2",
gateway_port=3675,
auto_reconnect=True,
secure_config=SecureConfig(
knxkeys_file_path="keyring.knxkeys", knxkeys_password="password"
),
threaded=True,
),
),
@ -251,3 +259,30 @@ async def test_init_connection_handling(
.connection_config()
.secure_config.knxkeys_file_path
)
async def test_async_remove_entry(
hass: HomeAssistant,
knx: KNXTestKit,
) -> None:
"""Test async_setup_entry (for coverage)."""
config_entry = MockConfigEntry(
title="KNX",
domain=KNX_DOMAIN,
data={
CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys",
},
)
knx.mock_config_entry = config_entry
await knx.setup_integration({})
with patch("pathlib.Path.unlink") as unlink_mock, patch(
"pathlib.Path.rmdir"
) as rmdir_mock:
assert await hass.config_entries.async_remove(config_entry.entry_id)
unlink_mock.assert_called_once()
rmdir_mock.assert_called_once()
await hass.async_block_till_done()
assert hass.config_entries.async_entries() == []
assert config_entry.state is config_entries.ConfigEntryState.NOT_LOADED