Merge branch 'release/0.0.95'

setup.py

@@ -4,7 +4,7 @@
 
 from setuptools import find_packages, setup
 
-VERSION = "0.0.94"
+VERSION = "0.0.95"
 
 
 setup(

tests/test_konke.py

@@ -14,7 +14,7 @@
     OFF,
     ON,
     PRESS_TYPE,
-    ZONE_STATE,
+    ZONE_STATUS_CHANGE_COMMAND,
 )
 import zhaquirks.konke.motion
 
@@ -46,7 +46,7 @@ async def test_konke_motion(zigpy_device_from_quirk, quirk):
 
     assert len(motion_listener.cluster_commands) == 1
     assert len(motion_listener.attribute_updates) == 0
-    assert motion_listener.cluster_commands[0][1] == ZONE_STATE
+    assert motion_listener.cluster_commands[0][1] == ZONE_STATUS_CHANGE_COMMAND
     assert motion_listener.cluster_commands[0][2][0] == ON
 
     assert len(occupancy_listener.cluster_commands) == 0
@@ -57,7 +57,7 @@ async def test_konke_motion(zigpy_device_from_quirk, quirk):
     await asyncio.sleep(0.1)
 
     assert len(motion_listener.cluster_commands) == 2
-    assert motion_listener.cluster_commands[1][1] == ZONE_STATE
+    assert motion_listener.cluster_commands[1][1] == ZONE_STATUS_CHANGE_COMMAND
     assert motion_listener.cluster_commands[1][2][0] == OFF
 
     assert len(occupancy_listener.cluster_commands) == 0

tests/test_orvibo.py

@@ -6,7 +6,7 @@
 import pytest
 
 import zhaquirks
-from zhaquirks.const import OFF, ON, ZONE_STATE
+from zhaquirks.const import OFF, ON, ZONE_STATUS_CHANGE_COMMAND
 import zhaquirks.orvibo.motion
 
 from tests.common import ZCL_IAS_MOTION_COMMAND, ClusterListener
@@ -35,7 +35,7 @@ async def test_orvibo_motion(zigpy_device_from_quirk, quirk):
 
     assert len(motion_listener.cluster_commands) == 1
     assert len(motion_listener.attribute_updates) == 0
-    assert motion_listener.cluster_commands[0][1] == ZONE_STATE
+    assert motion_listener.cluster_commands[0][1] == ZONE_STATUS_CHANGE_COMMAND
     assert motion_listener.cluster_commands[0][2][0] == ON
 
     assert len(occupancy_listener.cluster_commands) == 0
@@ -46,7 +46,7 @@ async def test_orvibo_motion(zigpy_device_from_quirk, quirk):
     await asyncio.sleep(0.1)
 
     assert len(motion_listener.cluster_commands) == 2
-    assert motion_listener.cluster_commands[1][1] == ZONE_STATE
+    assert motion_listener.cluster_commands[1][1] == ZONE_STATUS_CHANGE_COMMAND
     assert motion_listener.cluster_commands[1][2][0] == OFF
 
     assert len(occupancy_listener.cluster_commands) == 0

tests/test_tuya.py

@@ -2,13 +2,25 @@
 
 import asyncio
 import datetime
+import itertools
 from unittest import mock
 
 import pytest
+import zigpy
 from zigpy.profiles import zha
 from zigpy.quirks import CustomDevice, get_device
 import zigpy.types as t
 from zigpy.zcl import foundation
+from zigpy.zcl.clusters import CLUSTERS_BY_ID
+from zigpy.zcl.clusters.general import (
+    Basic,
+    GreenPowerProxy,
+    Groups,
+    Identify,
+    Ota,
+    PowerConfiguration,
+)
+from zigpy.zcl.clusters.lightlink import LightLink
 
 import zhaquirks
 from zhaquirks.const import (
@@ -20,9 +32,17 @@
     ON,
     OUTPUT_CLUSTERS,
     PROFILE_ID,
-    ZONE_STATE,
+    ZONE_STATUS_CHANGE_COMMAND,
+)
+from zhaquirks.tuya import (
+    Data,
+    TuyaEnchantableCluster,
+    TuyaManufClusterAttributes,
+    TuyaNewManufCluster,
 )
-from zhaquirks.tuya import Data, TuyaManufClusterAttributes, TuyaNewManufCluster
+import zhaquirks.tuya.sm0202_motion
+import zhaquirks.tuya.ts011f_plug
+import zhaquirks.tuya.ts0041
 import zhaquirks.tuya.ts0042
 import zhaquirks.tuya.ts0043
 import zhaquirks.tuya.ts0501_fan_switch
@@ -100,14 +120,13 @@ async def test_motion(zigpy_device_from_quirk, quirk):
         tuya_cluster.handle_message(hdr, args)
 
     assert len(motion_listener.cluster_commands) == 1
