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

Cleanup KNX integration (#68820)

Browse source 
* Cleanup KNX integration

* Cleanup KNX integration

* Update homeassistant/components/knx/__init__.py

Co-authored-by: Matthias Alphart <farmio@alphart.net>

Co-authored-by: Matthias Alphart <farmio@alphart.net>
This commit is contained in:
Marvin Wichmann 2022-03-29 16:46:02 +02:00 • committed by GitHub
parent 62aa7fe10e
commit cec3a08b95
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 211 additions and 500 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -21,7 +21,6 @@ from xknx.telegram.address import (
)
from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite
from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_EVENT,
@ -43,7 +42,13 @@ from .const import (
CONF_KNX_CONNECTION_TYPE,
CONF_KNX_EXPOSE,
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_LOCAL_IP,
CONF_KNX_MCAST_GRP,
CONF_KNX_MCAST_PORT,
CONF_KNX_RATE_LIMIT,
CONF_KNX_ROUTE_BACK,
CONF_KNX_ROUTING,
CONF_KNX_STATE_UPDATER,
CONF_KNX_TUNNELING,
CONF_KNX_TUNNELING_TCP,
DATA_HASS_CONFIG,
@ -57,7 +62,6 @@ from .schema import (
BinarySensorSchema,
ButtonSchema,
ClimateSchema,
ConnectionSchema,
CoverSchema,
EventSchema,
ExposeSchema,
@ -77,9 +81,6 @@ from .schema import (
_LOGGER = logging.getLogger(__name__)
CONF_KNX_FIRE_EVENT: Final = "fire_event"
CONF_KNX_EVENT_FILTER: Final = "event_filter"
SERVICE_KNX_SEND: Final = "send"
SERVICE_KNX_ATTR_PAYLOAD: Final = "payload"
SERVICE_KNX_ATTR_TYPE: Final = "type"
@ -93,26 +94,21 @@ CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.All(
# deprecated since 2021.12
cv.deprecated(ConnectionSchema.CONF_KNX_STATE_UPDATER),
cv.deprecated(ConnectionSchema.CONF_KNX_RATE_LIMIT),
cv.deprecated(CONF_KNX_STATE_UPDATER),
cv.deprecated(CONF_KNX_RATE_LIMIT),
cv.deprecated(CONF_KNX_ROUTING),
cv.deprecated(CONF_KNX_TUNNELING),
cv.deprecated(CONF_KNX_INDIVIDUAL_ADDRESS),
cv.deprecated(ConnectionSchema.CONF_KNX_MCAST_GRP),
cv.deprecated(ConnectionSchema.CONF_KNX_MCAST_PORT),
cv.deprecated(CONF_KNX_EVENT_FILTER),
cv.deprecated(CONF_KNX_MCAST_GRP),
cv.deprecated(CONF_KNX_MCAST_PORT),
cv.deprecated("event_filter"),
# deprecated since 2021.4
cv.deprecated("config_file"),
# deprecated since 2021.2
cv.deprecated(CONF_KNX_FIRE_EVENT),
cv.deprecated("fire_event_filter", replacement_key=CONF_KNX_EVENT_FILTER),
cv.deprecated("fire_event"),
cv.deprecated("fire_event_filter"),
vol.Schema(
{
**ConnectionSchema.SCHEMA,
vol.Optional(CONF_KNX_FIRE_EVENT): cv.boolean,
vol.Optional(CONF_KNX_EVENT_FILTER, default=[]): vol.All(
cv.ensure_list, [cv.string]
),
**EventSchema.SCHEMA,
**ExposeSchema.platform_node(),
**BinarySensorSchema.platform_node(),
@ -211,22 +207,13 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
conf = dict(conf)
hass.data[DATA_KNX_CONFIG] = conf
# Only import if we haven't before.
if not hass.config_entries.async_entries(DOMAIN):
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=conf
)
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Load a config entry."""
# `conf` is None when reloading the integration or no `knx` key in configuration.yaml
if (conf := hass.data.get(DATA_KNX_CONFIG)) is None:
# `config` is None when reloading the integration or no `knx` key in configuration.yaml
if (config := hass.data.get(DATA_KNX_CONFIG)) is None:
_conf = await async_integration_yaml_config(hass, DOMAIN)
if not _conf or DOMAIN not in _conf:
_LOGGER.warning(
@ -235,18 +222,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"for KNX entity configuration documentation"
)
# generate defaults
conf = CONFIG_SCHEMA({DOMAIN: {}})[DOMAIN]
config = CONFIG_SCHEMA({DOMAIN: {}})[DOMAIN]
else:
conf = _conf[DOMAIN]
config = {**conf, **entry.data}
config = _conf[DOMAIN]
try:
knx_module = KNXModule(hass, config, entry)
await knx_module.start()
except XKNXException as ex:
raise ConfigEntryNotReady from ex
hass.data[DATA_KNX_CONFIG] = conf
hass.data[DATA_KNX_CONFIG] = config
hass.data[DOMAIN] = knx_module
if CONF_KNX_EXPOSE in config:
@ -370,12 +355,12 @@ class KNXModule:
def init_xknx(self) -> None:
"""Initialize XKNX object."""
self.xknx = XKNX(
own_address=self.config[CONF_KNX_INDIVIDUAL_ADDRESS],
rate_limit=self.config[ConnectionSchema.CONF_KNX_RATE_LIMIT],
multicast_group=self.config[ConnectionSchema.CONF_KNX_MCAST_GRP],
multicast_port=self.config[ConnectionSchema.CONF_KNX_MCAST_PORT],
own_address=self.entry.data[CONF_KNX_INDIVIDUAL_ADDRESS],
rate_limit=self.entry.data[CONF_KNX_RATE_LIMIT],
multicast_group=self.entry.data[CONF_KNX_MCAST_GRP],
multicast_port=self.entry.data[CONF_KNX_MCAST_PORT],
connection_config=self.connection_config(),
state_updater=self.config[ConnectionSchema.CONF_KNX_STATE_UPDATER],
state_updater=self.entry.data[CONF_KNX_STATE_UPDATER],
)
async def start(self) -> None:
@ -388,29 +373,29 @@ class KNXModule:
def connection_config(self) -> ConnectionConfig:
"""Return the connection_config."""
_conn_type: str = self.config[CONF_KNX_CONNECTION_TYPE]
_conn_type: str = self.entry.data[CONF_KNX_CONNECTION_TYPE]
if _conn_type == CONF_KNX_ROUTING:
return ConnectionConfig(
connection_type=ConnectionType.ROUTING,
local_ip=self.config.get(ConnectionSchema.CONF_KNX_LOCAL_IP),
local_ip=self.entry.data.get(CONF_KNX_LOCAL_IP),
auto_reconnect=True,
threaded=True,
)
if _conn_type == CONF_KNX_TUNNELING:
return ConnectionConfig(
connection_type=ConnectionType.TUNNELING,
gateway_ip=self.config[CONF_HOST],
gateway_port=self.config[CONF_PORT],
local_ip=self.config.get(ConnectionSchema.CONF_KNX_LOCAL_IP),
route_back=self.config.get(ConnectionSchema.CONF_KNX_ROUTE_BACK, False),
gateway_ip=self.entry.data[CONF_HOST],
gateway_port=self.entry.data[CONF_PORT],
local_ip=self.entry.data.get(CONF_KNX_LOCAL_IP),
route_back=self.entry.data.get(CONF_KNX_ROUTE_BACK, False),
auto_reconnect=True,
threaded=True,
)
if _conn_type == CONF_KNX_TUNNELING_TCP:
return ConnectionConfig(
connection_type=ConnectionType.TUNNELING_TCP,
gateway_ip=self.config[CONF_HOST],
gateway_port=self.config[CONF_PORT],
gateway_ip=self.entry.data[CONF_HOST],
gateway_port=self.entry.data[CONF_PORT],
auto_reconnect=True,
threaded=True,
)
@ -475,9 +460,7 @@ class KNXModule:
def register_event_callback(self) -> TelegramQueue.Callback:
"""Register callback for knx_event within XKNX TelegramQueue."""
# backwards compatibility for deprecated CONF_KNX_EVENT_FILTER
# use `address_filters = []` when this is not needed anymore
address_filters = list(map(AddressFilter, self.config[CONF_KNX_EVENT_FILTER]))
address_filters = []
for filter_set in self.config[CONF_EVENT]:
_filters = list(map(AddressFilter, filter_set[KNX_ADDRESS]))
address_filters.extend(_filters)

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

@ -18,25 +18,32 @@ import homeassistant.helpers.config_validation as cv
from .const import (
CONF_KNX_AUTOMATIC,
CONF_KNX_CONNECTION_TYPE,
CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_INITIAL_CONNECTION_TYPES,
CONF_KNX_LOCAL_IP,
CONF_KNX_MCAST_GRP,
CONF_KNX_MCAST_PORT,
CONF_KNX_RATE_LIMIT,
CONF_KNX_ROUTE_BACK,
CONF_KNX_ROUTING,
CONF_KNX_STATE_UPDATER,
CONF_KNX_TUNNELING,
CONF_KNX_TUNNELING_TCP,
DOMAIN,
)
from .schema import ConnectionSchema
CONF_KNX_GATEWAY: Final = "gateway"
CONF_MAX_RATE_LIMIT: Final = 60
CONF_DEFAULT_LOCAL_IP: Final = "0.0.0.0"
DEFAULT_ENTRY_DATA: Final = {
ConnectionSchema.CONF_KNX_STATE_UPDATER: ConnectionSchema.CONF_KNX_DEFAULT_STATE_UPDATER,
ConnectionSchema.CONF_KNX_RATE_LIMIT: ConnectionSchema.CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
}
CONF_KNX_TUNNELING_TYPE: Final = "tunneling_type"
@ -144,12 +151,10 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
CONF_KNX_INDIVIDUAL_ADDRESS: user_input[
CONF_KNX_INDIVIDUAL_ADDRESS
],
ConnectionSchema.CONF_KNX_ROUTE_BACK: (
CONF_KNX_ROUTE_BACK: (
connection_type == CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK
),
ConnectionSchema.CONF_KNX_LOCAL_IP: user_input.get(
ConnectionSchema.CONF_KNX_LOCAL_IP
),
CONF_KNX_LOCAL_IP: user_input.get(CONF_KNX_LOCAL_IP),
CONF_KNX_CONNECTION_TYPE: (
CONF_KNX_TUNNELING_TCP
if connection_type == CONF_KNX_LABEL_TUNNELING_TCP
@ -182,7 +187,7 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
}
if self.show_advanced_options:
fields[vol.Optional(ConnectionSchema.CONF_KNX_LOCAL_IP)] = str
fields[vol.Optional(CONF_KNX_LOCAL_IP)] = str
return self.async_show_form(
step_id="manual_tunnel", data_schema=vol.Schema(fields), errors=errors
@ -195,18 +200,12 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
title=CONF_KNX_ROUTING.capitalize(),
data={
**DEFAULT_ENTRY_DATA,
ConnectionSchema.CONF_KNX_MCAST_GRP: user_input[
ConnectionSchema.CONF_KNX_MCAST_GRP
],
ConnectionSchema.CONF_KNX_MCAST_PORT: user_input[
ConnectionSchema.CONF_KNX_MCAST_PORT
],
CONF_KNX_MCAST_GRP: user_input[CONF_KNX_MCAST_GRP],
CONF_KNX_MCAST_PORT: user_input[CONF_KNX_MCAST_PORT],
CONF_KNX_INDIVIDUAL_ADDRESS: user_input[
CONF_KNX_INDIVIDUAL_ADDRESS
],
ConnectionSchema.CONF_KNX_LOCAL_IP: user_input.get(
ConnectionSchema.CONF_KNX_LOCAL_IP
),
CONF_KNX_LOCAL_IP: user_input.get(CONF_KNX_LOCAL_IP),
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
},
)
@ -216,83 +215,17 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
vol.Required(
CONF_KNX_INDIVIDUAL_ADDRESS, default=XKNX.DEFAULT_ADDRESS
): str,
vol.Required(
ConnectionSchema.CONF_KNX_MCAST_GRP, default=DEFAULT_MCAST_GRP
): str,
vol.Required(
ConnectionSchema.CONF_KNX_MCAST_PORT, default=DEFAULT_MCAST_PORT
): cv.port,
vol.Required(CONF_KNX_MCAST_GRP, default=DEFAULT_MCAST_GRP): str,
vol.Required(CONF_KNX_MCAST_PORT, default=DEFAULT_MCAST_PORT): cv.port,
}
if self.show_advanced_options:
fields[vol.Optional(ConnectionSchema.CONF_KNX_LOCAL_IP)] = str
fields[vol.Optional(CONF_KNX_LOCAL_IP)] = str
return self.async_show_form(
step_id="routing", data_schema=vol.Schema(fields), errors=errors
)
async def async_step_import(self, config: dict | None = None) -> FlowResult:
"""Import a config entry.
Performs a one time import of the YAML configuration and creates a config entry based on it
if not already done before.
"""
if self._async_current_entries() or not config:
return self.async_abort(reason="single_instance_allowed")
data = {
ConnectionSchema.CONF_KNX_RATE_LIMIT: min(
config[ConnectionSchema.CONF_KNX_RATE_LIMIT], CONF_MAX_RATE_LIMIT
),
ConnectionSchema.CONF_KNX_STATE_UPDATER: config[
ConnectionSchema.CONF_KNX_STATE_UPDATER
],
ConnectionSchema.CONF_KNX_MCAST_GRP: config[
ConnectionSchema.CONF_KNX_MCAST_GRP
],
ConnectionSchema.CONF_KNX_MCAST_PORT: config[
ConnectionSchema.CONF_KNX_MCAST_PORT
],
CONF_KNX_INDIVIDUAL_ADDRESS: config[CONF_KNX_INDIVIDUAL_ADDRESS],
}
if CONF_KNX_TUNNELING in config:
return self.async_create_entry(
title=f"{CONF_KNX_TUNNELING.capitalize()} @ {config[CONF_KNX_TUNNELING][CONF_HOST]}",
data={
**DEFAULT_ENTRY_DATA,
CONF_HOST: config[CONF_KNX_TUNNELING][CONF_HOST],
CONF_PORT: config[CONF_KNX_TUNNELING][CONF_PORT],
ConnectionSchema.CONF_KNX_LOCAL_IP: config[CONF_KNX_TUNNELING].get(
ConnectionSchema.CONF_KNX_LOCAL_IP
),
ConnectionSchema.CONF_KNX_ROUTE_BACK: config[CONF_KNX_TUNNELING][
ConnectionSchema.CONF_KNX_ROUTE_BACK
],
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
**data,
},
)
if CONF_KNX_ROUTING in config:
return self.async_create_entry(
title=CONF_KNX_ROUTING.capitalize(),
data={
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
**data,
},
)
return self.async_create_entry(
title=CONF_KNX_AUTOMATIC.capitalize(),
data={
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
**data,
},
)
class KNXOptionsFlowHandler(OptionsFlow):
"""Handle KNX options."""
@ -334,47 +267,44 @@ class KNXOptionsFlowHandler(OptionsFlow):
default=self.current_config[CONF_KNX_INDIVIDUAL_ADDRESS],
): str,
vol.Required(
ConnectionSchema.CONF_KNX_MCAST_GRP,
default=self.current_config.get(
ConnectionSchema.CONF_KNX_MCAST_GRP, DEFAULT_MCAST_GRP
),
CONF_KNX_MCAST_GRP,
default=self.current_config.get(CONF_KNX_MCAST_GRP, DEFAULT_MCAST_GRP),
): str,
vol.Required(
ConnectionSchema.CONF_KNX_MCAST_PORT,
CONF_KNX_MCAST_PORT,
default=self.current_config.get(
ConnectionSchema.CONF_KNX_MCAST_PORT, DEFAULT_MCAST_PORT
CONF_KNX_MCAST_PORT, DEFAULT_MCAST_PORT
),
): cv.port,
}
if self.show_advanced_options:
local_ip = (
self.current_config.get(ConnectionSchema.CONF_KNX_LOCAL_IP)
if self.current_config.get(ConnectionSchema.CONF_KNX_LOCAL_IP)
is not None
self.current_config.get(CONF_KNX_LOCAL_IP)
if self.current_config.get(CONF_KNX_LOCAL_IP) is not None
else CONF_DEFAULT_LOCAL_IP
)
data_schema[
vol.Required(
ConnectionSchema.CONF_KNX_LOCAL_IP,
CONF_KNX_LOCAL_IP,
default=local_ip,
)
] = str
data_schema[
vol.Required(
ConnectionSchema.CONF_KNX_STATE_UPDATER,
CONF_KNX_STATE_UPDATER,
default=self.current_config.get(
ConnectionSchema.CONF_KNX_STATE_UPDATER,
ConnectionSchema.CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_STATE_UPDATER,
CONF_KNX_DEFAULT_STATE_UPDATER,
),
)
] = bool
data_schema[
vol.Required(
ConnectionSchema.CONF_KNX_RATE_LIMIT,
CONF_KNX_RATE_LIMIT,
default=self.current_config.get(
ConnectionSchema.CONF_KNX_RATE_LIMIT,
ConnectionSchema.CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_RATE_LIMIT,
CONF_KNX_DEFAULT_RATE_LIMIT,
),
)
] = vol.All(vol.Coerce(int), vol.Range(min=1, max=CONF_MAX_RATE_LIMIT))
@ -421,11 +351,8 @@ class KNXOptionsFlowHandler(OptionsFlow):
entry_data = {
**DEFAULT_ENTRY_DATA,
**self.general_settings,
ConnectionSchema.CONF_KNX_LOCAL_IP: self.general_settings.get(
ConnectionSchema.CONF_KNX_LOCAL_IP
)
if self.general_settings.get(ConnectionSchema.CONF_KNX_LOCAL_IP)
!= CONF_DEFAULT_LOCAL_IP
CONF_KNX_LOCAL_IP: self.general_settings.get(CONF_KNX_LOCAL_IP)
if self.general_settings.get(CONF_KNX_LOCAL_IP) != CONF_DEFAULT_LOCAL_IP
else None,
CONF_HOST: self.current_config.get(CONF_HOST, ""),
}
@ -436,7 +363,7 @@ class KNXOptionsFlowHandler(OptionsFlow):
**entry_data,
CONF_HOST: user_input[CONF_HOST],
CONF_PORT: user_input[CONF_PORT],
ConnectionSchema.CONF_KNX_ROUTE_BACK: (
CONF_KNX_ROUTE_BACK: (
connection_type == CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK
),
CONF_KNX_CONNECTION_TYPE: (
@ -466,7 +393,7 @@ class KNXOptionsFlowHandler(OptionsFlow):
def get_knx_tunneling_type(config_entry_data: dict) -> str:
"""Obtain the knx tunneling type based on the data in the config entry data."""
connection_type = config_entry_data[CONF_KNX_CONNECTION_TYPE]
route_back = config_entry_data.get(ConnectionSchema.CONF_KNX_ROUTE_BACK, False)
route_back = config_entry_data.get(CONF_KNX_ROUTE_BACK, False)
if route_back and connection_type == CONF_KNX_TUNNELING:
return CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK
if connection_type == CONF_KNX_TUNNELING_TCP:

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

@ -30,11 +30,26 @@ KNX_ADDRESS: Final = "address"
CONF_INVERT: Final = "invert"
CONF_KNX_EXPOSE: Final = "expose"
CONF_KNX_INDIVIDUAL_ADDRESS: Final = "individual_address"
##
# Connection constants
##
CONF_KNX_CONNECTION_TYPE: Final = "connection_type"
CONF_KNX_AUTOMATIC: Final = "automatic"
CONF_KNX_ROUTING: Final = "routing"
CONF_KNX_TUNNELING: Final = "tunneling"
CONF_KNX_TUNNELING_TCP: Final = "tunneling_tcp"
CONF_KNX_LOCAL_IP = "local_ip"
CONF_KNX_MCAST_GRP = "multicast_group"
CONF_KNX_MCAST_PORT = "multicast_port"
CONF_KNX_RATE_LIMIT = "rate_limit"
CONF_KNX_ROUTE_BACK = "route_back"
CONF_KNX_STATE_UPDATER = "state_updater"
CONF_KNX_DEFAULT_STATE_UPDATER = True
CONF_KNX_DEFAULT_RATE_LIMIT = 20
CONF_PAYLOAD: Final = "payload"
CONF_PAYLOAD_LENGTH: Final = "payload_length"
CONF_RESET_AFTER: Final = "reset_after"

58
homeassistant/components/knx/schema.py
View file

@ -6,11 +6,9 @@ from collections import OrderedDict
from typing import Any, ClassVar, Final
import voluptuous as vol
from xknx import XKNX
from xknx.devices.climate import SetpointShiftMode
from xknx.dpt import DPTBase, DPTNumeric
from xknx.exceptions import ConversionError, CouldNotParseAddress
from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
from xknx.telegram.address import IndividualAddress, parse_device_group_address
from homeassistant.components.binary_sensor import (
@ -27,10 +25,8 @@ from homeassistant.const import (
CONF_ENTITY_CATEGORY,
CONF_ENTITY_ID,
CONF_EVENT,
CONF_HOST,
CONF_MODE,
CONF_NAME,
CONF_PORT,
CONF_TYPE,
Platform,
)
@ -40,9 +36,6 @@ from homeassistant.helpers.entity import validate_entity_category
from .const import (
CONF_INVERT,
CONF_KNX_EXPOSE,
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_ROUTING,
CONF_KNX_TUNNELING,
CONF_PAYLOAD,
CONF_PAYLOAD_LENGTH,
CONF_RESET_AFTER,
@ -196,57 +189,6 @@ sync_state_validator = vol.Any(
cv.matches_regex(r"^(init|expire|every)( \d*)?$"),
)
##############
# CONNECTION
##############
class ConnectionSchema:
"""
Voluptuous schema for KNX connection.
DEPRECATED: Migrated to config and options flow. Will be removed in a future version of Home Assistant.
"""
CONF_KNX_LOCAL_IP = "local_ip"
CONF_KNX_MCAST_GRP = "multicast_group"
CONF_KNX_MCAST_PORT = "multicast_port"
CONF_KNX_RATE_LIMIT = "rate_limit"
CONF_KNX_ROUTE_BACK = "route_back"
CONF_KNX_STATE_UPDATER = "state_updater"
CONF_KNX_DEFAULT_STATE_UPDATER = True
CONF_KNX_DEFAULT_RATE_LIMIT = 20
TUNNELING_SCHEMA = vol.Schema(
{
vol.Optional(CONF_PORT, default=DEFAULT_MCAST_PORT): cv.port,
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_KNX_LOCAL_IP): cv.string,
vol.Optional(CONF_KNX_ROUTE_BACK, default=False): cv.boolean,
}
)
ROUTING_SCHEMA = vol.Maybe(vol.Schema({vol.Optional(CONF_KNX_LOCAL_IP): cv.string}))
SCHEMA = {
vol.Exclusive(CONF_KNX_ROUTING, "connection_type"): ROUTING_SCHEMA,
vol.Exclusive(CONF_KNX_TUNNELING, "connection_type"): TUNNELING_SCHEMA,
vol.Optional(
CONF_KNX_INDIVIDUAL_ADDRESS, default=XKNX.DEFAULT_ADDRESS
): ia_validator,
vol.Optional(CONF_KNX_MCAST_GRP, default=DEFAULT_MCAST_GRP): cv.string,
vol.Optional(CONF_KNX_MCAST_PORT, default=DEFAULT_MCAST_PORT): cv.port,
vol.Optional(
CONF_KNX_STATE_UPDATER, default=CONF_KNX_DEFAULT_STATE_UPDATER
): cv.boolean,
vol.Optional(CONF_KNX_RATE_LIMIT, default=CONF_KNX_DEFAULT_RATE_LIMIT): vol.All(
vol.Coerce(int), vol.Range(min=1, max=100)
),
}
#########
# EVENT
#########

15
tests/components/knx/conftest.py
View file

@ -13,11 +13,16 @@ from xknx.telegram import Telegram, TelegramDirection
from xknx.telegram.address import GroupAddress, IndividualAddress
from xknx.telegram.apci import APCI, GroupValueRead, GroupValueResponse, GroupValueWrite
from homeassistant.components.knx import ConnectionSchema
from homeassistant.components.knx.const import (
CONF_KNX_AUTOMATIC,
CONF_KNX_CONNECTION_TYPE,
CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_MCAST_GRP,
CONF_KNX_MCAST_PORT,
CONF_KNX_RATE_LIMIT,
CONF_KNX_STATE_UPDATER,
DOMAIN as KNX_DOMAIN,
)
from homeassistant.core import HomeAssistant
@ -208,10 +213,12 @@ def mock_config_entry() -> MockConfigEntry:
title="KNX",
domain=KNX_DOMAIN,
data={
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
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,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS,
},
)

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

@ -2,12 +2,10 @@
from unittest.mock import patch
import pytest
from xknx import XKNX
from xknx.io import DEFAULT_MCAST_GRP
from xknx.io.gateway_scanner import GatewayDescriptor
from homeassistant import config_entries
from homeassistant.components.knx import ConnectionSchema
from homeassistant.components.knx.config_flow import (
CONF_DEFAULT_LOCAL_IP,
CONF_KNX_GATEWAY,
@ -22,7 +20,13 @@ from homeassistant.components.knx.const import (
CONF_KNX_AUTOMATIC,
CONF_KNX_CONNECTION_TYPE,
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_LOCAL_IP,
CONF_KNX_MCAST_GRP,
CONF_KNX_MCAST_PORT,
CONF_KNX_RATE_LIMIT,
CONF_KNX_ROUTE_BACK,
CONF_KNX_ROUTING,
CONF_KNX_STATE_UPDATER,
CONF_KNX_TUNNELING,
CONF_KNX_TUNNELING_TCP,
DOMAIN,
@ -89,8 +93,8 @@ async def test_routing_setup(hass: HomeAssistant) -> None:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110",
},
)
@ -100,9 +104,9 @@ async def test_routing_setup(hass: HomeAssistant) -> None:
assert result3["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_LOCAL_IP: None,
CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110",
}
@ -141,10 +145,10 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110",
ConnectionSchema.CONF_KNX_LOCAL_IP: "192.168.1.112",
CONF_KNX_LOCAL_IP: "192.168.1.112",
},
)
await hass.async_block_till_done()
@ -153,9 +157,9 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None:
assert result3["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_LOCAL_IP: "192.168.1.112",
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_LOCAL_IP: "192.168.1.112",
CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110",
}
@ -177,8 +181,8 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None:
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: None,
},
),
(
@ -193,8 +197,8 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None:
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: None,
},
),
(
@ -209,8 +213,8 @@ async def test_routing_setup_advanced(hass: HomeAssistant) -> None:
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
CONF_KNX_ROUTE_BACK: True,
CONF_KNX_LOCAL_IP: None,
},
),
],
@ -291,7 +295,7 @@ async def test_tunneling_setup_for_local_ip(hass: HomeAssistant) -> None:
CONF_KNX_TUNNELING_TYPE: CONF_KNX_LABEL_TUNNELING_UDP,
CONF_HOST: "192.168.0.2",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_LOCAL_IP: "192.168.1.112",
CONF_KNX_LOCAL_IP: "192.168.1.112",
},
)
await hass.async_block_till_done()
@ -303,8 +307,8 @@ async def test_tunneling_setup_for_local_ip(hass: HomeAssistant) -> None:
CONF_HOST: "192.168.0.2",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
ConnectionSchema.CONF_KNX_LOCAL_IP: "192.168.1.112",
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: "192.168.1.112",
}
assert len(mock_setup_entry.mock_calls) == 1
@ -361,8 +365,8 @@ async def test_tunneling_setup_for_multiple_found_gateways(hass: HomeAssistant)
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: None,
}
assert len(mock_setup_entry.mock_calls) == 1
@ -422,191 +426,6 @@ async def test_form_with_automatic_connection_handling(hass: HomeAssistant) -> N
assert len(mock_setup_entry.mock_calls) == 1
##
# Import Tests
##
async def test_import_config_tunneling(hass: HomeAssistant) -> None:
"""Test tunneling import from config.yaml."""
config = {
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, # has a default in the original config
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20, # has a default in the original config
ConnectionSchema.CONF_KNX_STATE_UPDATER: True, # has a default in the original config
CONF_KNX_TUNNELING: {
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_LOCAL_IP: "192.168.1.112",
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
},
}
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "Tunneling @ 192.168.1.1"
assert result["data"] == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_LOCAL_IP: "192.168.1.112",
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_config_routing(hass: HomeAssistant) -> None:
"""Test routing import from config.yaml."""
config = {
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, # has a default in the original config
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20, # has a default in the original config
ConnectionSchema.CONF_KNX_STATE_UPDATER: True, # has a default in the original config
CONF_KNX_ROUTING: {}, # is required when using routing
}
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == CONF_KNX_ROUTING.capitalize()
assert result["data"] == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_config_automatic(hass: HomeAssistant) -> None:
"""Test automatic import from config.yaml."""
config = {
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, # has a default in the original config
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20, # has a default in the original config
ConnectionSchema.CONF_KNX_STATE_UPDATER: True, # has a default in the original config
}
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == CONF_KNX_AUTOMATIC.capitalize()
assert result["data"] == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_rate_limit_out_of_range(hass: HomeAssistant) -> None:
"""Test automatic import from config.yaml."""
config = {
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, # has a default in the original config
ConnectionSchema.CONF_KNX_RATE_LIMIT: 80,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True, # has a default in the original config
}
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == CONF_KNX_AUTOMATIC.capitalize()
assert result["data"] == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 60,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_options(hass: HomeAssistant) -> None:
"""Test import from config.yaml with options."""
config = {
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, # has a default in the original config
ConnectionSchema.CONF_KNX_STATE_UPDATER: False,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 30,
}
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == CONF_KNX_AUTOMATIC.capitalize()
assert result["data"] == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_STATE_UPDATER: False,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 30,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_abort_if_entry_exists_already(hass: HomeAssistant) -> None:
"""Test routing import from config.yaml."""
MockConfigEntry(domain=DOMAIN).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}
)
assert result["type"] == "abort"
assert result["reason"] == "single_instance_allowed"
async def test_options_flow(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
@ -629,8 +448,8 @@ async def test_options_flow(
user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
},
)
@ -642,11 +461,11 @@ async def test_options_flow(
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
CONF_HOST: "",
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
CONF_KNX_LOCAL_IP: None,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_RATE_LIMIT: 20,
CONF_KNX_STATE_UPDATER: True,
}
@ -662,14 +481,14 @@ async def test_options_flow(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_RATE_LIMIT: 20,
CONF_KNX_STATE_UPDATER: True,
CONF_KNX_LOCAL_IP: None,
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
CONF_KNX_ROUTE_BACK: True,
},
),
(
@ -681,14 +500,14 @@ async def test_options_flow(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_RATE_LIMIT: 20,
CONF_KNX_STATE_UPDATER: True,
CONF_KNX_LOCAL_IP: None,
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
CONF_KNX_ROUTE_BACK: False,
},
),
(
@ -700,14 +519,14 @@ async def test_options_flow(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_RATE_LIMIT: 20,
CONF_KNX_STATE_UPDATER: True,
CONF_KNX_LOCAL_IP: None,
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
CONF_KNX_ROUTE_BACK: False,
},
),
],
@ -737,8 +556,8 @@ async def test_tunneling_options_flow(
user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
},
)
@ -765,42 +584,42 @@ async def test_tunneling_options_flow(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 25,
ConnectionSchema.CONF_KNX_STATE_UPDATER: False,
ConnectionSchema.CONF_KNX_LOCAL_IP: "192.168.1.112",
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_RATE_LIMIT: 25,
CONF_KNX_STATE_UPDATER: False,
CONF_KNX_LOCAL_IP: "192.168.1.112",
},
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
CONF_HOST: "",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 25,
ConnectionSchema.CONF_KNX_STATE_UPDATER: False,
ConnectionSchema.CONF_KNX_LOCAL_IP: "192.168.1.112",
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_RATE_LIMIT: 25,
CONF_KNX_STATE_UPDATER: False,
CONF_KNX_LOCAL_IP: "192.168.1.112",
},
),
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 25,
ConnectionSchema.CONF_KNX_STATE_UPDATER: False,
ConnectionSchema.CONF_KNX_LOCAL_IP: CONF_DEFAULT_LOCAL_IP,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_RATE_LIMIT: 25,
CONF_KNX_STATE_UPDATER: False,
CONF_KNX_LOCAL_IP: CONF_DEFAULT_LOCAL_IP,
},
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
CONF_HOST: "",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 25,
ConnectionSchema.CONF_KNX_STATE_UPDATER: False,
ConnectionSchema.CONF_KNX_LOCAL_IP: None,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_RATE_LIMIT: 25,
CONF_KNX_STATE_UPDATER: False,
CONF_KNX_LOCAL_IP: None,
},
),
],
@ -843,21 +662,21 @@ async def test_advanced_options(
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
CONF_KNX_ROUTE_BACK: False,
},
CONF_KNX_LABEL_TUNNELING_UDP,
),
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
CONF_KNX_ROUTE_BACK: True,
},
CONF_KNX_LABEL_TUNNELING_UDP_ROUTE_BACK,
),
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP,
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
CONF_KNX_ROUTE_BACK: False,
},
CONF_KNX_LABEL_TUNNELING_TCP,
),

4
tests/components/knx/test_diagnostic.py
View file

@ -30,6 +30,8 @@ async def test_diagnostics(
"individual_address": "15.15.250",
"multicast_group": "224.0.23.12",
"multicast_port": 3671,
"rate_limit": 20,
"state_updater": True,
},
"configuration_error": None,
"configuration_yaml": None,
@ -60,6 +62,8 @@ async def test_diagnostic_config_error(
"individual_address": "15.15.250",
"multicast_group": "224.0.23.12",
"multicast_port": 3671,
"rate_limit": 20,
"state_updater": True,
},
"configuration_error": "extra keys not allowed @ data['knx']['wrong_key']",
"configuration_yaml": {"wrong_key": {}},

14
tests/components/knx/test_events.py
View file

@ -1,11 +1,6 @@
"""Test KNX events."""
from homeassistant.components.knx import (
CONF_EVENT,
CONF_KNX_EVENT_FILTER,
CONF_TYPE,
KNX_ADDRESS,
)
from homeassistant.components.knx import CONF_EVENT, CONF_TYPE, KNX_ADDRESS
from homeassistant.core import HomeAssistant
from .conftest import KNXTestKit
@ -25,7 +20,6 @@ async def test_knx_event(hass: HomeAssistant, knx: KNXTestKit):
test_address_c_1 = "2/6/4"
test_address_c_2 = "2/6/5"
test_address_d = "5/4/3"
test_address_e = "6/4/3"
events = async_capture_events(hass, "knx_event")
async def test_event_data(address, payload, value=None):
@ -63,8 +57,6 @@ async def test_knx_event(hass: HomeAssistant, knx: KNXTestKit):
KNX_ADDRESS: [test_address_d],
},
],
# test legacy `event_filter` config
CONF_KNX_EVENT_FILTER: [test_address_e],
}
)
@ -97,10 +89,6 @@ async def test_knx_event(hass: HomeAssistant, knx: KNXTestKit):
await knx.receive_write(test_address_d, True)
await test_event_data(test_address_d, True)
# test legacy `event_filter` config
await knx.receive_write(test_address_e, (89, 43, 34, 11))
await test_event_data(test_address_e, (89, 43, 34, 11))
# receive telegrams for group addresses not matching any filter
await knx.receive_write("0/5/0", True)
await knx.receive_write("1/7/0", True)

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

@ -1,17 +1,29 @@
"""Test KNX init."""
import pytest
from xknx import XKNX
from xknx.io import ConnectionConfig, ConnectionType
from xknx.io import (
DEFAULT_MCAST_GRP,
DEFAULT_MCAST_PORT,
ConnectionConfig,
ConnectionType,
)
from homeassistant.components.knx.const import (
CONF_KNX_AUTOMATIC,
CONF_KNX_CONNECTION_TYPE,
CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_LOCAL_IP,
CONF_KNX_MCAST_GRP,
CONF_KNX_MCAST_PORT,
CONF_KNX_RATE_LIMIT,
CONF_KNX_ROUTE_BACK,
CONF_KNX_ROUTING,
CONF_KNX_STATE_UPDATER,
CONF_KNX_TUNNELING,
DOMAIN as KNX_DOMAIN,
)
from homeassistant.components.knx.schema import ConnectionSchema
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import HomeAssistant
@ -25,15 +37,24 @@ from tests.common import MockConfigEntry
[
(
{
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS,
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,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS,
},
ConnectionConfig(threaded=True),
),
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
ConnectionSchema.CONF_KNX_LOCAL_IP: "192.168.1.1",
CONF_KNX_LOCAL_IP: "192.168.1.1",
CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS,
},
ConnectionConfig(
connection_type=ConnectionType.ROUTING,
@ -46,8 +67,13 @@ from tests.common import MockConfigEntry
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.0.2",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
ConnectionSchema.CONF_KNX_LOCAL_IP: "192.168.1.112",
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: "192.168.1.112",
CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS,
},
ConnectionConfig(
connection_type=ConnectionType.TUNNELING,