home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-07-30 02:38:10 +02:00
Code Issues Packages Projects Releases Wiki Activity

Remove connection options from OptionsFlow

Browse Source
This commit is contained in:
farmio
2025-05-17 12:08:40 +02:00
parent fd0023ec9f
commit 6a998bb3ed
3 changed files with 67 additions and 576 deletions
Download Patch File Download Diff File Expand all files Collapse all files

143
homeassistant/components/knx/config_flow.py
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
@ -22,7 +21,6 @@ from xknx.secure.keyring import Keyring, XMLInterface
from homeassistant.config_entries import (
SOURCE_RECONFIGURE,
ConfigEntry,
ConfigEntryBaseFlow,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
@ -104,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
@ -122,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) or self.source == SOURCE_RECONFIGURE) 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."""
@ -151,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:
@ -442,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
@ -842,73 +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 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",
],
)
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={})
@ -916,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(

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

@ -5,7 +5,7 @@
"title": "KNX connection settings",
"menu_options": {
"connection_type": "Reconfigure KNX connection",
"secure_knxkeys": "Import a `.knxkeys` keyring file"
"secure_knxkeys": "Import KNX keyring file"
}
},
"connection_type": {
@ -72,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"
@ -169,16 +169,8 @@
},
"options": {
"step": {
"init": {
"title": "KNX Settings",
"menu_options": {
"connection_type": "[%key:component::knx::config::step::reconfigure::menu_options::connection_type%]",
"communication_settings": "Communication settings",
"secure_knxkeys": "[%key:component::knx::config::step::reconfigure::menu_options::secure_knxkeys%]"
}
},
"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",
@ -189,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": {

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

@ -1608,240 +1608,15 @@ async def test_reconfigure_keyfile_upload(hass: HomeAssistant, knx_setup) -> Non
knx_setup.assert_called_once()
async def test_options_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)
# 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)
with patch(
"homeassistant.components.knx.config_flow.GatewayScanner"
) as gateway_scanner_mock:
gateway_scanner_mock.return_value = GatewayScannerMock([gateway])
result = await hass.config_entries.options.async_configure(
menu_step["flow_id"],
{"next_step_id": "connection_type"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "connection_type"
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "tunnel"
result3 = await hass.config_entries.options.async_configure(
result2["flow_id"],
user_input={
CONF_KNX_GATEWAY: str(gateway),
},
)
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert not result3["data"]
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,
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,
CONF_KNX_TELEGRAM_LOG_SIZE: 1000,
}
async def test_options_flow_secure_manual_to_keyfile(
hass: HomeAssistant, knx_setup
) -> None:
"""Test options flow changing secure credential source."""
mock_config_entry = MockConfigEntry(
title="KNX",
domain="knx",
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_FILENAME: "knx/testcase.knxkeys",
CONF_KNX_KNXKEY_PASSWORD: "invalid_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,
},
)
gateway = _gateway_descriptor(
"192.168.0.1",
3675,
supports_tunnelling_tcp=True,
requires_secure=True,
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
with patch(
"homeassistant.components.knx.config_flow.GatewayScanner"
) as gateway_scanner_mock:
gateway_scanner_mock.return_value = GatewayScannerMock([gateway])
result = await hass.config_entries.options.async_configure(
menu_step["flow_id"],
{"next_step_id": "connection_type"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "connection_type"
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "tunnel"
assert not result2["errors"]
result3 = await hass.config_entries.options.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(
result3["flow_id"],
{"next_step_id": "secure_knxkeys"},
)
assert result4["type"] is FlowResultType.FORM
assert result4["step_id"] == "secure_knxkeys"
assert not result4["errors"]
with patch_file_upload():
secure_knxkeys = await hass.config_entries.options.async_configure(
result4["flow_id"],
{
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
CONF_KNX_KNXKEY_PASSWORD: "test",
},
)
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["flow_id"],
{CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1"},
)
assert secure_knxkeys["type"] is FlowResultType.CREATE_ENTRY
assert mock_config_entry.data == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys",
CONF_KNX_KNXKEY_PASSWORD: "test",
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None,
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",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240",
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: None,
}
knx_setup.assert_called_once()
async def test_options_flow_routing(hass: HomeAssistant, knx_setup) -> None:
"""Test options flow changing routing settings."""
mock_config_entry = MockConfigEntry(
title="KNX",
domain="knx",
data={
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
},
)
gateway = _gateway_descriptor("192.168.0.1", 3676)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
with patch(
"homeassistant.components.knx.config_flow.GatewayScanner"
) as gateway_scanner_mock:
gateway_scanner_mock.return_value = GatewayScannerMock([gateway])
result = await hass.config_entries.options.async_configure(
menu_step["flow_id"],
{"next_step_id": "connection_type"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "connection_type"
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "routing"
assert result2["errors"] == {}
result3 = await hass.config_entries.options.async_configure(
result2["flow_id"],
{
CONF_KNX_INDIVIDUAL_ADDRESS: "2.0.4",
},
)
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert mock_config_entry.data == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_LOCAL_IP: None,
CONF_KNX_INDIVIDUAL_ADDRESS: "2.0.4",
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None,
CONF_KNX_SECURE_USER_ID: None,
CONF_KNX_SECURE_USER_PASSWORD: None,
CONF_KNX_TUNNEL_ENDPOINT_IA: None,
}
knx_setup.assert_called_once()
async def test_options_communication_settings(
hass: HomeAssistant, knx_setup, mock_config_entry: MockConfigEntry
) -> None:
"""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)
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"},
)
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"
@ -1855,124 +1630,11 @@ async def test_options_communication_settings(
)
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 == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
**initial_data,
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."""
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"] is 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"] is 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"] is 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"] is 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",
},
)
assert result3["type"] is 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()