-    assert len(motion_listener.attribute_updates) == 1
-    assert motion_listener.cluster_commands[0][1] == ZONE_STATE
+    assert motion_listener.cluster_commands[0][1] == ZONE_STATUS_CHANGE_COMMAND
     assert motion_listener.cluster_commands[0][2][0] == ON
 
     await asyncio.gather(asyncio.sleep(0), asyncio.sleep(0), asyncio.sleep(0))
 
     assert len(motion_listener.cluster_commands) == 2
-    assert motion_listener.cluster_commands[1][1] == ZONE_STATE
+    assert motion_listener.cluster_commands[1][1] == ZONE_STATUS_CHANGE_COMMAND
     assert motion_listener.cluster_commands[1][2][0] == OFF
 
 
@@ -1527,3 +1546,147 @@ async def test_fan_switch_writes_attributes(zigpy_device_from_quirk, quirk):
             1,
             b"\x00\x01\x02\x01\x000\x00",
         )
+
+
+async def test_sm0202_motion_sensor_signature(assert_signature_matches_quirk):
+    """Test LH992ZB motion sensor remote signature is matched to its quirk."""
+    signature = {
+        "node_descriptor": "NodeDescriptor(logical_type=<LogicalType.EndDevice: 2>, complex_descriptor_available=0, user_descriptor_available=0, reserved=0, aps_flags=0, frequency_band=<FrequencyBand.Freq2400MHz: 8>, mac_capability_flags=<MACCapabilityFlags.AllocateAddress: 128>, manufacturer_code=4098, maximum_buffer_size=82, maximum_incoming_transfer_size=82, server_mask=11264, maximum_outgoing_transfer_size=82, descriptor_capability_field=<DescriptorCapability.NONE: 0>, *allocate_address=True, *is_alternate_pan_coordinator=False, *is_coordinator=False, *is_end_device=True, *is_full_function_device=False, *is_mains_powered=False, *is_receiver_on_when_idle=False, *is_router=False, *is_security_capable=False)",
+        "endpoints": {
+            "1": {
+                "profile_id": 260,
+                "device_type": "0x0402",
+                "in_clusters": ["0x0000", "0x0001", "0x0003", "0x0500", "0xeeff"],
+                "out_clusters": ["0x0019"],
+            }
+        },
+        "manufacturer": "_TYZB01_z2umiwvq",
+        "model": "SM0202",
+        "class": "zhaquirks.tuya.lh992zb.TuyaMotionSM0202",
+    }
+    assert_signature_matches_quirk(zhaquirks.tuya.sm0202_motion.SM0202Motion, signature)
+
+
+@pytest.mark.parametrize(
+    "quirk",
+    (zhaquirks.tuya.ts0041.TuyaSmartRemote0041TOPlusA,),
+)
+async def test_power_config_no_bind(zigpy_device_from_quirk, quirk):
+    """Test that the power configuration cluster is not bound and no attribute reporting is set up."""
+
+    device = zigpy_device_from_quirk(quirk)
+    power_cluster = device.endpoints[1].power
+
+    request_patch = mock.patch("zigpy.zcl.Cluster.request", mock.AsyncMock())
+    bind_patch = mock.patch("zigpy.zcl.Cluster.bind", mock.AsyncMock())
+
+    with request_patch as request_mock, bind_patch as bind_mock:
+        request_mock.return_value = (foundation.Status.SUCCESS, "done")
+
+        await power_cluster.bind()
+        await power_cluster.configure_reporting(
+            PowerConfiguration.attributes_by_name["battery_percentage_remaining"].id,
+            3600,
+            10800,
+            1,
+        )
+
+        assert len(request_mock.mock_calls) == 0
+        assert len(bind_mock.mock_calls) == 0
+
+
+ENCHANTED_QUIRKS = []
+for manufacturer in zigpy.quirks._DEVICE_REGISTRY._registry.values():
+    for model_quirk_list in manufacturer.values():
+        for quirk_entry in model_quirk_list:
+            if quirk_entry in ENCHANTED_QUIRKS:
+                continue
+            # right now, this basically includes `issubclass(quirk, EnchantedDevice)`, as that sets `TUYA_SPELL`
+            if getattr(quirk_entry, "TUYA_SPELL", False):
+                ENCHANTED_QUIRKS.append(quirk_entry)
+
+del quirk_entry, model_quirk_list, manufacturer
+
+
+@mock.patch("zigpy.zcl.Cluster.bind", mock.AsyncMock())
+async def test_tuya_spell(zigpy_device_from_quirk):
+    """Test that enchanted Tuya devices have their spell applied when binding OnOff cluster."""
+    non_bindable_cluster_ids = [
+        Basic.cluster_id,
+        Identify.cluster_id,
+        Groups.cluster_id,
+        Ota.cluster_id,
+        GreenPowerProxy.cluster_id,
+        LightLink.cluster_id,
+    ]
+
+    request_patch = mock.patch("zigpy.zcl.Cluster.request", mock.AsyncMock())
+    with request_patch as request_mock:
+        request_mock.return_value = (foundation.Status.SUCCESS, "done")
+
+        for quirk in ENCHANTED_QUIRKS:
+            device = zigpy_device_from_quirk(quirk)
+
+            for cluster in itertools.chain(
+                device.endpoints[1].in_clusters.values(),
+                device.endpoints[1].out_clusters.values(),
+            ):
+
+                # emulate ZHA calling bind() on most default clusters with an unchanged ep_attribute
+                if (
+                    not isinstance(cluster, int)
+                    and cluster.cluster_id not in non_bindable_cluster_ids
+                    and cluster.cluster_id in CLUSTERS_BY_ID
+                    and CLUSTERS_BY_ID[cluster.cluster_id].ep_attribute
+                    == cluster.ep_attribute
+                ):
+                    await cluster.bind()
+
+            # check that exactly one Tuya spell was cast
+            if len(request_mock.mock_calls) == 0:
+                pytest.fail(
+                    f"Enchanted quirk {quirk} did not cast a Tuya spell. "
+                    f"One bindable cluster subclassing `TuyaEnchantableCluster` on endpoint 1 needs to be implemented. "
+                    f"Also check that enchanted bindable clusters do not modify their `ep_attribute`, "
+                    f"as ZHA will not call bind() in that case."
+                )
+            elif len(request_mock.mock_calls) > 1:
+                pytest.fail(
+                    f"Enchanted quirk {quirk} cast more than one Tuya spell. "
+                    f"Make sure to only implement one cluster subclassing `TuyaEnchantableCluster` on endpoint 1."
+                )
+
+            assert (
+                request_mock.mock_calls[0][1][1]
+                == foundation.GeneralCommand.Read_Attributes
+            )  # read attributes
+            assert request_mock.mock_calls[0][1][3] == [4, 0, 1, 5, 7, 65534]  # spell
+            request_mock.reset_mock()
+
+
+def test_tuya_spell_devices_valid():
+    """Test that all enchanted Tuya devices implement at least one enchanted cluster."""
+
+    for quirk in ENCHANTED_QUIRKS:
+        enchanted_clusters = 0
+
+        # iterate over all clusters in the replacement
+        for endpoint_id, endpoint in quirk.replacement[ENDPOINTS].items():
+            if endpoint_id != 1:  # spell is only activated on endpoint 1 for now
+                continue
+            for cluster in endpoint[INPUT_CLUSTERS] + endpoint[OUTPUT_CLUSTERS]:
+                if not isinstance(cluster, int) and issubclass(
+                    cluster, TuyaEnchantableCluster
+                ):
+                    enchanted_clusters += 1
+
+        # one EnchantedDevice must have exactly one enchanted cluster on endpoint 1
+        if enchanted_clusters == 0:
+            pytest.fail(
+                f"{quirk} does not have a cluster subclassing `TuyaEnchantableCluster` on endpoint 1 "
+                f"as required by the Tuya spell."
+            )
+        elif enchanted_clusters > 1:
+            pytest.fail(
+                f"{quirk} has more than one cluster subclassing `TuyaEnchantableCluster` on endpoint 1"
+            )

