Compare commits

...

3 Commits

Author SHA1 Message Date
Matthias Alphart
87f56e3ae7 typo in comment
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
2025-06-11 22:37:37 +02:00
farmio
a2dddda8cd Remove connection options from OptionsFlow 2025-06-03 11:33:55 +02:00
farmio
bede0f5b45 Add reconfigure flow for KNX 2025-06-03 11:33:55 +02:00
5 changed files with 156 additions and 291 deletions

View File

@@ -2,7 +2,6 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from collections.abc import AsyncGenerator
from typing import Any, Final, Literal
@@ -20,8 +19,8 @@ from xknx.io.util import validate_ip as xknx_validate_ip
from xknx.secure.keyring import Keyring, XMLInterface
from homeassistant.config_entries import (
SOURCE_RECONFIGURE,
ConfigEntry,
ConfigEntryBaseFlow,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
@@ -103,12 +102,14 @@ _PORT_SELECTOR = vol.All(
)
class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
"""Base class for KNX flows."""
class KNXConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a KNX config flow."""
def __init__(self, initial_data: KNXConfigEntryData) -> None:
"""Initialize KNXCommonFlow."""
self.initial_data = initial_data
VERSION = 1
def __init__(self) -> None:
"""Initialize KNX config flow."""
self.initial_data = DEFAULT_ENTRY_DATA
self.new_entry_data = KNXConfigEntryData()
self.new_title: str | None = None
@@ -121,19 +122,21 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
self._gatewayscanner: GatewayScanner | None = None
self._async_scan_gen: AsyncGenerator[GatewayDescriptor] | None = None
@staticmethod
@callback
def async_get_options_flow(config_entry: ConfigEntry) -> KNXOptionsFlow:
"""Get the options flow for this handler."""
return KNXOptionsFlow(config_entry)
@property
def _xknx(self) -> XKNX:
"""Return XKNX instance."""
if isinstance(self, OptionsFlow) and (
if (self.source == SOURCE_RECONFIGURE) 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."""
@property
def connection_type(self) -> str:
"""Return the configured connection type."""
@@ -150,6 +153,40 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
self.initial_data.get(CONF_KNX_TUNNEL_ENDPOINT_IA),
)
@callback
def finish_flow(self) -> ConfigFlowResult:
"""Create or update the ConfigEntry."""
if self.source == SOURCE_RECONFIGURE:
return self.async_update_reload_and_abort(
self._get_reconfigure_entry(),
data_updates=self.new_entry_data,
title=self.new_title or UNDEFINED,
)
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,
)
async def async_step_user(self, user_input: dict | None = None) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
return await self.async_step_connection_type()
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle reconfiguration of existing entry."""
entry = self._get_reconfigure_entry()
self.initial_data = dict(entry.data) # type: ignore[assignment]
return self.async_show_menu(
step_id="reconfigure",
menu_options=[
"connection_type",
"secure_knxkeys",
],
)
async def async_step_connection_type(
self, user_input: dict | None = None
) -> ConfigFlowResult:
@@ -441,7 +478,7 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
)
ip_address: str | None
if ( # initial attempt on ConfigFlow or coming from automatic / routing
(isinstance(self, ConfigFlow) or not _reconfiguring_existing_tunnel)
not _reconfiguring_existing_tunnel
and not user_input
and self._selected_tunnel is not None
): # default to first found tunnel
@@ -841,52 +878,20 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
)
class KNXConfigFlow(KNXCommonFlow, ConfigFlow, domain=DOMAIN):
"""Handle a KNX config flow."""
VERSION = 1
def __init__(self) -> None:
"""Initialize KNX options flow."""
super().__init__(initial_data=DEFAULT_ENTRY_DATA)
@staticmethod
@callback
def async_get_options_flow(config_entry: ConfigEntry) -> KNXOptionsFlow:
"""Get the options flow for this handler."""
return KNXOptionsFlow(config_entry)
@callback
def finish_flow(self) -> ConfigFlowResult:
"""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,
)
async def async_step_user(self, user_input: dict | None = None) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
return await self.async_step_connection_type()
class KNXOptionsFlow(KNXCommonFlow, OptionsFlow):
class KNXOptionsFlow(OptionsFlow):
"""Handle KNX options."""
general_settings: dict
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize KNX options flow."""
super().__init__(initial_data=config_entry.data) # type: ignore[arg-type]
self.initial_data = dict(config_entry.data)
@callback
def finish_flow(self) -> ConfigFlowResult:
def finish_flow(self, new_entry_data: KNXConfigEntryData) -> ConfigFlowResult:
"""Update the ConfigEntry and finish the flow."""
new_data = DEFAULT_ENTRY_DATA | self.initial_data | self.new_entry_data
new_data = self.initial_data | new_entry_data
self.hass.config_entries.async_update_entry(
self.config_entry,
data=new_data,
title=self.new_title or UNDEFINED,
)
return self.async_create_entry(title="", data={})
@@ -894,26 +899,20 @@ class KNXOptionsFlow(KNXCommonFlow, OptionsFlow):
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage KNX options."""
return self.async_show_menu(
step_id="init",
menu_options=[
"connection_type",
"communication_settings",
"secure_knxkeys",
],
)
return await self.async_step_communication_settings()
async def async_step_communication_settings(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage KNX communication settings."""
if user_input is not None:
self.new_entry_data = KNXConfigEntryData(
state_updater=user_input[CONF_KNX_STATE_UPDATER],
rate_limit=user_input[CONF_KNX_RATE_LIMIT],
telegram_log_size=user_input[CONF_KNX_TELEGRAM_LOG_SIZE],
return self.finish_flow(
KNXConfigEntryData(
state_updater=user_input[CONF_KNX_STATE_UPDATER],
rate_limit=user_input[CONF_KNX_RATE_LIMIT],
telegram_log_size=user_input[CONF_KNX_TELEGRAM_LOG_SIZE],
)
)
return self.finish_flow()
data_schema = {
vol.Required(

View File

@@ -101,7 +101,7 @@ rules:
Since all entities are configured manually, names are user-defined.
exception-translations: done
icon-translations: done
reconfiguration-flow: todo
reconfiguration-flow: done
repair-issues: todo
stale-devices:
status: exempt

View File

@@ -1,6 +1,13 @@
{
"config": {
"step": {
"reconfigure": {
"title": "KNX connection settings",
"menu_options": {
"connection_type": "Reconfigure KNX connection",
"secure_knxkeys": "Import KNX keyring file"
}
},
"connection_type": {
"title": "KNX connection",
"description": "'Automatic' performs a gateway scan on start, to find a KNX IP interface. It will connect via a tunnel. (Not available if a gateway scan was not successful.)\n\n'Tunneling' will connect to a specific KNX IP interface over a tunnel.\n\n'Routing' will use Multicast to communicate with KNX IP routers.",
@@ -65,7 +72,7 @@
},
"secure_knxkeys": {
"title": "Import KNX Keyring",
"description": "The Keyring is used to encrypt and decrypt KNX IP Secure communication.",
"description": "The keyring is used to encrypt and decrypt KNX IP Secure communication. You can import a new keyring file or re-import to update existing keys if your configuration has changed.",
"data": {
"knxkeys_file": "Keyring file",
"knxkeys_password": "Keyring password"
@@ -129,6 +136,9 @@
}
}
},
"abort": {
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_backbone_key": "Invalid backbone key. 32 hexadecimal numbers expected.",
@@ -159,16 +169,8 @@
},
"options": {
"step": {
"init": {
"title": "KNX Settings",
"menu_options": {
"connection_type": "Configure KNX interface",
"communication_settings": "Communication settings",
"secure_knxkeys": "Import a `.knxkeys` file"
}
},
"communication_settings": {
"title": "[%key:component::knx::options::step::init::menu_options::communication_settings%]",
"title": "Communication settings",
"data": {
"state_updater": "State updater",
"rate_limit": "Rate limit",
@@ -179,147 +181,7 @@
"rate_limit": "Maximum outgoing telegrams per second.\n`0` to disable limit. Recommended: `0` or between `20` and `40`",
"telegram_log_size": "Telegrams to keep in memory for KNX panel group monitor. Maximum: {telegram_log_size_max}"
}
},
"connection_type": {
"title": "[%key:component::knx::config::step::connection_type::title%]",
"description": "[%key:component::knx::config::step::connection_type::description%]",
"data": {
"connection_type": "[%key:component::knx::config::step::connection_type::data::connection_type%]"
},
"data_description": {
"connection_type": "[%key:component::knx::config::step::connection_type::data_description::connection_type%]"
}
},
"tunnel": {
"title": "[%key:component::knx::config::step::tunnel::title%]",
"data": {
"gateway": "[%key:component::knx::config::step::tunnel::data::gateway%]"
},
"data_description": {
"gateway": "[%key:component::knx::config::step::tunnel::data_description::gateway%]"
}
},
"tcp_tunnel_endpoint": {
"title": "[%key:component::knx::config::step::tcp_tunnel_endpoint::title%]",
"data": {
"tunnel_endpoint_ia": "[%key:component::knx::config::step::tcp_tunnel_endpoint::data::tunnel_endpoint_ia%]"
},
"data_description": {
"tunnel_endpoint_ia": "[%key:component::knx::config::step::tcp_tunnel_endpoint::data_description::tunnel_endpoint_ia%]"
}
},
"manual_tunnel": {
"title": "[%key:component::knx::config::step::manual_tunnel::title%]",
"description": "[%key:component::knx::config::step::manual_tunnel::description%]",
"data": {
"tunneling_type": "[%key:component::knx::config::step::manual_tunnel::data::tunneling_type%]",
"port": "[%key:common::config_flow::data::port%]",
"host": "[%key:common::config_flow::data::host%]",
"route_back": "[%key:component::knx::config::step::manual_tunnel::data::route_back%]",
"local_ip": "[%key:component::knx::config::step::manual_tunnel::data::local_ip%]"
},
"data_description": {
"tunneling_type": "[%key:component::knx::config::step::manual_tunnel::data_description::tunneling_type%]",
"port": "[%key:component::knx::config::step::manual_tunnel::data_description::port%]",
"host": "[%key:component::knx::config::step::manual_tunnel::data_description::host%]",
"route_back": "[%key:component::knx::config::step::manual_tunnel::data_description::route_back%]",
"local_ip": "[%key:component::knx::config::step::manual_tunnel::data_description::local_ip%]"
}
},
"secure_key_source_menu_tunnel": {
"title": "[%key:component::knx::config::step::secure_key_source_menu_tunnel::title%]",
"description": "[%key:component::knx::config::step::secure_key_source_menu_tunnel::description%]",
"menu_options": {
"secure_knxkeys": "[%key:component::knx::config::step::secure_key_source_menu_tunnel::menu_options::secure_knxkeys%]",
"secure_tunnel_manual": "[%key:component::knx::config::step::secure_key_source_menu_tunnel::menu_options::secure_tunnel_manual%]"
}
},
"secure_key_source_menu_routing": {
"title": "[%key:component::knx::config::step::secure_key_source_menu_tunnel::title%]",
"description": "[%key:component::knx::config::step::secure_key_source_menu_tunnel::description%]",
"menu_options": {
"secure_knxkeys": "[%key:component::knx::config::step::secure_key_source_menu_tunnel::menu_options::secure_knxkeys%]",
"secure_routing_manual": "[%key:component::knx::config::step::secure_key_source_menu_routing::menu_options::secure_routing_manual%]"
}
},
"secure_knxkeys": {
"title": "[%key:component::knx::config::step::secure_knxkeys::title%]",
"description": "[%key:component::knx::config::step::secure_knxkeys::description%]",
"data": {
"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_file": "[%key:component::knx::config::step::secure_knxkeys::data_description::knxkeys_file%]",
"knxkeys_password": "[%key:component::knx::config::step::secure_knxkeys::data_description::knxkeys_password%]"
}
},
"knxkeys_tunnel_select": {
"title": "[%key:component::knx::config::step::tcp_tunnel_endpoint::title%]",
"data": {
"tunnel_endpoint_ia": "[%key:component::knx::config::step::tcp_tunnel_endpoint::data::tunnel_endpoint_ia%]"
},
"data_description": {
"tunnel_endpoint_ia": "[%key:component::knx::config::step::tcp_tunnel_endpoint::data_description::tunnel_endpoint_ia%]"
}
},
"secure_tunnel_manual": {
"title": "[%key:component::knx::config::step::secure_tunnel_manual::title%]",
"description": "[%key:component::knx::config::step::secure_tunnel_manual::description%]",
"data": {
"user_id": "[%key:component::knx::config::step::secure_tunnel_manual::data::user_id%]",
"user_password": "[%key:component::knx::config::step::secure_tunnel_manual::data::user_password%]",
"device_authentication": "[%key:component::knx::config::step::secure_tunnel_manual::data::device_authentication%]"
},
"data_description": {
"user_id": "[%key:component::knx::config::step::secure_tunnel_manual::data_description::user_id%]",
"user_password": "[%key:component::knx::config::step::secure_tunnel_manual::data_description::user_password%]",
"device_authentication": "[%key:component::knx::config::step::secure_tunnel_manual::data_description::device_authentication%]"
}
},
"secure_routing_manual": {
"title": "[%key:component::knx::config::step::secure_routing_manual::title%]",
"description": "[%key:component::knx::config::step::secure_tunnel_manual::description%]",
"data": {
"backbone_key": "[%key:component::knx::config::step::secure_routing_manual::data::backbone_key%]",
"sync_latency_tolerance": "[%key:component::knx::config::step::secure_routing_manual::data::sync_latency_tolerance%]"
},
"data_description": {
"backbone_key": "[%key:component::knx::config::step::secure_routing_manual::data_description::backbone_key%]",
"sync_latency_tolerance": "[%key:component::knx::config::step::secure_routing_manual::data_description::sync_latency_tolerance%]"
}
},
"routing": {
"title": "[%key:component::knx::config::step::routing::title%]",
"description": "[%key:component::knx::config::step::routing::description%]",
"data": {
"individual_address": "[%key:component::knx::config::step::routing::data::individual_address%]",
"routing_secure": "[%key:component::knx::config::step::routing::data::routing_secure%]",
"multicast_group": "[%key:component::knx::config::step::routing::data::multicast_group%]",
"multicast_port": "[%key:component::knx::config::step::routing::data::multicast_port%]",
"local_ip": "[%key:component::knx::config::step::manual_tunnel::data::local_ip%]"
},
"data_description": {
"individual_address": "[%key:component::knx::config::step::routing::data_description::individual_address%]",
"routing_secure": "[%key:component::knx::config::step::routing::data_description::routing_secure%]",
"multicast_group": "[%key:component::knx::config::step::routing::data_description::multicast_group%]",
"multicast_port": "[%key:component::knx::config::step::routing::data_description::multicast_port%]",
"local_ip": "[%key:component::knx::config::step::manual_tunnel::data_description::local_ip%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_backbone_key": "[%key:component::knx::config::error::invalid_backbone_key%]",
"invalid_individual_address": "[%key:component::knx::config::error::invalid_individual_address%]",
"invalid_ip_address": "[%key:component::knx::config::error::invalid_ip_address%]",
"keyfile_no_backbone_key": "[%key:component::knx::config::error::keyfile_no_backbone_key%]",
"keyfile_invalid_signature": "[%key:component::knx::config::error::keyfile_invalid_signature%]",
"keyfile_no_tunnel_for_host": "[%key:component::knx::config::error::keyfile_no_tunnel_for_host%]",
"keyfile_not_found": "[%key:component::knx::config::error::keyfile_not_found%]",
"no_router_discovered": "[%key:component::knx::config::error::no_router_discovered%]",
"no_tunnel_discovered": "[%key:component::knx::config::error::no_tunnel_discovered%]",
"unsupported_tunnel_type": "[%key:component::knx::config::error::unsupported_tunnel_type%]"
}
},
"entity": {

View File

@@ -315,6 +315,9 @@ def mock_config_entry() -> MockConfigEntry:
title="KNX",
domain=DOMAIN,
data={
# homeassistant.components.knx.config_flow.DEFAULT_ENTRY_DATA has additional keys
# there are installations out there without these keys so we test with legacy data
# to ensure backwards compatibility (local_ip, telegram_log_size)
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,

View File

@@ -1269,30 +1269,30 @@ async def test_configure_secure_knxkeys_no_tunnel_for_host(hass: HomeAssistant)
assert secure_knxkeys["errors"] == {"base": "keyfile_no_tunnel_for_host"}
async def test_options_flow_connection_type(
async def test_reconfigure_flow_connection_type(
hass: HomeAssistant, knx, mock_config_entry: MockConfigEntry
) -> None:
"""Test options flow changing interface."""
# run one option flow test with a set up integration (knx fixture)
"""Test reconfigure flow changing interface."""
# run one flow test with a set up integration (knx fixture)
# instead of mocking async_setup_entry (knx_setup fixture) to test
# usage of the already running XKNX instance for gateway scanner
gateway = _gateway_descriptor("192.168.0.1", 3675)
await knx.setup_integration()
menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
menu_step = await knx.mock_config_entry.start_reconfigure_flow(hass)
with patch(
"homeassistant.components.knx.config_flow.GatewayScanner"
) as gateway_scanner_mock:
gateway_scanner_mock.return_value = GatewayScannerMock([gateway])
result = await hass.config_entries.options.async_configure(
result = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
{"next_step_id": "connection_type"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "connection_type"
result2 = await hass.config_entries.options.async_configure(
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
@@ -1301,20 +1301,19 @@ async def test_options_flow_connection_type(
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "tunnel"
result3 = await hass.config_entries.options.async_configure(
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
user_input={
CONF_KNX_GATEWAY: str(gateway),
},
)
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert not result3["data"]
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "reconfigure_successful"
assert mock_config_entry.data == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240",
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
CONF_KNX_LOCAL_IP: None,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_RATE_LIMIT: 0,
@@ -1324,14 +1323,13 @@ async def test_options_flow_connection_type(
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None,
CONF_KNX_SECURE_USER_ID: None,
CONF_KNX_SECURE_USER_PASSWORD: None,
CONF_KNX_TELEGRAM_LOG_SIZE: 1000,
}
async def test_options_flow_secure_manual_to_keyfile(
async def test_reconfigure_flow_secure_manual_to_keyfile(
hass: HomeAssistant, knx_setup
) -> None:
"""Test options flow changing secure credential source."""
"""Test reconfigure flow changing secure credential source."""
mock_config_entry = MockConfigEntry(
title="KNX",
domain="knx",
@@ -1359,19 +1357,20 @@ async def test_options_flow_secure_manual_to_keyfile(
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)
knx_setup.reset_mock()
menu_step = await mock_config_entry.start_reconfigure_flow(hass)
with patch(
"homeassistant.components.knx.config_flow.GatewayScanner"
) as gateway_scanner_mock:
gateway_scanner_mock.return_value = GatewayScannerMock([gateway])
result = await hass.config_entries.options.async_configure(
result = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
{"next_step_id": "connection_type"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "connection_type"
result2 = await hass.config_entries.options.async_configure(
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
@@ -1381,14 +1380,14 @@ async def test_options_flow_secure_manual_to_keyfile(
assert result2["step_id"] == "tunnel"
assert not result2["errors"]
result3 = await hass.config_entries.options.async_configure(
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{CONF_KNX_GATEWAY: str(gateway)},
)
assert result3["type"] is FlowResultType.MENU
assert result3["step_id"] == "secure_key_source_menu_tunnel"
result4 = await hass.config_entries.options.async_configure(
result4 = await hass.config_entries.flow.async_configure(
result3["flow_id"],
{"next_step_id": "secure_knxkeys"},
)
@@ -1397,7 +1396,7 @@ async def test_options_flow_secure_manual_to_keyfile(
assert not result4["errors"]
with patch_file_upload():
secure_knxkeys = await hass.config_entries.options.async_configure(
secure_knxkeys = await hass.config_entries.flow.async_configure(
result4["flow_id"],
{
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
@@ -1407,12 +1406,13 @@ async def test_options_flow_secure_manual_to_keyfile(
assert result["type"] is FlowResultType.FORM
assert secure_knxkeys["step_id"] == "knxkeys_tunnel_select"
assert not result["errors"]
secure_knxkeys = await hass.config_entries.options.async_configure(
secure_knxkeys = await hass.config_entries.flow.async_configure(
secure_knxkeys["flow_id"],
{CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1"},
)
assert secure_knxkeys["type"] is FlowResultType.CREATE_ENTRY
assert secure_knxkeys["type"] is FlowResultType.ABORT
assert secure_knxkeys["reason"] == "reconfigure_successful"
assert mock_config_entry.data == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
@@ -1433,8 +1433,8 @@ 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."""
async def test_reconfigure_flow_routing(hass: HomeAssistant, knx_setup) -> None:
"""Test reconfigure flow changing routing settings."""
mock_config_entry = MockConfigEntry(
title="KNX",
domain="knx",
@@ -1446,20 +1446,21 @@ async def test_options_flow_routing(hass: HomeAssistant, knx_setup) -> None:
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)
knx_setup.reset_mock()
menu_step = await mock_config_entry.start_reconfigure_flow(hass)
with patch(
"homeassistant.components.knx.config_flow.GatewayScanner"
) as gateway_scanner_mock:
gateway_scanner_mock.return_value = GatewayScannerMock([gateway])
result = await hass.config_entries.options.async_configure(
result = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
{"next_step_id": "connection_type"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "connection_type"
result2 = await hass.config_entries.options.async_configure(
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
@@ -1469,13 +1470,14 @@ async def test_options_flow_routing(hass: HomeAssistant, knx_setup) -> None:
assert result2["step_id"] == "routing"
assert result2["errors"] == {}
result3 = await hass.config_entries.options.async_configure(
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_KNX_INDIVIDUAL_ADDRESS: "2.0.4",
},
)
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "reconfigure_successful"
assert mock_config_entry.data == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
@@ -1491,43 +1493,8 @@ async def test_options_flow_routing(hass: HomeAssistant, knx_setup) -> None:
knx_setup.assert_called_once()
async def test_options_communication_settings(
hass: HomeAssistant, knx_setup, mock_config_entry: MockConfigEntry
) -> None:
"""Test options flow changing communication settings."""
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": "communication_settings"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "communication_settings"
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_KNX_STATE_UPDATER: False,
CONF_KNX_RATE_LIMIT: 40,
CONF_KNX_TELEGRAM_LOG_SIZE: 3000,
},
)
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert not result2.get("data")
assert mock_config_entry.data == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_STATE_UPDATER: False,
CONF_KNX_RATE_LIMIT: 40,
CONF_KNX_TELEGRAM_LOG_SIZE: 3000,
}
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."""
async def test_reconfigure_update_keyfile(hass: HomeAssistant, knx_setup) -> None:
"""Test reconfigure flow updating keyfile when tunnel endpoint is already configured."""
start_data = {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
@@ -1549,9 +1516,10 @@ async def test_options_update_keyfile(hass: HomeAssistant, knx_setup) -> None:
)
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)
knx_setup.reset_mock()
menu_step = await mock_config_entry.start_reconfigure_flow(hass)
result = await hass.config_entries.options.async_configure(
result = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
{"next_step_id": "secure_knxkeys"},
)
@@ -1559,15 +1527,15 @@ async def test_options_update_keyfile(hass: HomeAssistant, knx_setup) -> None:
assert result["step_id"] == "secure_knxkeys"
with patch_file_upload():
result2 = await hass.config_entries.options.async_configure(
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
CONF_KNX_KNXKEY_PASSWORD: "password",
},
)
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert not result2.get("data")
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "reconfigure_successful"
assert mock_config_entry.data == {
**start_data,
CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys",
@@ -1578,8 +1546,8 @@ async def test_options_update_keyfile(hass: HomeAssistant, knx_setup) -> 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."""
async def test_reconfigure_keyfile_upload(hass: HomeAssistant, knx_setup) -> None:
"""Test reconfigure flow uploading a keyfile for the first time."""
start_data = {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP,
@@ -1596,9 +1564,10 @@ async def test_options_keyfile_upload(hass: HomeAssistant, knx_setup) -> None:
)
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)
knx_setup.reset_mock()
menu_step = await mock_config_entry.start_reconfigure_flow(hass)
result = await hass.config_entries.options.async_configure(
result = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
{"next_step_id": "secure_knxkeys"},
)
@@ -1606,7 +1575,7 @@ async def test_options_keyfile_upload(hass: HomeAssistant, knx_setup) -> None:
assert result["step_id"] == "secure_knxkeys"
with patch_file_upload():
result2 = await hass.config_entries.options.async_configure(
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
@@ -1617,14 +1586,14 @@ async def test_options_keyfile_upload(hass: HomeAssistant, knx_setup) -> None:
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "knxkeys_tunnel_select"
result3 = await hass.config_entries.options.async_configure(
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1",
},
)
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert not result3.get("data")
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "reconfigure_successful"
assert mock_config_entry.data == {
**start_data,
CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys",
@@ -1637,3 +1606,35 @@ async def test_options_keyfile_upload(hass: HomeAssistant, knx_setup) -> None:
CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None,
}
knx_setup.assert_called_once()
async def test_options_communication_settings(
hass: HomeAssistant, knx_setup, mock_config_entry: MockConfigEntry
) -> None:
"""Test options flow changing communication settings."""
initial_data = dict(mock_config_entry.data)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
result = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "communication_settings"
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_KNX_STATE_UPDATER: False,
CONF_KNX_RATE_LIMIT: 40,
CONF_KNX_TELEGRAM_LOG_SIZE: 3000,
},
)
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert not result2.get("data")
assert initial_data != dict(mock_config_entry.data)
assert mock_config_entry.data == {
**initial_data,
CONF_KNX_STATE_UPDATER: False,
CONF_KNX_RATE_LIMIT: 40,
CONF_KNX_TELEGRAM_LOG_SIZE: 3000,
}
knx_setup.assert_called_once()