diff --git a/homeassistant/components/knx/__init__.py b/homeassistant/components/knx/__init__.py
index 60104545dea..91d1d9fa1c5 100644
--- a/homeassistant/components/knx/__init__.py
+++ b/homeassistant/components/knx/__init__.py
@@ -12,7 +12,7 @@ from xknx import XKNX
from xknx.core import XknxConnectionState
from xknx.core.telegram_queue import TelegramQueue
from xknx.dpt import DPTArray, DPTBase, DPTBinary
-from xknx.exceptions import ConversionError, XKNXException
+from xknx.exceptions import ConversionError, CouldNotParseTelegram, XKNXException
from xknx.io import ConnectionConfig, ConnectionType, SecureConfig
from xknx.telegram import AddressFilter, Telegram
from xknx.telegram.address import (
@@ -513,31 +513,29 @@ class KNXModule:
)
):
data = telegram.payload.value.value
-
- if isinstance(data, tuple):
- if transcoder := (
- self._group_address_transcoder.get(telegram.destination_address)
- or next(
+ if transcoder := (
+ self._group_address_transcoder.get(telegram.destination_address)
+ or next(
+ (
+ _transcoder
+ for _filter, _transcoder in self._address_filter_transcoder.items()
+ if _filter.match(telegram.destination_address)
+ ),
+ None,
+ )
+ ):
+ try:
+ value = transcoder.from_knx(telegram.payload.value)
+ except (ConversionError, CouldNotParseTelegram) as err:
+ _LOGGER.warning(
(
- _transcoder
- for _filter, _transcoder in self._address_filter_transcoder.items()
- if _filter.match(telegram.destination_address)
+ "Error in `knx_event` at decoding type '%s' from"
+ " telegram %s\n%s"
),
- None,
+ transcoder.__name__,
+ telegram,
+ err,
)
- ):
- try:
- value = transcoder.from_knx(data)
- except ConversionError as err:
- _LOGGER.warning(
- (
- "Error in `knx_event` at decoding type '%s' from"
- " telegram %s\n%s"
- ),
- transcoder.__name__,
- telegram,
- err,
- )
self.hass.bus.async_fire(
"knx_event",
@@ -656,7 +654,7 @@ class KNXModule:
transcoder = DPTBase.parse_transcoder(attr_type)
if transcoder is None:
raise ValueError(f"Invalid type for knx.send service: {attr_type}")
- payload = DPTArray(transcoder.to_knx(attr_payload))
+ payload = transcoder.to_knx(attr_payload)
elif isinstance(attr_payload, int):
payload = DPTBinary(attr_payload)
else:
diff --git a/homeassistant/components/knx/config_flow.py b/homeassistant/components/knx/config_flow.py
index 85e23cbe547..81610d62dcf 100644
--- a/homeassistant/components/knx/config_flow.py
+++ b/homeassistant/components/knx/config_flow.py
@@ -9,10 +9,15 @@ from typing import Any, Final
import voluptuous as vol
from xknx import XKNX
-from xknx.exceptions.exception import CommunicationError, InvalidSecureConfiguration
+from xknx.exceptions.exception import (
+ CommunicationError,
+ InvalidSecureConfiguration,
+ XKNXException,
+)
from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
from xknx.io.gateway_scanner import GatewayDescriptor, GatewayScanner
from xknx.io.self_description import request_description
+from xknx.io.util import validate_ip as xknx_validate_ip
from xknx.secure.keyring import Keyring, XMLInterface, sync_load_keyring
from homeassistant.components.file_upload import process_uploaded_file
@@ -258,21 +263,25 @@ class KNXCommonFlow(ABC, FlowHandler):
if user_input is not None:
try:
- _host = ip_v4_validator(user_input[CONF_HOST], multicast=False)
- except vol.Invalid:
+ _host = user_input[CONF_HOST]
+ _host_ip = await xknx_validate_ip(_host)
+ ip_v4_validator(_host_ip, multicast=False)
+ except (vol.Invalid, XKNXException):
errors[CONF_HOST] = "invalid_ip_address"
- if _local_ip := user_input.get(CONF_KNX_LOCAL_IP):
+ _local_ip = None
+ if _local := user_input.get(CONF_KNX_LOCAL_IP):
try:
- _local_ip = ip_v4_validator(_local_ip, multicast=False)
- except vol.Invalid:
+ _local_ip = await xknx_validate_ip(_local)
+ ip_v4_validator(_local_ip, multicast=False)
+ except (vol.Invalid, XKNXException):
errors[CONF_KNX_LOCAL_IP] = "invalid_ip_address"
selected_tunnelling_type = user_input[CONF_KNX_TUNNELING_TYPE]
if not errors:
try:
self._selected_tunnel = await request_description(
- gateway_ip=_host,
+ gateway_ip=_host_ip,
gateway_port=user_input[CONF_PORT],
local_ip=_local_ip,
route_back=user_input[CONF_KNX_ROUTE_BACK],
@@ -296,7 +305,7 @@ class KNXCommonFlow(ABC, FlowHandler):
host=_host,
port=user_input[CONF_PORT],
route_back=user_input[CONF_KNX_ROUTE_BACK],
- local_ip=_local_ip,
+ local_ip=_local,
device_authentication=None,
user_id=None,
user_password=None,
@@ -636,10 +645,11 @@ class KNXCommonFlow(ABC, FlowHandler):
ip_v4_validator(_multicast_group, multicast=True)
except vol.Invalid:
errors[CONF_KNX_MCAST_GRP] = "invalid_ip_address"
- if _local_ip := user_input.get(CONF_KNX_LOCAL_IP):
+ if _local := user_input.get(CONF_KNX_LOCAL_IP):
try:
+ _local_ip = await xknx_validate_ip(_local)
ip_v4_validator(_local_ip, multicast=False)
- except vol.Invalid:
+ except (vol.Invalid, XKNXException):
errors[CONF_KNX_LOCAL_IP] = "invalid_ip_address"
if not errors:
@@ -653,7 +663,7 @@ class KNXCommonFlow(ABC, FlowHandler):
individual_address=_individual_address,
multicast_group=_multicast_group,
multicast_port=_multicast_port,
- local_ip=_local_ip,
+ local_ip=_local,
device_authentication=None,
user_id=None,
user_password=None,
diff --git a/homeassistant/components/knx/manifest.json b/homeassistant/components/knx/manifest.json
index 0ad4404290a..d3aeced46c9 100644
--- a/homeassistant/components/knx/manifest.json
+++ b/homeassistant/components/knx/manifest.json
@@ -9,5 +9,5 @@
"iot_class": "local_push",
"loggers": ["xknx"],
"quality_scale": "platinum",
- "requirements": ["xknx==2.7.0"]
+ "requirements": ["xknx==2.9.0"]
}
diff --git a/homeassistant/components/knx/schema.py b/homeassistant/components/knx/schema.py
index a505714c0d0..0f627b724cb 100644
--- a/homeassistant/components/knx/schema.py
+++ b/homeassistant/components/knx/schema.py
@@ -10,7 +10,7 @@ from typing import Any, ClassVar, Final
import voluptuous as vol
from xknx.devices.climate import SetpointShiftMode
from xknx.dpt import DPTBase, DPTNumeric, DPTString
-from xknx.exceptions import ConversionError, CouldNotParseAddress
+from xknx.exceptions import ConversionError, CouldNotParseAddress, CouldNotParseTelegram
from xknx.telegram.address import IndividualAddress, parse_device_group_address
from homeassistant.components.binary_sensor import (
@@ -185,13 +185,13 @@ def button_payload_sub_validator(entity_config: OrderedDict) -> OrderedDict:
raise vol.Invalid(f"'type: {_type}' is not a valid sensor type.")
entity_config[CONF_PAYLOAD_LENGTH] = transcoder.payload_length
try:
- entity_config[CONF_PAYLOAD] = int.from_bytes(
- transcoder.to_knx(_payload), byteorder="big"
- )
- except ConversionError as ex:
+ _dpt_payload = transcoder.to_knx(_payload)
+ _raw_payload = transcoder.validate_payload(_dpt_payload)
+ except (ConversionError, CouldNotParseTelegram) as ex:
raise vol.Invalid(
f"'payload: {_payload}' not valid for 'type: {_type}'"
) from ex
+ entity_config[CONF_PAYLOAD] = int.from_bytes(_raw_payload, byteorder="big")
return entity_config
_payload = entity_config[CONF_PAYLOAD]
diff --git a/homeassistant/components/knx/strings.json b/homeassistant/components/knx/strings.json
index a781f9d73cc..800945ab6bd 100644
--- a/homeassistant/components/knx/strings.json
+++ b/homeassistant/components/knx/strings.json
@@ -23,13 +23,13 @@
"port": "[%key:common::config_flow::data::port%]",
"host": "[%key:common::config_flow::data::host%]",
"route_back": "Route back / NAT mode",
- "local_ip": "Local IP of Home Assistant"
+ "local_ip": "Local IP interface"
},
"data_description": {
"port": "Port of the KNX/IP tunneling device.",
- "host": "IP address of the KNX/IP tunneling device.",
+ "host": "IP address or hostname of the KNX/IP tunneling device.",
"route_back": "Enable if your KNXnet/IP tunneling server is behind NAT. Only applies for UDP connections.",
- "local_ip": "Leave blank to use auto-discovery."
+ "local_ip": "Local IP or interface name used for the connection from Home Assistant. Leave blank to use auto-discovery."
}
},
"secure_key_source": {
@@ -93,11 +93,11 @@
"routing_secure": "Use KNX IP Secure",
"multicast_group": "Multicast group",
"multicast_port": "Multicast port",
- "local_ip": "Local IP of Home Assistant"
+ "local_ip": "[%key:component::knx::config::step::manual_tunnel::data::local_ip%]"
},
"data_description": {
"individual_address": "KNX address to be used by Home Assistant, e.g. `0.0.4`",
- "local_ip": "Leave blank to use auto-discovery."
+ "local_ip": "[%key:component::knx::config::step::manual_tunnel::data_description::local_ip%]"
}
}
},
diff --git a/requirements_all.txt b/requirements_all.txt
index 529ebde6c3c..fe0cb3a76d4 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -2664,7 +2664,7 @@ xbox-webapi==2.0.11
xiaomi-ble==0.17.0
# homeassistant.components.knx
-xknx==2.7.0
+xknx==2.9.0
# homeassistant.components.bluesound
# homeassistant.components.fritz
diff --git a/requirements_test_all.txt b/requirements_test_all.txt
index dfb982cb30e..5300ddf973a 100644
--- a/requirements_test_all.txt
+++ b/requirements_test_all.txt
@@ -1922,7 +1922,7 @@ xbox-webapi==2.0.11
xiaomi-ble==0.17.0
# homeassistant.components.knx
-xknx==2.7.0
+xknx==2.9.0
# homeassistant.components.bluesound
# homeassistant.components.fritz
diff --git a/tests/components/knx/test_button.py b/tests/components/knx/test_button.py
index 4fa8d02716f..eb3fee7eaf5 100644
--- a/tests/components/knx/test_button.py
+++ b/tests/components/knx/test_button.py
@@ -1,9 +1,13 @@
"""Test KNX button."""
from datetime import timedelta
+import logging
+
+import pytest
from homeassistant.components.knx.const import (
CONF_PAYLOAD,
CONF_PAYLOAD_LENGTH,
+ DOMAIN,
KNX_ADDRESS,
)
from homeassistant.components.knx.schema import ButtonSchema
@@ -86,3 +90,49 @@ async def test_button_type(hass: HomeAssistant, knx: KNXTestKit) -> None:
"button", "press", {"entity_id": "button.test"}, blocking=True
)
await knx.assert_write("1/2/3", (0x0C, 0x33))
+
+
+@pytest.mark.parametrize(
+ ("conf_type", "conf_value", "error_msg"),
+ [
+ (
+ "2byte_float",
+ "not_valid",
+ "'payload: not_valid' not valid for 'type: 2byte_float'",
+ ),
+ (
+ "not_valid",
+ 3,
+ "type 'not_valid' is not a valid DPT identifier",
+ ),
+ ],
+)
+async def test_button_invalid(
+ hass: HomeAssistant,
+ caplog: pytest.LogCaptureFixture,
+ knx: KNXTestKit,
+ conf_type: str,
+ conf_value: str,
+ error_msg: str,
+) -> None:
+ """Test KNX button with configured payload that can't be encoded."""
+ with caplog.at_level(logging.ERROR):
+ await knx.setup_integration(
+ {
+ ButtonSchema.PLATFORM: {
+ CONF_NAME: "test",
+ KNX_ADDRESS: "1/2/3",
+ ButtonSchema.CONF_VALUE: conf_value,
+ CONF_TYPE: conf_type,
+ }
+ }
+ )
+ assert len(caplog.messages) == 2
+ record = caplog.records[0]
+ assert record.levelname == "ERROR"
+ assert f"Invalid config for [knx]: {error_msg}" in record.message
+ record = caplog.records[1]
+ assert record.levelname == "ERROR"
+ assert "Setup failed for knx: Invalid config." in record.message
+ assert hass.states.get("button.test") is None
+ assert hass.data.get(DOMAIN) is None
diff --git a/tests/components/knx/test_events.py b/tests/components/knx/test_events.py
index a20c6663f08..f5c1aed7fde 100644
--- a/tests/components/knx/test_events.py
+++ b/tests/components/knx/test_events.py
@@ -1,4 +1,7 @@
"""Test KNX events."""
+import logging
+
+import pytest
from homeassistant.components.knx import CONF_EVENT, CONF_TYPE, KNX_ADDRESS
from homeassistant.core import HomeAssistant
@@ -8,7 +11,11 @@ from .conftest import KNXTestKit
from tests.common import async_capture_events
-async def test_knx_event(hass: HomeAssistant, knx: KNXTestKit) -> None:
+async def test_knx_event(
+ hass: HomeAssistant,
+ caplog: pytest.LogCaptureFixture,
+ knx: KNXTestKit,
+) -> None:
"""Test the `knx_event` event."""
test_group_a = "0/4/*"
test_address_a_1 = "0/4/0"
@@ -95,3 +102,16 @@ async def test_knx_event(hass: HomeAssistant, knx: KNXTestKit) -> None:
await knx.receive_write("2/6/6", True)
await hass.async_block_till_done()
assert len(events) == 0
+
+ # receive telegrams with wrong payload length
+ caplog.clear()
+ with caplog.at_level(logging.WARNING):
+ await knx.receive_write(test_address_a_1, (0x03, 0x2F, 0xFF))
+ assert len(caplog.messages) == 1
+ record = caplog.records[0]
+ assert record.levelname == "WARNING"
+ assert (
+ "Error in `knx_event` at decoding type "
+ "'DPT2ByteUnsigned' from telegram" in record.message
+ )
+ await test_event_data(test_address_a_1, (0x03, 0x2F, 0xFF), value=None)