tests/test_tuya_air.py

@@ -49,6 +49,16 @@ def air_quality_device(zigpy_device_from_quirk):
             "temperature",
             2880,
         ),
+        (
+            b"\t\x02\x01\x00\x00\x12\x02\x00\x04\x00\x00\xff\xfb",
+            "temperature",
+            -50,
+        ),
+        (
+            b"\t\x02\x01\x00\x00\x12\x02\x00\x04\x00\x00\xff\xef",
+            "temperature",
+            -170,
+        ),
     ),
 )
 def test_co2_sensor(air_quality_device, data, ep_attr, expected_value):

tests/test_tuya_valve.py

@@ -12,7 +12,6 @@
 zhaquirks.setup()
 
 
-@mock.patch("zhaquirks.tuya.mcu.EnchantedDevice.spell", mock.AsyncMock())
 @pytest.mark.parametrize("quirk", (zhaquirks.tuya.ts0601_valve.ParksidePSBZS,))
 async def test_command_psbzs(zigpy_device_from_quirk, quirk):
     """Test executing cluster commands for PARKSIDE water valve."""
@@ -41,7 +40,6 @@ async def test_command_psbzs(zigpy_device_from_quirk, quirk):
         assert rsp.status == foundation.Status.SUCCESS
 
 
-@mock.patch("zhaquirks.tuya.mcu.EnchantedDevice.spell", mock.AsyncMock())
 @pytest.mark.parametrize("quirk", (zhaquirks.tuya.ts0601_valve.ParksidePSBZS,))
 async def test_write_attr_psbzs(zigpy_device_from_quirk, quirk):
     """Test write cluster attributes for PARKSIDE water valve."""