jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-28 15:47:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add reconfigure flow for KNX (#145067)

Browse Source 
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Matthias Alphart 2025-07-14 21:49:52 +02:00 committed by GitHub
parent 66641356cc
commit f65fa38429
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 290 additions and 402 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

138
homeassistant/components/knx/config_flow.py
View File

@ -2,7 +2,6 @@
from __future__ import annotations from __future__ import annotations
from abc import ABC, abstractmethod
from collections.abc import AsyncGenerator from collections.abc import AsyncGenerator
from typing import Any, Final, Literal 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 xknx.secure.keyring import Keyring, XMLInterface
from homeassistant.config_entries import ( from homeassistant.config_entries import (
SOURCE_RECONFIGURE,
ConfigEntry, ConfigEntry,
ConfigEntryBaseFlow,
ConfigFlow, ConfigFlow,
ConfigFlowResult, ConfigFlowResult,
OptionsFlow, OptionsFlow,
@ -103,12 +102,14 @@ _PORT_SELECTOR = vol.All(
) )
class KNXCommonFlow(ABC, ConfigEntryBaseFlow): class KNXConfigFlow(ConfigFlow, domain=DOMAIN):
"""Base class for KNX flows.""" """Handle a KNX config flow."""
def __init__(self, initial_data: KNXConfigEntryData) -> None: VERSION = 1
"""Initialize KNXCommonFlow."""
self.initial_data = initial_data def __init__(self) -> None:
"""Initialize KNX config flow."""
self.initial_data = DEFAULT_ENTRY_DATA
self.new_entry_data = KNXConfigEntryData() self.new_entry_data = KNXConfigEntryData()
self.new_title: str | None = None self.new_title: str | None = None
@ -121,19 +122,21 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
self._gatewayscanner: GatewayScanner | None = None self._gatewayscanner: GatewayScanner | None = None
self._async_scan_gen: AsyncGenerator[GatewayDescriptor] | 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 @property
def _xknx(self) -> XKNX: def _xknx(self) -> XKNX:
"""Return XKNX instance.""" """Return XKNX instance."""
if isinstance(self, OptionsFlow) and ( if (self.source == SOURCE_RECONFIGURE) and (
knx_module := self.hass.data.get(KNX_MODULE_KEY) knx_module := self.hass.data.get(KNX_MODULE_KEY)
): ):
return knx_module.xknx return knx_module.xknx
return XKNX() return XKNX()
@abstractmethod
def finish_flow(self) -> ConfigFlowResult:
"""Finish the flow."""
@property @property
def connection_type(self) -> str: def connection_type(self) -> str:
"""Return the configured connection type.""" """Return the configured connection type."""
@ -150,6 +153,61 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
self.initial_data.get(CONF_KNX_TUNNEL_ENDPOINT_IA), 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:
entry = self._get_reconfigure_entry()
_tunnel_endpoint_str = self.initial_data.get(
CONF_KNX_TUNNEL_ENDPOINT_IA, "Tunneling"
)
if self.new_title and not entry.title.startswith(
# Overwrite standard titles, but not user defined ones
(
f"KNX {self.initial_data[CONF_KNX_CONNECTION_TYPE]}",
CONF_KNX_AUTOMATIC.capitalize(),
"Tunneling @ ",
f"{_tunnel_endpoint_str} @",
"Tunneling UDP @ ",
"Tunneling TCP @ ",
"Secure Tunneling",
"Routing as ",
"Secure Routing as ",
)
):
self.new_title = None
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[str, Any] | 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( async def async_step_connection_type(
self, user_input: dict | None = None self, user_input: dict | None = None
) -> ConfigFlowResult: ) -> ConfigFlowResult:
@ -441,7 +499,7 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
) )
ip_address: str | None ip_address: str | None
if ( # initial attempt on ConfigFlow or coming from automatic / routing 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 not user_input
and self._selected_tunnel is not None and self._selected_tunnel is not None
): # default to first found tunnel ): # default to first found tunnel
@ -841,52 +899,20 @@ class KNXCommonFlow(ABC, ConfigEntryBaseFlow):
) )
class KNXConfigFlow(KNXCommonFlow, ConfigFlow, domain=DOMAIN): class KNXOptionsFlow(OptionsFlow):
"""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):
"""Handle KNX options.""" """Handle KNX options."""
general_settings: dict
def __init__(self, config_entry: ConfigEntry) -> None: def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize KNX options flow.""" """Initialize KNX options flow."""
super().__init__(initial_data=config_entry.data) # type: ignore[arg-type] self.initial_data = dict(config_entry.data)
@callback @callback
def finish_flow(self) -> ConfigFlowResult: def finish_flow(self, new_entry_data: KNXConfigEntryData) -> ConfigFlowResult:
"""Update the ConfigEntry and finish the flow.""" """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.hass.config_entries.async_update_entry(
self.config_entry, self.config_entry,
data=new_data, data=new_data,
title=self.new_title or UNDEFINED,
) )
return self.async_create_entry(title="", data={}) return self.async_create_entry(title="", data={})
@ -894,26 +920,20 @@ class KNXOptionsFlow(KNXCommonFlow, OptionsFlow):
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult: ) -> ConfigFlowResult:
"""Manage KNX options.""" """Manage KNX options."""
return self.async_show_menu( return await self.async_step_communication_settings()
step_id="init",
menu_options=[
"connection_type",
"communication_settings",
"secure_knxkeys",
],
)
async def async_step_communication_settings( async def async_step_communication_settings(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult: ) -> ConfigFlowResult:
"""Manage KNX communication settings.""" """Manage KNX communication settings."""
if user_input is not None: if user_input is not None:
self.new_entry_data = KNXConfigEntryData( return self.finish_flow(
KNXConfigEntryData(
state_updater=user_input[CONF_KNX_STATE_UPDATER], state_updater=user_input[CONF_KNX_STATE_UPDATER],
rate_limit=user_input[CONF_KNX_RATE_LIMIT], rate_limit=user_input[CONF_KNX_RATE_LIMIT],
telegram_log_size=user_input[CONF_KNX_TELEGRAM_LOG_SIZE], telegram_log_size=user_input[CONF_KNX_TELEGRAM_LOG_SIZE],
) )
return self.finish_flow() )
data_schema = { data_schema = {
vol.Required( vol.Required(

2
homeassistant/components/knx/quality_scale.yaml
View File

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

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

@ -1,6 +1,13 @@
{ {
"config": { "config": {
"step": { "step": {
"reconfigure": {
"title": "KNX connection settings",
"menu_options": {
"connection_type": "Reconfigure KNX connection",
"secure_knxkeys": "Import KNX keyring file"
}
},
"connection_type": { "connection_type": {
"title": "KNX connection", "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.", "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": { "secure_knxkeys": {
"title": "Import KNX Keyring", "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": { "data": {
"knxkeys_file": "Keyring file", "knxkeys_file": "Keyring file",
"knxkeys_password": "Keyring password" "knxkeys_password": "Keyring password"
@ -129,6 +136,9 @@
} }
} }
}, },
"abort": {
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
},
"error": { "error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_backbone_key": "Invalid backbone key. 32 hexadecimal digits expected.", "invalid_backbone_key": "Invalid backbone key. 32 hexadecimal digits expected.",
@ -159,16 +169,8 @@
}, },
"options": { "options": {
"step": { "step": {
"init": {
"title": "KNX Settings",
"menu_options": {
"connection_type": "Configure KNX interface",
"communication_settings": "Communication settings",
"secure_knxkeys": "Import a `.knxkeys` file"
}
},
"communication_settings": { "communication_settings": {
"title": "[%key:component::knx::options::step::init::menu_options::communication_settings%]", "title": "Communication settings",
"data": { "data": {
"state_updater": "State updater", "state_updater": "State updater",
"rate_limit": "Rate limit", "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`", "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}" "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": { "entity": {

3
tests/components/knx/conftest.py
View File

@ -309,6 +309,9 @@ def mock_config_entry() -> MockConfigEntry:
title="KNX", title="KNX",
domain=DOMAIN, domain=DOMAIN,
data={ 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_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT, CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER, CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,

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

@ -48,7 +48,7 @@ from homeassistant.components.knx.const import (
) )
from homeassistant.const import CONF_HOST, CONF_PORT from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult, FlowResultType from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry, get_fixture_path from tests.common import MockConfigEntry, get_fixture_path
@ -174,27 +174,27 @@ async def test_routing_setup(
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert not result["errors"] assert not result["errors"]
result2 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
}, },
) )
assert result2["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result2["step_id"] == "routing" assert result["step_id"] == "routing"
assert result2["errors"] == {"base": "no_router_discovered"} assert result["errors"] == {"base": "no_router_discovered"}
result3 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
{ {
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675, CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110", CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110",
}, },
) )
assert result3["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.CREATE_ENTRY
assert result3["title"] == "Routing as 1.1.110" assert result["title"] == "Routing as 1.1.110"
assert result3["data"] == { assert result["data"] == {
**DEFAULT_ENTRY_DATA, **DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
@ -227,19 +227,19 @@ async def test_routing_setup_advanced(
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert not result["errors"] assert not result["errors"]
result2 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
}, },
) )
assert result2["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result2["step_id"] == "routing" assert result["step_id"] == "routing"
assert result2["errors"] == {"base": "no_router_discovered"} assert result["errors"] == {"base": "no_router_discovered"}
# invalid user input # invalid user input
result_invalid_input = await hass.config_entries.flow.async_configure( result_invalid_input = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
{ {
CONF_KNX_MCAST_GRP: "10.1.2.3", # no valid multicast group CONF_KNX_MCAST_GRP: "10.1.2.3", # no valid multicast group
CONF_KNX_MCAST_PORT: 3675, CONF_KNX_MCAST_PORT: 3675,
@ -257,8 +257,8 @@ async def test_routing_setup_advanced(
} }
# valid user input # valid user input
result3 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
{ {
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675, CONF_KNX_MCAST_PORT: 3675,
@ -266,9 +266,9 @@ async def test_routing_setup_advanced(
CONF_KNX_LOCAL_IP: "192.168.1.112", CONF_KNX_LOCAL_IP: "192.168.1.112",
}, },
) )
assert result3["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.CREATE_ENTRY
assert result3["title"] == "Routing as 1.1.110" assert result["title"] == "Routing as 1.1.110"
assert result3["data"] == { assert result["data"] == {
**DEFAULT_ENTRY_DATA, **DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
@ -297,18 +297,18 @@ async def test_routing_secure_manual_setup(
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert not result["errors"] assert not result["errors"]
result2 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
}, },
) )
assert result2["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result2["step_id"] == "routing" assert result["step_id"] == "routing"
assert result2["errors"] == {"base": "no_router_discovered"} assert result["errors"] == {"base": "no_router_discovered"}
result3 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
{ {
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3671, CONF_KNX_MCAST_PORT: 3671,
@ -316,19 +316,19 @@ async def test_routing_secure_manual_setup(
CONF_KNX_ROUTING_SECURE: True, CONF_KNX_ROUTING_SECURE: True,
}, },
) )
assert result3["type"] is FlowResultType.MENU assert result["type"] is FlowResultType.MENU
assert result3["step_id"] == "secure_key_source_menu_routing" assert result["step_id"] == "secure_key_source_menu_routing"
result4 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result3["flow_id"], result["flow_id"],
{"next_step_id": "secure_routing_manual"}, {"next_step_id": "secure_routing_manual"},
) )
assert result4["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result4["step_id"] == "secure_routing_manual" assert result["step_id"] == "secure_routing_manual"
assert not result4["errors"] assert not result["errors"]
result_invalid_key1 = await hass.config_entries.flow.async_configure( result_invalid_key1 = await hass.config_entries.flow.async_configure(
result4["flow_id"], result["flow_id"],
{ {
CONF_KNX_ROUTING_BACKBONE_KEY: "xxaacc44bbaacc44bbaacc44bbaaccyy", # invalid hex string CONF_KNX_ROUTING_BACKBONE_KEY: "xxaacc44bbaacc44bbaacc44bbaaccyy", # invalid hex string
CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000,
@ -339,7 +339,7 @@ async def test_routing_secure_manual_setup(
assert result_invalid_key1["errors"] == {"backbone_key": "invalid_backbone_key"} assert result_invalid_key1["errors"] == {"backbone_key": "invalid_backbone_key"}
result_invalid_key2 = await hass.config_entries.flow.async_configure( result_invalid_key2 = await hass.config_entries.flow.async_configure(
result4["flow_id"], result["flow_id"],
{ {
CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44", # invalid length CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44", # invalid length
CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000,
@ -386,18 +386,18 @@ async def test_routing_secure_keyfile(
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert not result["errors"] assert not result["errors"]
result2 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
}, },
) )
assert result2["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result2["step_id"] == "routing" assert result["step_id"] == "routing"
assert result2["errors"] == {"base": "no_router_discovered"} assert result["errors"] == {"base": "no_router_discovered"}
result3 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
{ {
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3671, CONF_KNX_MCAST_PORT: 3671,
@ -405,20 +405,20 @@ async def test_routing_secure_keyfile(
CONF_KNX_ROUTING_SECURE: True, CONF_KNX_ROUTING_SECURE: True,
}, },
) )
assert result3["type"] is FlowResultType.MENU assert result["type"] is FlowResultType.MENU
assert result3["step_id"] == "secure_key_source_menu_routing" assert result["step_id"] == "secure_key_source_menu_routing"
result4 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result3["flow_id"], result["flow_id"],
{"next_step_id": "secure_knxkeys"}, {"next_step_id": "secure_knxkeys"},
) )
assert result4["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result4["step_id"] == "secure_knxkeys" assert result["step_id"] == "secure_knxkeys"
assert not result4["errors"] assert not result["errors"]
with patch_file_upload(): with patch_file_upload():
routing_secure_knxkeys = await hass.config_entries.flow.async_configure( routing_secure_knxkeys = await hass.config_entries.flow.async_configure(
result4["flow_id"], result["flow_id"],
{ {
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
CONF_KNX_KNXKEY_PASSWORD: "password", CONF_KNX_KNXKEY_PASSWORD: "password",
@ -532,15 +532,15 @@ async def test_tunneling_setup_manual(
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert not result["errors"] assert not result["errors"]
result2 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
}, },
) )
assert result2["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result2["step_id"] == "manual_tunnel" assert result["step_id"] == "manual_tunnel"
assert result2["errors"] == {"base": "no_tunnel_discovered"} assert result["errors"] == {"base": "no_tunnel_discovered"}
with patch( with patch(
"homeassistant.components.knx.config_flow.request_description", "homeassistant.components.knx.config_flow.request_description",
@ -552,13 +552,13 @@ async def test_tunneling_setup_manual(
), ),
), ),
): ):
result3 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
user_input, user_input,
) )
assert result3["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.CREATE_ENTRY
assert result3["title"] == title assert result["title"] == title
assert result3["data"] == config_entry_data assert result["data"] == config_entry_data
knx_setup.assert_called_once() knx_setup.assert_called_once()
@ -724,19 +724,19 @@ async def test_tunneling_setup_for_local_ip(
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert not result["errors"] assert not result["errors"]
result2 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
}, },
) )
assert result2["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result2["step_id"] == "manual_tunnel" assert result["step_id"] == "manual_tunnel"
assert result2["errors"] == {"base": "no_tunnel_discovered"} assert result["errors"] == {"base": "no_tunnel_discovered"}
# invalid host ip address # invalid host ip address
result_invalid_host = await hass.config_entries.flow.async_configure( result_invalid_host = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
{ {
CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING, CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: DEFAULT_MCAST_GRP, # multicast addresses are invalid CONF_HOST: DEFAULT_MCAST_GRP, # multicast addresses are invalid
@ -752,7 +752,7 @@ async def test_tunneling_setup_for_local_ip(
} }
# invalid local ip address # invalid local ip address
result_invalid_local = await hass.config_entries.flow.async_configure( result_invalid_local = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
{ {
CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING, CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.0.2", CONF_HOST: "192.168.0.2",
@ -768,8 +768,8 @@ async def test_tunneling_setup_for_local_ip(
} }
# valid user input # valid user input
result3 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
{ {
CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING, CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.0.2", CONF_HOST: "192.168.0.2",
@ -777,9 +777,9 @@ async def test_tunneling_setup_for_local_ip(
CONF_KNX_LOCAL_IP: "192.168.1.112", CONF_KNX_LOCAL_IP: "192.168.1.112",
}, },
) )
assert result3["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.CREATE_ENTRY
assert result3["title"] == "Tunneling UDP @ 192.168.0.2" assert result["title"] == "Tunneling UDP @ 192.168.0.2"
assert result3["data"] == { assert result["data"] == {
**DEFAULT_ENTRY_DATA, **DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.0.2", CONF_HOST: "192.168.0.2",
@ -1008,15 +1008,15 @@ async def test_form_with_automatic_connection_handling(
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert not result["errors"] assert not result["errors"]
result2 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
}, },
) )
assert result2["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == CONF_KNX_AUTOMATIC.capitalize() assert result["title"] == CONF_KNX_AUTOMATIC.capitalize()
assert result2["data"] == { assert result["data"] == {
# don't use **DEFAULT_ENTRY_DATA here to check for correct usage of defaults # don't use **DEFAULT_ENTRY_DATA here to check for correct usage of defaults
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240",
@ -1032,7 +1032,9 @@ async def test_form_with_automatic_connection_handling(
knx_setup.assert_called_once() knx_setup.assert_called_once()
async def _get_menu_step_secure_tunnel(hass: HomeAssistant) -> FlowResult: async def _get_menu_step_secure_tunnel(
hass: HomeAssistant,
) -> config_entries.ConfigFlowResult:
"""Return flow in secure_tunnel menu step.""" """Return flow in secure_tunnel menu step."""
gateway = _gateway_descriptor( gateway = _gateway_descriptor(
"192.168.0.1", "192.168.0.1",
@ -1050,23 +1052,23 @@ async def _get_menu_step_secure_tunnel(hass: HomeAssistant) -> FlowResult:
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert not result["errors"] assert not result["errors"]
result2 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
}, },
) )
assert result2["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result2["step_id"] == "tunnel" assert result["step_id"] == "tunnel"
assert not result2["errors"] assert not result["errors"]
result3 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
{CONF_KNX_GATEWAY: str(gateway)}, {CONF_KNX_GATEWAY: str(gateway)},
) )
assert result3["type"] is FlowResultType.MENU assert result["type"] is FlowResultType.MENU
assert result3["step_id"] == "secure_key_source_menu_tunnel" assert result["step_id"] == "secure_key_source_menu_tunnel"
return result3 return result
@patch( @patch(
@ -1099,24 +1101,24 @@ async def test_get_secure_menu_step_manual_tunnelling(
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert not result["errors"] assert not result["errors"]
result2 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
}, },
) )
assert result2["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result2["step_id"] == "tunnel" assert result["step_id"] == "tunnel"
assert not result2["errors"] assert not result["errors"]
manual_tunnel_flow = await hass.config_entries.flow.async_configure( manual_tunnel_flow = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
{ {
CONF_KNX_GATEWAY: OPTION_MANUAL_TUNNEL, CONF_KNX_GATEWAY: OPTION_MANUAL_TUNNEL,
}, },
) )
result3 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
manual_tunnel_flow["flow_id"], manual_tunnel_flow["flow_id"],
{ {
CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
@ -1124,8 +1126,8 @@ async def test_get_secure_menu_step_manual_tunnelling(
CONF_PORT: 3675, CONF_PORT: 3675,
}, },
) )
assert result3["type"] is FlowResultType.MENU assert result["type"] is FlowResultType.MENU
assert result3["step_id"] == "secure_key_source_menu_tunnel" assert result["step_id"] == "secure_key_source_menu_tunnel"
async def test_configure_secure_tunnel_manual(hass: HomeAssistant, knx_setup) -> None: async def test_configure_secure_tunnel_manual(hass: HomeAssistant, knx_setup) -> None:
@ -1269,52 +1271,51 @@ async def test_configure_secure_knxkeys_no_tunnel_for_host(hass: HomeAssistant)
assert secure_knxkeys["errors"] == {"base": "keyfile_no_tunnel_for_host"} 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 hass: HomeAssistant, knx, mock_config_entry: MockConfigEntry
) -> None: ) -> None:
"""Test options flow changing interface.""" """Test reconfigure flow changing interface."""
# run one option flow test with a set up integration (knx fixture) # run one flow test with a set up integration (knx fixture)
# instead of mocking async_setup_entry (knx_setup fixture) to test # instead of mocking async_setup_entry (knx_setup fixture) to test
# usage of the already running XKNX instance for gateway scanner # usage of the already running XKNX instance for gateway scanner
gateway = _gateway_descriptor("192.168.0.1", 3675) gateway = _gateway_descriptor("192.168.0.1", 3675)
await knx.setup_integration() 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( with patch(
"homeassistant.components.knx.config_flow.GatewayScanner" "homeassistant.components.knx.config_flow.GatewayScanner"
) as gateway_scanner_mock: ) as gateway_scanner_mock:
gateway_scanner_mock.return_value = GatewayScannerMock([gateway]) 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"], menu_step["flow_id"],
{"next_step_id": "connection_type"}, {"next_step_id": "connection_type"},
) )
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "connection_type" assert result["step_id"] == "connection_type"
result2 = await hass.config_entries.options.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
user_input={ user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
}, },
) )
assert result2["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result2["step_id"] == "tunnel" assert result["step_id"] == "tunnel"
result3 = await hass.config_entries.options.async_configure( result = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
user_input={ user_input={
CONF_KNX_GATEWAY: str(gateway), CONF_KNX_GATEWAY: str(gateway),
}, },
) )
assert result3["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.ABORT
assert not result3["data"] assert result["reason"] == "reconfigure_successful"
assert mock_config_entry.data == { assert mock_config_entry.data == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240",
CONF_HOST: "192.168.0.1", CONF_HOST: "192.168.0.1",
CONF_PORT: 3675, CONF_PORT: 3675,
CONF_KNX_LOCAL_IP: None,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT, CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_RATE_LIMIT: 0, CONF_KNX_RATE_LIMIT: 0,
@ -1324,14 +1325,13 @@ async def test_options_flow_connection_type(
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None,
CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_ID: None,
CONF_KNX_SECURE_USER_PASSWORD: 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 hass: HomeAssistant, knx_setup
) -> None: ) -> None:
"""Test options flow changing secure credential source.""" """Test reconfigure flow changing secure credential source."""
mock_config_entry = MockConfigEntry( mock_config_entry = MockConfigEntry(
title="KNX", title="KNX",
domain="knx", domain="knx",
@ -1359,46 +1359,47 @@ async def test_options_flow_secure_manual_to_keyfile(
mock_config_entry.add_to_hass(hass) mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id) 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( with patch(
"homeassistant.components.knx.config_flow.GatewayScanner" "homeassistant.components.knx.config_flow.GatewayScanner"
) as gateway_scanner_mock: ) as gateway_scanner_mock:
gateway_scanner_mock.return_value = GatewayScannerMock([gateway]) 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"], menu_step["flow_id"],
{"next_step_id": "connection_type"}, {"next_step_id": "connection_type"},
) )
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "connection_type" assert result["step_id"] == "connection_type"
result2 = await hass.config_entries.options.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
user_input={ user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
}, },
) )
assert result2["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result2["step_id"] == "tunnel" assert result["step_id"] == "tunnel"
assert not result2["errors"] assert not result["errors"]
result3 = await hass.config_entries.options.async_configure( result = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
{CONF_KNX_GATEWAY: str(gateway)}, {CONF_KNX_GATEWAY: str(gateway)},
) )
assert result3["type"] is FlowResultType.MENU assert result["type"] is FlowResultType.MENU
assert result3["step_id"] == "secure_key_source_menu_tunnel" assert result["step_id"] == "secure_key_source_menu_tunnel"
result4 = await hass.config_entries.options.async_configure( result = await hass.config_entries.flow.async_configure(
result3["flow_id"], result["flow_id"],
{"next_step_id": "secure_knxkeys"}, {"next_step_id": "secure_knxkeys"},
) )
assert result4["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result4["step_id"] == "secure_knxkeys" assert result["step_id"] == "secure_knxkeys"
assert not result4["errors"] assert not result["errors"]
with patch_file_upload(): 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"], result["flow_id"],
{ {
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
CONF_KNX_KNXKEY_PASSWORD: "test", CONF_KNX_KNXKEY_PASSWORD: "test",
@ -1407,12 +1408,13 @@ async def test_options_flow_secure_manual_to_keyfile(
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert secure_knxkeys["step_id"] == "knxkeys_tunnel_select" assert secure_knxkeys["step_id"] == "knxkeys_tunnel_select"
assert not result["errors"] 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"], secure_knxkeys["flow_id"],
{CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1"}, {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 == { assert mock_config_entry.data == {
**DEFAULT_ENTRY_DATA, **DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
@ -1433,8 +1435,8 @@ async def test_options_flow_secure_manual_to_keyfile(
knx_setup.assert_called_once() knx_setup.assert_called_once()
async def test_options_flow_routing(hass: HomeAssistant, knx_setup) -> None: async def test_reconfigure_flow_routing(hass: HomeAssistant, knx_setup) -> None:
"""Test options flow changing routing settings.""" """Test reconfigure flow changing routing settings."""
mock_config_entry = MockConfigEntry( mock_config_entry = MockConfigEntry(
title="KNX", title="KNX",
domain="knx", domain="knx",
@ -1446,36 +1448,38 @@ async def test_options_flow_routing(hass: HomeAssistant, knx_setup) -> None:
gateway = _gateway_descriptor("192.168.0.1", 3676) gateway = _gateway_descriptor("192.168.0.1", 3676)
mock_config_entry.add_to_hass(hass) mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id) 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( with patch(
"homeassistant.components.knx.config_flow.GatewayScanner" "homeassistant.components.knx.config_flow.GatewayScanner"
) as gateway_scanner_mock: ) as gateway_scanner_mock:
gateway_scanner_mock.return_value = GatewayScannerMock([gateway]) 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"], menu_step["flow_id"],
{"next_step_id": "connection_type"}, {"next_step_id": "connection_type"},
) )
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "connection_type" assert result["step_id"] == "connection_type"
result2 = await hass.config_entries.options.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
user_input={ user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
}, },
) )
assert result2["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result2["step_id"] == "routing" assert result["step_id"] == "routing"
assert result2["errors"] == {} assert result["errors"] == {}
result3 = await hass.config_entries.options.async_configure( result = await hass.config_entries.flow.async_configure(
result2["flow_id"], result["flow_id"],
{ {
CONF_KNX_INDIVIDUAL_ADDRESS: "2.0.4", CONF_KNX_INDIVIDUAL_ADDRESS: "2.0.4",
}, },
) )
assert result3["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert mock_config_entry.data == { assert mock_config_entry.data == {
**DEFAULT_ENTRY_DATA, **DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
@ -1491,43 +1495,8 @@ async def test_options_flow_routing(hass: HomeAssistant, knx_setup) -> None:
knx_setup.assert_called_once() knx_setup.assert_called_once()
async def test_options_communication_settings( async def test_reconfigure_update_keyfile(hass: HomeAssistant, knx_setup) -> None:
hass: HomeAssistant, knx_setup, mock_config_entry: MockConfigEntry """Test reconfigure flow updating keyfile when tunnel endpoint is already configured."""
) -> 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."""
start_data = { start_data = {
**DEFAULT_ENTRY_DATA, **DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
@ -1549,9 +1518,10 @@ async def test_options_update_keyfile(hass: HomeAssistant, knx_setup) -> None:
) )
mock_config_entry.add_to_hass(hass) mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id) 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"], menu_step["flow_id"],
{"next_step_id": "secure_knxkeys"}, {"next_step_id": "secure_knxkeys"},
) )
@ -1559,15 +1529,15 @@ async def test_options_update_keyfile(hass: HomeAssistant, knx_setup) -> None:
assert result["step_id"] == "secure_knxkeys" assert result["step_id"] == "secure_knxkeys"
with patch_file_upload(): with patch_file_upload():
result2 = await hass.config_entries.options.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
CONF_KNX_KNXKEY_PASSWORD: "password", CONF_KNX_KNXKEY_PASSWORD: "password",
}, },
) )
assert result2["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.ABORT
assert not result2.get("data") assert result["reason"] == "reconfigure_successful"
assert mock_config_entry.data == { assert mock_config_entry.data == {
**start_data, **start_data,
CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys", CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys",
@ -1578,8 +1548,8 @@ async def test_options_update_keyfile(hass: HomeAssistant, knx_setup) -> None:
knx_setup.assert_called_once() knx_setup.assert_called_once()
async def test_options_keyfile_upload(hass: HomeAssistant, knx_setup) -> None: async def test_reconfigure_keyfile_upload(hass: HomeAssistant, knx_setup) -> None:
"""Test options flow uploading a keyfile for the first time.""" """Test reconfigure flow uploading a keyfile for the first time."""
start_data = { start_data = {
**DEFAULT_ENTRY_DATA, **DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP,
@ -1596,9 +1566,10 @@ async def test_options_keyfile_upload(hass: HomeAssistant, knx_setup) -> None:
) )
mock_config_entry.add_to_hass(hass) mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id) 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"], menu_step["flow_id"],
{"next_step_id": "secure_knxkeys"}, {"next_step_id": "secure_knxkeys"},
) )
@ -1606,7 +1577,7 @@ async def test_options_keyfile_upload(hass: HomeAssistant, knx_setup) -> None:
assert result["step_id"] == "secure_knxkeys" assert result["step_id"] == "secure_knxkeys"
with patch_file_upload(): with patch_file_upload():
result2 = await hass.config_entries.options.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{ {
CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID, CONF_KEYRING_FILE: FIXTURE_UPLOAD_UUID,
@ -1614,17 +1585,17 @@ async def test_options_keyfile_upload(hass: HomeAssistant, knx_setup) -> None:
}, },
) )
assert result2["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result2["step_id"] == "knxkeys_tunnel_select" assert result["step_id"] == "knxkeys_tunnel_select"
result3 = await hass.config_entries.options.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
user_input={ user_input={
CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1", CONF_KNX_TUNNEL_ENDPOINT_IA: "1.0.1",
}, },
) )
assert result3["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.ABORT
assert not result3.get("data") assert result["reason"] == "reconfigure_successful"
assert mock_config_entry.data == { assert mock_config_entry.data == {
**start_data, **start_data,
CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys", CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys",
@ -1637,3 +1608,35 @@ async def test_options_keyfile_upload(hass: HomeAssistant, knx_setup) -> None:
CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None,
} }
knx_setup.assert_called_once() 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"
result = 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 result["type"] is FlowResultType.CREATE_ENTRY
assert not result.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()