Add Tuya v2 quirk builder

requirements_test.txt

@@ -13,5 +13,5 @@ pytest-sugar
 pytest-timeout
 pytest-asyncio
 pytest>=7.1.3
-zigpy>=0.68.1
+zigpy>=0.69
 ruff==0.0.261

tests/async_mock.py

@@ -0,0 +1,43 @@
+"""Mock utilities that are async aware."""
+
+from unittest.mock import *  # noqa: F401, F403
+
+
+class _IntSentinelObject(int):
+    """Sentinel-like object that is also an integer subclass.
+
+    Allows sentinels to be used
+    in loggers that perform int-specific string formatting.
+    """
+
+    def __new__(cls, name):
+        instance = super().__new__(cls, 0)
+        instance.name = name
+        return instance
+
+    def __repr__(self):
+        return f"int_sentinel.{self.name}"
+
+    def __hash__(self):
+        return hash((int(self), self.name))
+
+    def __eq__(self, other):
+        return self is other
+
+    __str__ = __reduce__ = __repr__
+
+
+class _IntSentinel:
+    def __init__(self):
+        self._sentinels = {}
+
+    def __getattr__(self, name):
+        if name == "__bases__":
+            raise AttributeError
+        return self._sentinels.setdefault(name, _IntSentinelObject(name))
+
+    def __reduce__(self):
+        return "int_sentinel"
+
+
+int_sentinel = _IntSentinel()

tests/test_tuya_builder.py

