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

Update xknx to 2.9.0 (#91282)

Browse source 
* Update xknx to 2.8.0

* add tests for validators

* Update strings.json

* Update xknx to 2.9.0
This commit is contained in:
Matthias Alphart 2023-04-22 18:25:14 +02:00 • committed by GitHub
parent 2e9dc209f9
commit 95d44e100b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 127 additions and 49 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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:

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

@ -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,

2
homeassistant/components/knx/manifest.json
View file

@ -9,5 +9,5 @@
"iot_class": "local_push",
"loggers": ["xknx"],
"quality_scale": "platinum",
"requirements": ["xknx==2.7.0"]
"requirements": ["xknx==2.9.0"]
}

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

@ -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]

10
homeassistant/components/knx/strings.json
View file

@ -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%]"
}
}
},

2
requirements_all.txt
View file

@ -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

2
requirements_test_all.txt
View file

@ -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

50
tests/components/knx/test_button.py
View file

@ -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

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

@ -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)