Compare commits

...

11 Commits

Author SHA1 Message Date
farmio 73967ec3b8 override decorator 2026-06-23 22:25:28 +02:00
farmio 7e7c846c78 Remove unused logger 2026-06-23 22:24:47 +02:00
farmio 7ff7c45b18 improve typing 2026-06-23 22:24:47 +02:00
farmio 4481e90ee6 fix flaky tests 2026-06-23 22:24:47 +02:00
farmio 7670dcb227 store raw payload as hex string
this helps FE work with payloads >7byte
2026-06-23 22:24:47 +02:00
farmio db0f9d1f54 tests 2026-06-23 22:24:47 +02:00
farmio f1535bf8a4 Schema fixes 2026-06-23 22:24:47 +02:00
farmio c460100fa1 Validation 2026-06-23 22:24:46 +02:00
farmio 53e2a9341f Fix loading raw value 2026-06-23 22:24:46 +02:00
farmio 0e713d549c Rely on store validation 2026-06-23 22:24:46 +02:00
farmio a068574fe2 Add KNX button to UI configured entities 2026-06-23 22:24:46 +02:00
11 changed files with 508 additions and 29 deletions
+84 -14
View File
@@ -1,19 +1,24 @@
"""Support for KNX button entities."""
from typing import override
from typing import Any, override
from xknx.devices import RawValue as XknxRawValue
from xknx.devices import ExposeSensor as XknxExposeSensor, RawValue as XknxRawValue
from homeassistant import config_entries
from homeassistant.components.button import ButtonEntity
from homeassistant.const import CONF_ENTITY_CATEGORY, CONF_NAME, CONF_PAYLOAD, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.entity_platform import (
AddConfigEntryEntitiesCallback,
async_get_current_platform,
)
from homeassistant.helpers.typing import ConfigType
from .const import CONF_PAYLOAD_LENGTH, KNX_ADDRESS, KNX_MODULE_KEY
from .entity import KnxYamlEntity
from .const import CONF_PAYLOAD_LENGTH, CONF_VALUE, DOMAIN, KNX_ADDRESS, KNX_MODULE_KEY
from .entity import KnxUiEntity, KnxUiEntityPlatformController, KnxYamlEntity
from .knx_module import KNXModule
from .storage.const import CONF_DATA, CONF_ENTITY, CONF_GA_SEND
from .storage.util import ConfigExtractor
async def async_setup_entry(
@@ -21,27 +26,60 @@ async def async_setup_entry(
config_entry: config_entries.ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the KNX binary sensor platform."""
"""Set up button(s) for KNX platform."""
knx_module = hass.data[KNX_MODULE_KEY]
config: list[ConfigType] = knx_module.config_yaml[Platform.BUTTON]
platform = async_get_current_platform()
knx_module.config_store.add_platform(
platform=Platform.BUTTON,
controller=KnxUiEntityPlatformController(
knx_module=knx_module,
entity_platform=platform,
entity_class=KnxUiButton,
),
)
async_add_entities(KNXButton(knx_module, entity_config) for entity_config in config)
entities: list[KnxYamlEntity | KnxUiEntity] = []
if yaml_platform_config := knx_module.config_yaml.get(Platform.BUTTON):
entities.extend(
KnxYamlButton(knx_module, entity_config)
for entity_config in yaml_platform_config
)
if ui_config := knx_module.config_store.data["entities"].get(Platform.BUTTON):
entities.extend(
KnxUiButton(knx_module, unique_id, config)
for unique_id, config in ui_config.items()
)
if entities:
async_add_entities(entities)
class KNXButton(KnxYamlEntity, ButtonEntity):
class _KnxButton(ButtonEntity):
"""Representation of a KNX button."""
_device: XknxRawValue | XknxExposeSensor
_payload: Any
@override
async def async_press(self) -> None:
"""Press the button."""
await self._device.set(self._payload)
class KnxYamlButton(_KnxButton, KnxYamlEntity):
"""Representation of a KNX button configured via YAML."""
_device: XknxRawValue
def __init__(self, knx_module: KNXModule, config: ConfigType) -> None:
"""Initialize a KNX button."""
# dpt-value to payload conversion is done in schema validation for yaml config
self._payload = config[CONF_PAYLOAD]
self._device = XknxRawValue(
xknx=knx_module.xknx,
name=config[CONF_NAME],
payload_length=config[CONF_PAYLOAD_LENGTH],
group_address=config[KNX_ADDRESS],
)
self._payload = config[CONF_PAYLOAD]
super().__init__(
knx_module=knx_module,
unique_id=f"{self._device.remote_value.group_address}_{self._payload}",
@@ -49,7 +87,39 @@ class KNXButton(KnxYamlEntity, ButtonEntity):
entity_category=config.get(CONF_ENTITY_CATEGORY),
)
@override
async def async_press(self) -> None:
"""Press the button."""
await self._device.set(self._payload)
class KnxUiButton(_KnxButton, KnxUiEntity):
"""Representation of a KNX button configured via the UI."""
_device: XknxRawValue | XknxExposeSensor
def __init__(
self, knx_module: KNXModule, unique_id: str, config: dict[str, Any]
) -> None:
"""Initialize a KNX button."""
knx_conf = ConfigExtractor(config[DOMAIN])
button_data = knx_conf.get(CONF_DATA)
if CONF_PAYLOAD in button_data and CONF_PAYLOAD_LENGTH in button_data:
self._payload = int(button_data[CONF_PAYLOAD], 16)
self._device = XknxRawValue(
xknx=knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
payload_length=button_data[CONF_PAYLOAD_LENGTH],
group_address=knx_conf.get_write(CONF_GA_SEND),
)
else:
dpt_string = knx_conf.get_dpt(CONF_GA_SEND)
self._payload = button_data[CONF_VALUE]
self._device = XknxExposeSensor(
xknx=knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
value_type=dpt_string,
group_address=knx_conf.get_write(CONF_GA_SEND),
respond_to_read=False,
)
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
)
+2
View File
@@ -26,6 +26,7 @@ KNX_ADDRESS: Final = "address"
CONF_INVERT: Final = "invert"
CONF_KNX_EXPOSE: Final = "expose"
CONF_KNX_INDIVIDUAL_ADDRESS: Final = "individual_address"
CONF_VALUE: Final = "value"
##
# Connection constants
@@ -178,6 +179,7 @@ SUPPORTED_PLATFORMS_YAML: Final = {
SUPPORTED_PLATFORMS_UI: Final = {
Platform.BINARY_SENSOR,
Platform.BUTTON,
Platform.CLIMATE,
Platform.COVER,
Platform.DATE,
+42 -4
View File
@@ -2,9 +2,9 @@
from collections.abc import Mapping
from functools import cache
from typing import Literal, TypedDict
from typing import Literal, NotRequired, TypedDict, cast
from xknx.dpt import DPTBase, DPTComplex, DPTEnum, DPTNumeric
from xknx.dpt import DPTBase, DPTComplex, DPTComplexFieldSchema, DPTEnum, DPTNumeric
from xknx.dpt.dpt_16 import DPTString
from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
@@ -24,15 +24,28 @@ class DPTInfo(TypedDict):
sensor_device_class: SensorDeviceClass | None
sensor_state_class: SensorStateClass | None
payload_length: int
# numeric specific
min: NotRequired[float]
max: NotRequired[float]
step: NotRequired[float]
# enum specific
options: NotRequired[list[str]]
# complex specific
schema: NotRequired[list[DPTComplexFieldSchema]]
@cache
def get_supported_dpts() -> Mapping[str, DPTInfo]:
"""Return a mapping of supported DPTs with HA specific attributes."""
dpts = {}
dpts: dict[str, DPTInfo] = {}
for dpt_class in DPTBase.dpt_class_tree():
dpt_number_str = dpt_class.dpt_number_str()
ha_dpt_class = _ha_dpt_class(dpt_class)
dpts[dpt_number_str] = DPTInfo(
info = DPTInfo(
dpt_class=ha_dpt_class,
main=dpt_class.dpt_main_number, # type: ignore[typeddict-item] # checked in xknx unit tests
sub=dpt_class.dpt_sub_number,
@@ -40,7 +53,15 @@ def get_supported_dpts() -> Mapping[str, DPTInfo]:
unit=_sensor_unit_overrides.get(dpt_number_str, dpt_class.unit),
sensor_device_class=_sensor_device_classes.get(dpt_number_str),
sensor_state_class=_get_sensor_state_class(ha_dpt_class, dpt_number_str),
payload_length=dpt_class.payload_length,
)
if ha_dpt_class == "numeric":
_add_numeric_details(info, cast(type[DPTNumeric], dpt_class))
elif ha_dpt_class == "enum":
_add_enum_details(info, cast(type[DPTEnum], dpt_class))
elif ha_dpt_class == "complex":
_add_complex_details(info, cast(type[DPTComplex], dpt_class))
dpts[dpt_number_str] = info
return dpts
@@ -57,6 +78,23 @@ def _ha_dpt_class(dpt_cls: type[DPTBase]) -> HaDptClass:
raise ValueError("Unsupported DPT class")
def _add_numeric_details(dpt_info: DPTInfo, dpt_cls: type[DPTNumeric]) -> None:
"""Add numeric specific details to the DPTInfo."""
dpt_info["min"] = dpt_cls.value_min
dpt_info["max"] = dpt_cls.value_max
dpt_info["step"] = dpt_cls.resolution
def _add_enum_details(dpt_info: DPTInfo, dpt_cls: type[DPTEnum]) -> None:
"""Add enum specific details to the DPTInfo."""
dpt_info["options"] = [o.name.lower() for o in dpt_cls.get_valid_values()]
def _add_complex_details(dpt_info: DPTInfo, dpt_cls: type[DPTComplex]) -> None:
"""Add complex specific details to the DPTInfo."""
dpt_info["schema"] = dpt_cls.get_dict_schema()
_sensor_device_classes: Mapping[str, SensorDeviceClass] = {
"7.011": SensorDeviceClass.DISTANCE,
"7.012": SensorDeviceClass.CURRENT,
+6 -4
View File
@@ -57,6 +57,7 @@ from .const import (
CONF_RESPOND_TO_READ,
CONF_STATE_ADDRESS,
CONF_SYNC_STATE,
CONF_VALUE,
KNX_ADDRESS,
ClimateConf,
ColorTempModes,
@@ -98,9 +99,12 @@ def _max_payload_value(payload_length: int) -> int:
def button_payload_sub_validator(entity_config: OrderedDict) -> OrderedDict:
"""Validate a button entity payload configuration."""
"""Validate a button entity payload configuration.
Returns raw payload and length from value and type (DPT), if given.
"""
if _type := entity_config.get(CONF_TYPE):
_payload = entity_config[ButtonSchema.CONF_VALUE]
_payload = entity_config[CONF_VALUE]
if (transcoder := DPTBase.parse_transcoder(_type)) is None:
raise vol.Invalid(f"'type: {_type}' is not a valid sensor type.")
entity_config[CONF_PAYLOAD_LENGTH] = transcoder.payload_length
@@ -234,8 +238,6 @@ class ButtonSchema(KNXPlatformSchema):
PLATFORM = Platform.BUTTON
CONF_VALUE = "value"
payload_or_value_msg = f"Please use only one of `{CONF_PAYLOAD}` or `{CONF_VALUE}`"
length_or_type_msg = (
f"Please use only one of `{CONF_PAYLOAD_LENGTH}` or `{CONF_TYPE}`"
@@ -19,6 +19,9 @@ CONF_GA_TIME: Final = "ga_time"
CONF_GA_STEP: Final = "ga_step"
# Button
CONF_GA_SEND: Final = "ga_send"
# Climate
CONF_GA_TEMPERATURE_CURRENT: Final = "ga_temperature_current"
CONF_GA_HUMIDITY_CURRENT: Final = "ga_humidity_current"
@@ -3,7 +3,8 @@
from enum import StrEnum, unique
import voluptuous as vol
from xknx.dpt import DPTNumeric
from xknx.dpt import DPTBase, DPTBinary, DPTNumeric
from xknx.exceptions import ConversionError
from homeassistant.components.climate import HVACMode
from homeassistant.components.number import (
@@ -36,9 +37,11 @@ from ..const import (
CONF_CONTEXT_TIMEOUT,
CONF_IGNORE_INTERNAL_STATE,
CONF_INVERT,
CONF_PAYLOAD_LENGTH,
CONF_RESET_AFTER,
CONF_RESPOND_TO_READ,
CONF_SYNC_STATE,
CONF_VALUE,
DOMAIN,
SUPPORTED_PLATFORMS_UI,
ClimateConf,
@@ -92,6 +95,7 @@ from .const import (
CONF_GA_RED_SWITCH,
CONF_GA_SATURATION,
CONF_GA_SCENE,
CONF_GA_SEND,
CONF_GA_SENSOR,
CONF_GA_SETPOINT_SHIFT,
CONF_GA_SPEED,
@@ -115,6 +119,7 @@ from .knx_selector import (
GASelector,
GroupSelect,
GroupSelectOption,
KnxPayloadSelector,
KNXSectionFlat,
SyncStateSelector,
)
@@ -169,6 +174,55 @@ BINARY_SENSOR_KNX_SCHEMA = vol.Schema(
},
)
def _button_data_sub_validator(config: dict) -> dict:
"""Validate data matching configured DPT."""
dpt = config[CONF_GA_SEND].get(CONF_DPT)
transcoder = None
if dpt:
transcoder = DPTBase.parse_transcoder(dpt)
assert transcoder is not None # already checked by GASelector
if CONF_VALUE in config[CONF_DATA]:
try:
transcoder.to_knx(config[CONF_DATA][CONF_VALUE])
except ConversionError as ex:
raise vol.Invalid(
f"Value invalid for DPT {transcoder.dpt_number_str()}",
path=([CONF_DATA]),
) from ex
elif CONF_PAYLOAD_LENGTH in config[CONF_DATA]:
length = config[CONF_DATA][CONF_PAYLOAD_LENGTH]
if length != transcoder.payload_length or (
length != 0 and transcoder.payload_type is DPTBinary
):
raise vol.Invalid(
f"Payload length invalid for DPT {transcoder.dpt_number_str()}",
path=([CONF_DATA]),
)
return config
# without DPT only raw allowed -> payload + payload_length (checked by KnxPayloadSelector)
if CONF_PAYLOAD_LENGTH in config[CONF_DATA]:
return config
raise vol.Invalid("Invalid configuration for button entity")
BUTTON_KNX_SCHEMA = AllSerializeFirst(
vol.Schema(
{
vol.Required(CONF_GA_SEND): GASelector(
state=False,
write_required=True,
passive=False,
dpt=["numeric", "enum", "complex", "string"],
dpt_required=False, # for raw payload support
),
vol.Required(CONF_DATA): KnxPayloadSelector(ga_path=CONF_GA_SEND),
},
),
_button_data_sub_validator,
)
COVER_KNX_SCHEMA = AllSerializeFirst(
vol.Schema(
{
@@ -741,6 +795,7 @@ SENSOR_KNX_SCHEMA = AllSerializeFirst(
KNX_SCHEMA_FOR_PLATFORM = {
Platform.BINARY_SENSOR: BINARY_SENSOR_KNX_SCHEMA,
Platform.BUTTON: BUTTON_KNX_SCHEMA,
Platform.CLIMATE: CLIMATE_KNX_SCHEMA,
Platform.COVER: COVER_KNX_SCHEMA,
Platform.DATE: DATE_KNX_SCHEMA,
@@ -6,6 +6,9 @@ from typing import Any, override
import voluptuous as vol
from homeassistant.const import CONF_PAYLOAD
from ..const import CONF_PAYLOAD_LENGTH, CONF_VALUE
from ..dpt import HaDptClass, get_supported_dpts
from ..validation import ga_validator, maybe_ga_validator, sync_state_validator
from .const import CONF_DPT, CONF_GA_PASSIVE, CONF_GA_STATE, CONF_GA_WRITE
@@ -159,7 +162,11 @@ class GroupSelect(KNXSelectorBase):
class GASelector(KNXSelectorBase):
"""Selector for a KNX group address structure."""
"""Selector for a KNX group address structure.
`dpt_required` optional dpt only apply to dpt-class lists, enums are always required.
`valid_dpt` is used in frontend to filter dropdown menu - no validation is done.
"""
selector_type = "knx_group_address"
@@ -171,6 +178,7 @@ class GASelector(KNXSelectorBase):
write_required: bool = False,
state_required: bool = False,
dpt: type[Enum] | list[HaDptClass] | None = None,
dpt_required: bool = True,
valid_dpt: str | Iterable[str] | None = None,
) -> None:
"""Initialize the group address selector."""
@@ -180,7 +188,7 @@ class GASelector(KNXSelectorBase):
self.write_required = write_required
self.state_required = state_required
self.dpt = dpt
# valid_dpt is used in frontend to filter dropdown menu - no validation is done
self.dpt_required = dpt_required
self.valid_dpt = (valid_dpt,) if isinstance(valid_dpt, str) else valid_dpt
self.schema = self.build_schema()
@@ -196,6 +204,7 @@ class GASelector(KNXSelectorBase):
}
if self.dpt is not None:
if isinstance(self.dpt, list):
# optional / required is not passed to FE - only validated in BE
options["dptClasses"] = self.dpt
else:
options["dptSelect"] = [
@@ -267,7 +276,8 @@ class GASelector(KNXSelectorBase):
"""Add DPT validator to the schema."""
if self.dpt is not None:
if isinstance(self.dpt, list):
schema[vol.Required(CONF_DPT)] = vol.In(get_supported_dpts())
marker = vol.Required if self.dpt_required else vol.Optional
schema[marker(CONF_DPT)] = vol.In(get_supported_dpts())
else:
schema[vol.Required(CONF_DPT)] = vol.In(
{item.value for item in self.dpt}
@@ -300,3 +310,64 @@ class SyncStateSelector(KNXSelectorBase):
if not self.allow_false and not data:
raise vol.Invalid(f"Sync state cannot be {data}")
return self.schema(data)
class KnxPayloadSelector(KNXSelectorBase):
"""Selector for KNX payload configuration.
Raw payloads are stored as hex strings.
"""
schema = vol.Any(
{
vol.Required(CONF_VALUE): object,
},
{
vol.Required(CONF_PAYLOAD): str,
vol.Required(CONF_PAYLOAD_LENGTH): vol.All(int, vol.Range(min=0, max=14)),
},
)
selector_type = "knx_payload"
def __init__(self, ga_path: str) -> None:
"""Initialize the KNX payload selector."""
self.ga_path = ga_path
@override
def serialize(self) -> dict[str, Any]:
"""Serialize the selector to a dictionary."""
return {
"type": self.selector_type,
"ga_path": self.ga_path,
}
@override
def __call__(self, data: Any) -> Any:
"""Validate the passed data."""
validated = self.schema(data)
if CONF_PAYLOAD in validated and CONF_PAYLOAD_LENGTH in validated:
payload = validated[CONF_PAYLOAD]
payload_length = validated[CONF_PAYLOAD_LENGTH]
try:
int_payload = int(payload, 16)
except ValueError as ex:
raise vol.Invalid(f"Invalid payload format: {payload}") from ex
validated[CONF_PAYLOAD] = hex(int_payload) # prepends "0x" if not present
if int_payload < 0:
raise vol.Invalid(f"Payload cannot be negative: {payload}")
if payload_length == 0:
# DPT 1,2,3 is marked length 0, has 6 bit size
if int_payload > 63:
raise vol.Invalid(
f"Payload exceeds DPT 1,2,3 limit of 0x3f (63): {payload}"
)
else:
max_payload = (1 << (payload_length * 8)) - 1
if int_payload > max_payload:
raise vol.Invalid(
f"Payload {payload} exceeds possible maximum for "
f"length {payload_length}: {hex(max_payload)}"
)
# CONF_VALUE branch needs subvalidator as we don't have the DPT available here
return validated
+26
View File
@@ -453,6 +453,19 @@
}
}
},
"button": {
"description": "Entity for sending predefined values.",
"knx": {
"data": {
"description": "The value sent when the button is pressed. The format of the value depends on the DPT of the configured address.",
"label": "Data"
},
"ga_send": {
"description": "Group address the value is sent to.",
"label": "Address"
}
}
},
"climate": {
"description": "The KNX climate platform is used as an interface to heating actuators, HVAC gateways, etc.",
"knx": {
@@ -1014,6 +1027,19 @@
"project": {
"description": "Inspect imported group addresses",
"title": "Project"
},
"selectors": {
"knx-payload-selector": {
"dpt_missing": "No DPT selected Typed mode not available",
"mode": {
"label": "Payload format",
"raw": "Raw payload",
"typed": "Typed value"
},
"raw_length": "Payload length",
"raw_length_description": "Length of the raw payload in bytes. For DPT 1, 2 and 3 use `0`.",
"raw_payload": "Raw payload"
}
}
},
"device_automation": {
@@ -0,0 +1,44 @@
{
"version": 2,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {
"button": {
"knx_es_01KVFEGP54VJW94TR9GQW2XA4R": {
"entity": {
"name": "test raw",
"device_info": null,
"entity_category": null
},
"knx": {
"data": {
"payload": "0x1",
"payload_length": 1
},
"ga_send": {
"write": "1/1/1"
}
}
},
"knx_es_01KVFEHE937CQGWP81RZNQ6D8E": {
"entity": {
"name": "test typed",
"device_info": null,
"entity_category": null
},
"knx": {
"ga_send": {
"write": "1/1/2",
"dpt": "1.001"
},
"data": {
"value": "on"
}
}
}
}
},
"time_server": {}
}
}
@@ -129,6 +129,39 @@
'type': 'result',
})
# ---
# name: test_knx_get_schema[button]
dict({
'id': 1,
'result': list([
dict({
'name': 'ga_send',
'options': dict({
'dptClasses': list([
'numeric',
'enum',
'complex',
'string',
]),
'passive': False,
'state': False,
'write': dict({
'required': True,
}),
}),
'required': True,
'type': 'knx_group_address',
}),
dict({
'ga_path': 'ga_send',
'name': 'data',
'required': True,
'type': 'knx_payload',
}),
]),
'success': True,
'type': 'result',
})
# ---
# name: test_knx_get_schema[climate]
dict({
'id': 1,
+138 -3
View File
@@ -2,22 +2,32 @@
from datetime import timedelta
import logging
from typing import Any
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components.knx.const import (
CONF_PAYLOAD_LENGTH,
CONF_VALUE,
KNX_ADDRESS,
KNX_MODULE_KEY,
)
from homeassistant.components.knx.schema import ButtonSchema
from homeassistant.const import CONF_NAME, CONF_PAYLOAD, CONF_TYPE
from homeassistant.const import (
CONF_NAME,
CONF_PAYLOAD,
CONF_TYPE,
STATE_UNKNOWN,
Platform,
)
from homeassistant.core import HomeAssistant
from . import KnxEntityGenerator
from .conftest import KNXTestKit
from tests.common import async_capture_events, async_fire_time_changed
from tests.typing import WebSocketGenerator
async def test_button_simple(
@@ -83,7 +93,7 @@ async def test_button_type(hass: HomeAssistant, knx: KNXTestKit) -> None:
ButtonSchema.PLATFORM: {
CONF_NAME: "test",
KNX_ADDRESS: "1/2/3",
ButtonSchema.CONF_VALUE: 21.5,
CONF_VALUE: 21.5,
CONF_TYPE: "2byte_float",
}
}
@@ -125,7 +135,7 @@ async def test_button_invalid(
ButtonSchema.PLATFORM: {
CONF_NAME: "test",
KNX_ADDRESS: "1/2/3",
ButtonSchema.CONF_VALUE: conf_value,
CONF_VALUE: conf_value,
CONF_TYPE: conf_type,
}
}
@@ -139,3 +149,128 @@ async def test_button_invalid(
assert "Setup failed for 'knx': Invalid config." in record.message
assert hass.states.get("button.test") is None
assert hass.data.get(KNX_MODULE_KEY) is None
@pytest.mark.parametrize(
"knx_config",
[
(
{
"ga_send": {"write": "1/1/1"},
"data": {"payload": "1", "payload_length": 1}, # raw payload
}
),
(
{
"ga_send": {"write": "1/1/1", "dpt": "5"}, # generic 1byte uint
"data": {"payload": "0x01", "payload_length": 1}, # raw payload
}
),
(
{
"ga_send": {"write": "1/1/1", "dpt": "5"}, # generic 1byte uint
"data": {"value": 1}, # typed value
}
),
],
)
async def test_button_ui_create(
hass: HomeAssistant,
knx: KNXTestKit,
create_ui_entity: KnxEntityGenerator,
knx_config: dict[str, Any],
) -> None:
"""Test creating a button."""
await knx.setup_integration()
await create_ui_entity(
platform=Platform.BUTTON,
entity_data={"name": "test"},
knx_data=knx_config,
)
await hass.services.async_call(
"button", "press", {"entity_id": "button.test"}, blocking=True
)
await knx.assert_write("1/1/1", (1,))
async def test_button_ui_load(hass: HomeAssistant, knx: KNXTestKit) -> None:
"""Test loading a button from storage."""
await knx.setup_integration(config_store_fixture="config_store_button.json")
# Raw button configuration
knx.assert_state(
"button.test_raw",
STATE_UNKNOWN,
)
await hass.services.async_call(
"button", "press", {"entity_id": "button.test_raw"}, blocking=True
)
await knx.assert_write("1/1/1", (1,))
# Typed button configuration
knx.assert_state(
"button.test_typed",
STATE_UNKNOWN,
)
await hass.services.async_call(
"button", "press", {"entity_id": "button.test_typed"}, blocking=True
)
await knx.assert_write("1/1/2", True)
@pytest.mark.parametrize(
"knx_config",
[
{ # missing data
"ga_send": {"write": "1/1/1", "dpt": "9.001"},
},
{ # missing DPT
"ga_send": {"write": "1/1/1"},
"data": {"value": 1},
},
{ # invalid value for DPT
"ga_send": {"write": "1/1/1", "dpt": "9.001"},
"data": {"value": "not_valid"},
},
{ # invalid length for DPT
"ga_send": {"write": "1/1/1", "dpt": "9.001"},
"data": {"payload": "0x1", "payload_length": 1},
},
{ # out of bound value for DPT
"ga_send": {"write": "1/1/1", "dpt": "5.001"},
"data": {"value": 101},
},
{ # out of bound value for length
"ga_send": {"write": "1/1/1"},
"data": {"payload": "0x100", "payload_length": 1},
},
{ # out of bound value for zero-length
"ga_send": {"write": "1/1/1"},
"data": {"payload": "0x40", "payload_length": 0},
},
],
)
async def test_button_ui_create_data_validation(
hass: HomeAssistant,
knx: KNXTestKit,
hass_ws_client: WebSocketGenerator,
knx_config: dict[str, Any],
) -> None:
"""Test creating a button with invalid data."""
await knx.setup_integration()
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "knx/create_entity",
"platform": Platform.BUTTON,
"data": {
"entity": {"name": "test"},
"knx": knx_config,
},
}
)
res = await client.receive_json()
assert res["success"], res
assert res["result"]["success"] is False
assert res["result"]["error_base"]
assert res["result"]["errors"][0]["path"]