@@ -0,0 +1,163 @@
+"""Tests for TuyaQuirkBuilder."""
+
+from unittest import mock
+
+import pytest
+from zigpy.device import Device
+from zigpy.quirks.registry import DeviceRegistry
+from zigpy.quirks.v2 import CustomDeviceV2
+import zigpy.types as t
+from zigpy.zcl import foundation
+from zigpy.zcl.clusters.general import Basic
+
+from tests.common import ClusterListener, wait_for_zigpy_tasks
+import zhaquirks
+from zhaquirks.tuya.builder import (
+    TuyaPowerConfigurationCluster2AAA,
+    TuyaQuirkBuilder,
+    TuyaRelativeHumidity,
+    TuyaSoilMoisture,
+    TuyaTemperatureMeasurement,
+    TuyaValveWaterConsumed,
+)
+from zhaquirks.tuya.mcu import TuyaMCUCluster, TuyaOnOffNM
+
+from .async_mock import sentinel
+
+zhaquirks.setup()
+
+
+@pytest.fixture(name="device_mock")
+def real_device(MockAppController):
+    """Device fixture with a single endpoint."""
+    ieee = sentinel.ieee
+    nwk = 0x2233
+    device = Device(MockAppController, ieee, nwk)
+
+    device.add_endpoint(1)
+    device[1].profile_id = 0x0104
+    device[1].device_type = 0x0051
+    device.model = "model"
+    device.manufacturer = "manufacturer"
+    device[1].add_input_cluster(0x0000)
+    device[1].add_input_cluster(0xEF00)
+    device[1].add_output_cluster(0x000A)
+    device[1].add_output_cluster(0x0019)
+    return device
+
+
+async def test_tuya_quirkbuilder(device_mock):
+    """Test adding a v2 Tuya Quirk to the registry and getting back a quirked device."""
+
+    registry = DeviceRegistry()
+
+    class TestEnum(t.enum8):
+        """Test Enum."""
+
+        A = 0x00
+        B = 0x01
+
+    entry = (
+        TuyaQuirkBuilder(device_mock.manufacturer, device_mock.model, registry=registry)
+        .tuya_battery(dp_id=1)
+        .tuya_metering(dp_id=2)
+        .tuya_onoff(dp_id=3)
+        .tuya_soil_moisture(dp_id=4)
+        .tuya_temperature(dp_id=5)
+        .tuya_switch(
+            dp_id=6,
+            attribute_name="test_onoff",
+        )
+        .tuya_number(
+            dp_id=7,
+            attribute_name="test_number",
+            type=t.uint16_t,
+        )
+        .tuya_binary_sensor(
+            dp_id=8,
+            attribute_name="test_binary",
+        )
+        .tuya_sensor(
+            dp_id=9,
+            attribute_name="test_sensor",
+            type=t.uint8_t,
+        )
+        .tuya_enum(
+            dp_id=10,
+            attribute_name="test_enum",
+            enum_class=TestEnum,
+        )
+        .tuya_humidity(dp_id=11)
+        .add_to_registry()
+    )
+
+    # coverage for overridden __eq__ method
+    assert entry.adds_metadata[0] != entry.adds_metadata[1]
+    assert entry.adds_metadata[0] != entry
+
+    quirked = registry.get_device(device_mock)
+    assert isinstance(quirked, CustomDeviceV2)
+    assert quirked in registry
+
+    ep = quirked.endpoints[1]
+
+    assert ep.basic is not None
+    assert isinstance(ep.basic, Basic)
+
+    assert ep.tuya_manufacturer is not None
+    assert isinstance(ep.tuya_manufacturer, TuyaMCUCluster)
+
+    tuya_cluster = ep.tuya_manufacturer
+    tuya_listener = ClusterListener(tuya_cluster)
+    assert tuya_cluster.attributes_by_name["mcu_version"].id == 0xEF00
+    assert tuya_cluster.attributes_by_name["test_onoff"].id == 0xEF06
+    assert tuya_cluster.attributes_by_name["test_number"].id == 0xEF07
+    assert tuya_cluster.attributes_by_name["test_binary"].id == 0xEF08
+    assert tuya_cluster.attributes_by_name["test_sensor"].id == 0xEF09
+    assert tuya_cluster.attributes_by_name["test_enum"].id == 0xEF0A
+
+    assert ep.power is not None
+    assert isinstance(ep.power, TuyaPowerConfigurationCluster2AAA)
+
+    assert ep.smartenergy_metering is not None
+    assert isinstance(ep.smartenergy_metering, TuyaValveWaterConsumed)
+
+    assert ep.on_off is not None
+    assert isinstance(ep.on_off, TuyaOnOffNM)
+
+    assert ep.soil_moisture is not None
+    assert isinstance(ep.soil_moisture, TuyaSoilMoisture)
+
+    assert ep.temperature is not None
+    assert isinstance(ep.temperature, TuyaTemperatureMeasurement)
+
+    assert ep.humidity is not None
+    assert isinstance(ep.humidity, TuyaRelativeHumidity)
+
+    with mock.patch.object(
+        tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS
+    ) as m1:
+        (status,) = await tuya_cluster.write_attributes(
+            {
+                "test_enum": 0x01,
+            }
+        )
+
+        await wait_for_zigpy_tasks()
+        m1.assert_called_with(
+            cluster=61184,
+            sequence=1,
+            data=b"\x01\x01\x00\x00\x01\n\x04\x00\x01\x01",
+            command_id=0,
+            timeout=5,
+            expect_reply=False,
+            use_ieee=False,
+            ask_for_ack=None,
+            priority=t.PacketPriority.NORMAL,
+        )
+        assert status == [
+            foundation.WriteAttributesStatusRecord(foundation.Status.SUCCESS)
+        ]
+
+    assert tuya_listener.attribute_updates[0][0] == 0xEF0A
+    assert tuya_listener.attribute_updates[0][1] == TestEnum.B

zhaquirks/tuya/__init__.py

@@ -9,7 +9,7 @@
 
 from zigpy.quirks import CustomCluster, CustomDevice
 import zigpy.types as t
-from zigpy.zcl import foundation
+from zigpy.zcl import BaseAttributeDefs, foundation
 from zigpy.zcl.clusters.closures import WindowCovering
 from zigpy.zcl.clusters.general import Basic, LevelControl, OnOff, PowerConfiguration
 from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement
@@ -1507,6 +1507,9 @@ class TuyaNewManufCluster(CustomCluster):
     cluster_id: t.uint16_t = TUYA_CLUSTER_ID
     ep_attribute: str = "tuya_manufacturer"
 
+    class AttributeDefs(BaseAttributeDefs):
+        """Attribute Definitions."""
+
     server_commands = {
         TUYA_QUERY_DATA: foundation.ZCLCommandDef(
             "query_data", {}, False, is_manufacturer_specific=True