starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-24 07:25:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Defensively validate ZHA quirks v2 supplied entity metadata (#112643)

Browse Source
This commit is contained in:
David F. Mulcahey
2024-03-27 17:48:43 +01:00
committed by GitHub
parent 65230908c6
commit c518acfef3
17 changed files with 602 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/components/zha/test_discover.py
+303 -12
View File
@@ -1,6 +1,8 @@
"""Test ZHA device discovery."""
from collections.abc import Callable
import enum
import itertools
import re
from typing import Any
from unittest import mock
@@ -20,7 +22,16 @@ from zhaquirks.xiaomi.aqara.driver_curtain_e1 import (
from zigpy.const import SIG_ENDPOINTS, SIG_MANUFACTURER, SIG_MODEL, SIG_NODE_DESC
import zigpy.profiles.zha
import zigpy.quirks
from zigpy.quirks.v2 import EntityType, add_to_registry_v2
from zigpy.quirks.v2 import (
BinarySensorMetadata,
EntityMetadata,
EntityType,
NumberMetadata,
QuirksV2RegistryEntry,
ZCLCommandButtonMetadata,
ZCLSensorMetadata,
add_to_registry_v2,
)
from zigpy.quirks.v2.homeassistant import UnitOfTime
import zigpy.types
from zigpy.zcl import ClusterType
@@ -40,6 +51,7 @@ from homeassistant.const import STATE_OFF, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_platform import EntityPlatform
from homeassistant.util.json import load_json
from .common import find_entity_id, update_attribute_cache
from .conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE
@@ -520,6 +532,7 @@ async def test_quirks_v2_entity_discovery(
step=1,
unit=UnitOfTime.SECONDS,
multiplier=1,
translation_key="on_off_transition_time",
)
)
@@ -618,7 +631,11 @@ async def test_quirks_v2_entity_discovery_e1_curtain(
entity_platform=Platform.SENSOR,
entity_type=EntityType.DIAGNOSTIC,
)
.binary_sensor("error_detected", FakeXiaomiAqaraDriverE1.cluster_id)
.binary_sensor(
"error_detected",
FakeXiaomiAqaraDriverE1.cluster_id,
translation_key="valve_alarm",
)
)
aqara_E1_device = zigpy.quirks._DEVICE_REGISTRY.get_device(aqara_E1_device)
@@ -683,7 +700,13 @@ async def test_quirks_v2_entity_discovery_e1_curtain(
assert state.state == STATE_OFF
def _get_test_device(zigpy_device_mock, manufacturer: str, model: str):
def _get_test_device(
zigpy_device_mock,
manufacturer: str,
model: str,
augment_method: Callable[[QuirksV2RegistryEntry], QuirksV2RegistryEntry]
| None = None,
):
zigpy_device = zigpy_device_mock(
{
1: {
@@ -703,7 +726,7 @@ def _get_test_device(zigpy_device_mock, manufacturer: str, model: str):
model=model,
)
(
v2_quirk = (
add_to_registry_v2(manufacturer, model, zigpy.quirks._DEVICE_REGISTRY)
.replaces(PowerConfig1CRCluster)
.replaces(ScenesCluster, cluster_type=ClusterType.Client)
@@ -716,6 +739,7 @@ def _get_test_device(zigpy_device_mock, manufacturer: str, model: str):
step=1,
unit=UnitOfTime.SECONDS,
multiplier=1,
translation_key="on_off_transition_time",
)
.number(
zigpy.zcl.clusters.general.OnOff.AttributeDefs.off_wait_time.name,
@@ -725,14 +749,19 @@ def _get_test_device(zigpy_device_mock, manufacturer: str, model: str):
step=1,
unit=UnitOfTime.SECONDS,
multiplier=1,
translation_key="on_off_transition_time",
)
.sensor(
zigpy.zcl.clusters.general.OnOff.AttributeDefs.off_wait_time.name,
zigpy.zcl.clusters.general.OnOff.cluster_id,
entity_type=EntityType.CONFIG,
translation_key="analog_input",
)
)
if augment_method:
v2_quirk = augment_method(v2_quirk)
zigpy_device = zigpy.quirks._DEVICE_REGISTRY.get_device(zigpy_device)
zigpy_device.endpoints[1].power.PLUGGED_ATTR_READS = {
"battery_voltage": 3,
@@ -792,14 +821,13 @@ async def test_quirks_v2_entity_discovery_errors(
# fmt: off
entity_details = (
"{'cluster_details': (1, 6, <ClusterType.Server: 0>), "
"'quirk_metadata': EntityMetadata(entity_metadata=ZCLSensorMetadata("
"attribute_name='off_wait_time', divisor=1, multiplier=1, unit=None, "
"device_class=None, state_class=None), entity_platform=<EntityPlatform."
"SENSOR: 'sensor'>, entity_type=<EntityType.CONFIG: 'config'>, "
"cluster_id=6, endpoint_id=1, cluster_type=<ClusterType.Server: 0>, "
"initially_disabled=False, attribute_initialized_from_cache=True, "
"translation_key=None)}"
"{'cluster_details': (1, 6, <ClusterType.Server: 0>), 'entity_metadata': "
"ZCLSensorMetadata(entity_platform=<EntityPlatform.SENSOR: 'sensor'>, "
"entity_type=<EntityType.CONFIG: 'config'>, cluster_id=6, endpoint_id=1, "
"cluster_type=<ClusterType.Server: 0>, initially_disabled=False, "
"attribute_initialized_from_cache=True, translation_key='analog_input', "
"attribute_name='off_wait_time', divisor=1, multiplier=1, "
"unit=None, device_class=None, state_class=None)}"
)
# fmt: on
@@ -807,3 +835,266 @@ async def test_quirks_v2_entity_discovery_errors(
m2 = f"details: {entity_details} that does not have an entity class mapping - "
m3 = "unable to create entity"
assert f"{m1}{m2}{m3}" in caplog.text
DEVICE_CLASS_TYPES = [NumberMetadata, BinarySensorMetadata, ZCLSensorMetadata]
def validate_device_class_unit(
quirk: QuirksV2RegistryEntry,
entity_metadata: EntityMetadata,
platform: Platform,
translations: dict,
) -> None:
"""Ensure device class and unit are used correctly."""
if (
hasattr(entity_metadata, "unit")
and entity_metadata.unit is not None
and hasattr(entity_metadata, "device_class")
and entity_metadata.device_class is not None
):
m1 = "device_class and unit are both set - unit: "
m2 = f"{entity_metadata.unit} device_class: "
m3 = f"{entity_metadata.device_class} for {platform.name} "
raise ValueError(f"{m1}{m2}{m3}{quirk}")
def validate_translation_keys(
quirk: QuirksV2RegistryEntry,
entity_metadata: EntityMetadata,
platform: Platform,
translations: dict,
) -> None:
"""Ensure translation keys exist for all v2 quirks."""
if isinstance(entity_metadata, ZCLCommandButtonMetadata):
default_translation_key = entity_metadata.command_name
else:
default_translation_key = entity_metadata.attribute_name
translation_key = entity_metadata.translation_key or default_translation_key
if (
translation_key is not None
and translation_key not in translations["entity"][platform]
):
raise ValueError(
f"Missing translation key: {translation_key} for {platform.name} {quirk}"
)
def validate_translation_keys_device_class(
quirk: QuirksV2RegistryEntry,
entity_metadata: EntityMetadata,
platform: Platform,
translations: dict,
) -> None:
"""Validate translation keys and device class usage."""
if isinstance(entity_metadata, ZCLCommandButtonMetadata):
default_translation_key = entity_metadata.command_name
else:
default_translation_key = entity_metadata.attribute_name
translation_key = entity_metadata.translation_key or default_translation_key
metadata_type = type(entity_metadata)
if metadata_type in DEVICE_CLASS_TYPES:
device_class = entity_metadata.device_class
if device_class is not None and translation_key is not None:
m1 = "translation_key and device_class are both set - translation_key: "
m2 = f"{translation_key} device_class: {device_class} for {platform.name} "
raise ValueError(f"{m1}{m2}{quirk}")
def validate_metadata(validator: Callable) -> None:
"""Ensure v2 quirks metadata does not violate HA rules."""
all_v2_quirks = itertools.chain.from_iterable(
zigpy.quirks._DEVICE_REGISTRY._registry_v2.values()
)
translations = load_json("homeassistant/components/zha/strings.json")
for quirk in all_v2_quirks:
for entity_metadata in quirk.entity_metadata:
platform = Platform(entity_metadata.entity_platform.value)
validator(quirk, entity_metadata, platform, translations)
def bad_translation_key(v2_quirk: QuirksV2RegistryEntry) -> QuirksV2RegistryEntry:
"""Introduce a bad translation key."""
return v2_quirk.sensor(
zigpy.zcl.clusters.general.OnOff.AttributeDefs.off_wait_time.name,
zigpy.zcl.clusters.general.OnOff.cluster_id,
entity_type=EntityType.CONFIG,
translation_key="missing_translation_key",
)
def bad_device_class_unit_combination(
v2_quirk: QuirksV2RegistryEntry,
) -> QuirksV2RegistryEntry:
"""Introduce a bad device class and unit combination."""
return v2_quirk.sensor(
zigpy.zcl.clusters.general.OnOff.AttributeDefs.off_wait_time.name,
zigpy.zcl.clusters.general.OnOff.cluster_id,
entity_type=EntityType.CONFIG,
unit="invalid",
device_class="invalid",
translation_key="analog_input",
)
def bad_device_class_translation_key_usage(
v2_quirk: QuirksV2RegistryEntry,
) -> QuirksV2RegistryEntry:
"""Introduce a bad device class and translation key combination."""
return v2_quirk.sensor(
zigpy.zcl.clusters.general.OnOff.AttributeDefs.off_wait_time.name,
zigpy.zcl.clusters.general.OnOff.cluster_id,
entity_type=EntityType.CONFIG,
translation_key="invalid",
device_class="invalid",
)
@pytest.mark.parametrize(
("augment_method", "validate_method", "expected_exception_string"),
[
(
bad_translation_key,
validate_translation_keys,
"Missing translation key: missing_translation_key",
),
(
bad_device_class_unit_combination,
validate_device_class_unit,
"cannot have both unit and device_class",
),
(
bad_device_class_translation_key_usage,
validate_translation_keys_device_class,
"cannot have both a translation_key and a device_class",
),
],
)
async def test_quirks_v2_metadata_errors(
hass: HomeAssistant,
zigpy_device_mock,
zha_device_joined,
augment_method: Callable[[QuirksV2RegistryEntry], QuirksV2RegistryEntry],
validate_method: Callable,
expected_exception_string: str,
) -> None:
"""Ensure all v2 quirks translation keys exist."""
# no error yet
validate_metadata(validate_method)
# ensure the error is caught and raised
with pytest.raises(ValueError, match=expected_exception_string):
try:
# introduce an error
zigpy_device = _get_test_device(
zigpy_device_mock,
"Ikea of Sweden4",
"TRADFRI remote control4",
augment_method=augment_method,
)
await zha_device_joined(zigpy_device)
validate_metadata(validate_method)
# if the device was created we remove it
# so we don't pollute the rest of the tests
zigpy.quirks._DEVICE_REGISTRY.remove(zigpy_device)
except ValueError as e:
# if the device was not created we remove it
# so we don't pollute the rest of the tests
zigpy.quirks._DEVICE_REGISTRY._registry_v2.pop(
(
"Ikea of Sweden4",
"TRADFRI remote control4",
)
)
raise e
class BadDeviceClass(enum.Enum):
"""Bad device class."""
BAD = "bad"
def bad_binary_sensor_device_class(
v2_quirk: QuirksV2RegistryEntry,
) -> QuirksV2RegistryEntry:
"""Introduce a bad device class on a binary sensor."""
return v2_quirk.binary_sensor(
zigpy.zcl.clusters.general.OnOff.AttributeDefs.on_off.name,
zigpy.zcl.clusters.general.OnOff.cluster_id,
device_class=BadDeviceClass.BAD,
)
def bad_sensor_device_class(
v2_quirk: QuirksV2RegistryEntry,
) -> QuirksV2RegistryEntry:
"""Introduce a bad device class on a sensor."""
return v2_quirk.sensor(
zigpy.zcl.clusters.general.OnOff.AttributeDefs.off_wait_time.name,
zigpy.zcl.clusters.general.OnOff.cluster_id,
device_class=BadDeviceClass.BAD,
)
def bad_number_device_class(
v2_quirk: QuirksV2RegistryEntry,
) -> QuirksV2RegistryEntry:
"""Introduce a bad device class on a number."""
return v2_quirk.number(
zigpy.zcl.clusters.general.OnOff.AttributeDefs.on_time.name,
zigpy.zcl.clusters.general.OnOff.cluster_id,
device_class=BadDeviceClass.BAD,
)
ERROR_ROOT = "Quirks provided an invalid device class"
@pytest.mark.parametrize(
("augment_method", "expected_exception_string"),
[
(
bad_binary_sensor_device_class,
f"{ERROR_ROOT}: BadDeviceClass.BAD for platform binary_sensor",
),
(
bad_sensor_device_class,
f"{ERROR_ROOT}: BadDeviceClass.BAD for platform sensor",
),
(
bad_number_device_class,
f"{ERROR_ROOT}: BadDeviceClass.BAD for platform number",
),
],
)
async def test_quirks_v2_metadata_bad_device_classes(
hass: HomeAssistant,
zigpy_device_mock,
zha_device_joined,
caplog: pytest.LogCaptureFixture,
augment_method: Callable[[QuirksV2RegistryEntry], QuirksV2RegistryEntry],
expected_exception_string: str,
) -> None:
"""Test bad quirks v2 device classes."""
# introduce an error
zigpy_device = _get_test_device(
zigpy_device_mock,
"Ikea of Sweden4",
"TRADFRI remote control4",
augment_method=augment_method,
)
await zha_device_joined(zigpy_device)
assert expected_exception_string in caplog.text
# remove the device so we don't pollute the rest of the tests
zigpy.quirks._DEVICE_REGISTRY.remove(zigpy_device)
tests/components/zha/test_helpers.py
+27 -2
View File
@@ -1,11 +1,13 @@
"""Tests for ZHA helpers."""
import enum
import logging
from unittest.mock import patch
import pytest
import voluptuous_serialize
import zigpy.profiles.zha as zha
from zigpy.quirks.v2.homeassistant import UnitOfPower as QuirksUnitOfPower
from zigpy.types.basic import uint16_t
import zigpy.zcl.clusters.general as general
import zigpy.zcl.clusters.lighting as lighting
@@ -13,8 +15,9 @@ import zigpy.zcl.clusters.lighting as lighting
from homeassistant.components.zha.core.helpers import (
cluster_command_schema_to_vol_schema,
convert_to_zcl_values,
validate_unit,
)
from homeassistant.const import Platform
from homeassistant.const import Platform, UnitOfPower
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
@@ -40,7 +43,7 @@ def light_platform_only():
@pytest.fixture
async def device_light(hass, zigpy_device_mock, zha_device_joined):
async def device_light(hass: HomeAssistant, zigpy_device_mock, zha_device_joined):
"""Test light."""
zigpy_device = zigpy_device_mock(
@@ -211,3 +214,25 @@ async def test_zcl_schema_conversions(hass: HomeAssistant, device_light) -> None
# No flags are passed through
assert converted_data["update_flags"] == 0
def test_unit_validation() -> None:
"""Test unit validation."""
assert validate_unit(QuirksUnitOfPower.WATT) == UnitOfPower.WATT
class FooUnit(enum.Enum):
"""Foo unit."""
BAR = "bar"
class UnitOfMass(enum.Enum):
"""UnitOfMass."""
BAR = "bar"
with pytest.raises(KeyError):
validate_unit(FooUnit.BAR)
with pytest.raises(ValueError):
validate_unit(UnitOfMass.BAR)
tests/components/zha/test_select.py
+4 -5
View File
@@ -435,7 +435,7 @@ async def test_on_off_select_attribute_report(
"motion_sensitivity_disabled",
AqaraMotionSensitivities,
MotionSensitivityQuirk.OppleCluster.cluster_id,
translation_key="motion_sensitivity_translation_key",
translation_key="motion_sensitivity",
initially_disabled=True,
)
)
@@ -491,9 +491,8 @@ async def test_on_off_select_attribute_report_v2(
assert hass.states.get(entity_id).state == AqaraMotionSensitivities.Low.name
entity_registry = er.async_get(hass)
# none in id because the translation key does not exist
entity_entry = entity_registry.async_get("select.fake_manufacturer_fake_model_none")
entity_entry = entity_registry.async_get(entity_id)
assert entity_entry
assert entity_entry.entity_category == EntityCategory.CONFIG
assert entity_entry.disabled is True
assert entity_entry.translation_key == "motion_sensitivity_translation_key"
assert entity_entry.disabled is False
assert entity_entry.translation_key == "motion_sensitivity"
tests/components/zha/test_sensor.py
+2 -2
View File
@@ -1260,10 +1260,10 @@ async def test_last_feeding_size_sensor_v2(
assert entity_id is not None
await send_attributes_report(hass, cluster, {0x010C: 1})
assert_state(hass, entity_id, "1.0", UnitOfMass.GRAMS)
assert_state(hass, entity_id, "1.0", UnitOfMass.GRAMS.value)
await send_attributes_report(hass, cluster, {0x010C: 5})
assert_state(hass, entity_id, "5.0", UnitOfMass.GRAMS)
assert_state(hass, entity_id, "5.0", UnitOfMass.GRAMS.value)
@pytest.fixture