Merge 83ef4df into 2c4d567

tests/common.py

@@ -1,5 +1,5 @@
 """Quirks common helpers."""
-
+import datetime
 
 ZCL_IAS_MOTION_COMMAND = b"\t!\x00\x01\x00\x00\x00\x00\x00"
 ZCL_OCC_ATTR_RPT_OCC = b"\x18d\n\x00\x00\x18\x01"
@@ -21,3 +21,19 @@ def attribute_updated(self, attr_id, value):
     def cluster_command(self, tsn, commdand_id, args):
         """Command received listener."""
         self.cluster_commands.append((tsn, commdand_id, args))
+
+
+class MockDatetime(datetime.datetime):
+    """Override for datetime functions."""
+
+    @classmethod
+    def now(cls):
+        """Return testvalue."""
+
+        return cls(1970, 1, 1, 1, 0, 0)
+
+    @classmethod
+    def utcnow(cls):
+        """Return testvalue."""
+
+        return cls(1970, 1, 1, 2, 0, 0)

tests/test_tuya.py

@@ -31,7 +31,7 @@
 import zhaquirks.tuya.ts0601_trv
 import zhaquirks.tuya.ts0601_valve
 
-from tests.common import ClusterListener
+from tests.common import ClusterListener, MockDatetime
 
 zhaquirks.setup()
 
@@ -82,22 +82,6 @@
 ZCL_TUYA_EHEAT_TARGET_TEMP = b"\t3\x01\x03\x05\x10\x02\x00\x04\x00\x00\x00\x15"
 
 
-class NewDatetime(datetime.datetime):
-    """Override for datetime functions."""
-
-    @classmethod
-    def now(cls):
-        """Return testvalue."""
-
-        return cls(1970, 1, 1, 1, 0, 0)
-
-    @classmethod
-    def utcnow(cls):
-        """Return testvalue."""
-
-        return cls(1970, 1, 1, 2, 0, 0)
-
-
 @pytest.mark.parametrize("quirk", (zhaquirks.tuya.ts0601_motion.TuyaMotion,))
 async def test_motion(zigpy_device_from_quirk, quirk):
     """Test tuya motion sensor."""
@@ -1233,7 +1217,7 @@ async def async_success(*args, **kwargs):
         assert status == foundation.Status.UNSUP_CLUSTER_COMMAND
 
         origdatetime = datetime.datetime
-        datetime.datetime = NewDatetime
+        datetime.datetime = MockDatetime
 
         hdr, args = tuya_cluster.deserialize(ZCL_TUYA_SET_TIME_REQUEST)
         tuya_cluster.handle_message(hdr, args)

tests/test_tuya_mcu.py

@@ -1,25 +1,28 @@
 """Tests for Tuya quirks."""
 
+import datetime
 from unittest import mock
 
 import pytest
 import zigpy.types as t
 from zigpy.zcl import foundation
 
 import zhaquirks
-from zhaquirks.tuya import TUYA_MCU_VERSION_RSP
+from zhaquirks.tuya import TUYA_MCU_VERSION_RSP, TUYA_SET_TIME
 from zhaquirks.tuya.mcu import (
     ATTR_MCU_VERSION,
+    TuyaAttributesCluster,
     TuyaClusterData,
     TuyaDPType,
     TuyaMCUCluster,
 )
 
-from tests.common import ClusterListener
+from tests.common import ClusterListener, MockDatetime
 
 zhaquirks.setup()
 
 ZCL_TUYA_VERSION_RSP = b"\x09\x06\x11\x01\x6D\x82"
+ZCL_TUYA_SET_TIME = b"\x09\x12\x24\x0D\x00"
 
 
 @pytest.mark.parametrize(
@@ -44,11 +47,58 @@ async def test_tuya_version(zigpy_device_from_quirk, quirk):
     assert cluster_listener.attribute_updates[0][0] == ATTR_MCU_VERSION
     assert cluster_listener.attribute_updates[0][1] == "2.0.2"
 
+    with mock.patch.object(tuya_cluster, "handle_mcu_version_response") as m1:
+        tuya_cluster.handle_message(hdr, args)
+
+        assert len(cluster_listener.cluster_commands) == 2
+        assert cluster_listener.cluster_commands[1][1] == TUYA_MCU_VERSION_RSP
+        assert cluster_listener.cluster_commands[1][2].version.version_raw == 130
+        assert cluster_listener.cluster_commands[1][2].version.version == "2.0.2"
+
+        m1.assert_called_once_with(
+            tuya_cluster.MCUVersion(status=1, tsn=109, version_raw=130)
+        )
+
     # read 'mcu_version' from cluster's attributes
     succ, fail = await tuya_cluster.read_attributes(("mcu_version",))
     assert succ["mcu_version"] == "2.0.2"
 
 
+@pytest.mark.parametrize(
+    "quirk", (zhaquirks.tuya.ts0601_dimmer.TuyaDoubleSwitchDimmer,)
+)
+async def test_tuya_mcu_set_time(zigpy_device_from_quirk, quirk):
+    """Test set_time requests (0x24) messages for MCU devices."""
+
+    tuya_device = zigpy_device_from_quirk(quirk)
+
+    tuya_cluster = tuya_device.endpoints[1].tuya_manufacturer
+    cluster_listener = ClusterListener(tuya_cluster)
+
+    # Mock datetime
+    origdatetime = datetime.datetime
+    datetime.datetime = MockDatetime
+
+    # simulate a SET_TIME message
+    hdr, args = tuya_cluster.deserialize(ZCL_TUYA_SET_TIME)
+    assert hdr.command_id == TUYA_SET_TIME
+
+    with mock.patch.object(
+        TuyaAttributesCluster, "command"
+    ) as m1:  # tuya_cluster parent class (because of super() call)
+        tuya_cluster.handle_message(hdr, args)
+
+        assert len(cluster_listener.cluster_commands) == 1
+        assert cluster_listener.cluster_commands[0][1] == TUYA_SET_TIME
+
+        m1.assert_called_once_with(
+            TUYA_SET_TIME, [0, 0, 28, 32, 0, 0, 14, 16], expect_reply=False
+        )
+
+    # restore datetime
+    datetime.datetime = origdatetime  # restore datetime
+
+
 @pytest.mark.parametrize(
     "quirk", (zhaquirks.tuya.ts0601_dimmer.TuyaDoubleSwitchDimmer,)
 )
@@ -66,6 +116,12 @@ async def test_tuya_methods(zigpy_device_from_quirk, quirk):
     tcd_switch1_on = TuyaClusterData(
         endpoint_id=1, cluster_attr="on_off", attr_value=1, expect_reply=True
     )
+    tcd_dimmer2_on = TuyaClusterData(
+        endpoint_id=2, cluster_attr="on_off", attr_value=1, expect_reply=True
+    )
+    tcd_dimmer2_level = TuyaClusterData(
+        endpoint_id=2, cluster_attr="current_level", attr_value=75, expect_reply=True
+    )
 
     result_1 = tuya_cluster.from_cluster_data(tcd_1)
     assert result_1
@@ -92,10 +148,19 @@ async def test_tuya_methods(zigpy_device_from_quirk, quirk):
 
         m1.assert_called_once_with(tcd_switch1_on)
         assert rsp.status == foundation.Status.SUCCESS
+        assert m1.call_count == 1
 
         rsp = await switch1_cluster.command(0x0004)
         m1.assert_called_once_with(tcd_switch1_on)  # no extra calls
         assert rsp.status == foundation.Status.UNSUP_CLUSTER_COMMAND
+        assert m1.call_count == 1
+
+        # test `move_to_level_with_on_off` quirk (call on_off + current_level)
+        rsp = await dimmer2_cluster.command(0x0004, 75, 1)
+        assert rsp.status == foundation.Status.SUCCESS
+        m1.assert_any_call(tcd_dimmer2_on)  # on_off
+        m1.assert_called_with(tcd_dimmer2_level)  # current_level
+        assert m1.call_count == 3
 
 
 async def test_tuya_mcu_classes():

zhaquirks/tuya/mcu/__init__.py

@@ -1,6 +1,7 @@
 """Tuya MCU comunications."""
 import asyncio
 import dataclasses
+import datetime
 from typing import Any, Callable, Dict, Optional, Tuple, Union
 
 from zigpy.quirks import CustomDevice
@@ -13,13 +14,15 @@
     TUYA_MCU_COMMAND,
     TUYA_MCU_VERSION_RSP,
     TUYA_SET_DATA,
+    TUYA_SET_TIME,
     Data,
     NoManufacturerCluster,
     PowerOnState,
     TuyaCommand,
     TuyaData,
     TuyaLocalCluster,
     TuyaNewManufCluster,
+    TuyaTimePayload,
 )
 
 # New manufacturer attributes
@@ -135,6 +138,9 @@ async def write_attributes(self, attributes, manufacturer=None):
 class TuyaMCUCluster(TuyaAttributesCluster, TuyaNewManufCluster):
     """Manufacturer specific cluster for sending Tuya MCU commands."""
 
+    set_time_offset = 1970  # MCU timestamp from 1/1/1970
+    set_time_local_offset = None
+
     class MCUVersion(t.Struct):
         """Tuya MCU version response Zcl payload."""
 
@@ -265,6 +271,31 @@ def handle_mcu_version_response(self, payload: MCUVersion) -> foundation.Status:
         self.update_attribute("mcu_version", payload.version)
         return foundation.Status.SUCCESS
 
+    def handle_set_time_request(self, payload: t.uint16_t) -> foundation.Status:
+        """Handle set_time requests (0x24)."""
+
+        payload_rsp = TuyaTimePayload()
+
+        utc_now = datetime.datetime.utcnow()
+        now = datetime.datetime.now()
+
+        offset_time = datetime.datetime(self.set_time_offset, 1, 1)
+        offset_time_local = datetime.datetime(
+            self.set_time_local_offset or self.set_time_offset, 1, 1
+        )
+
+        utc_timestamp = int((utc_now - offset_time).total_seconds())
+        local_timestamp = int((now - offset_time_local).total_seconds())
+
+        payload_rsp.extend(utc_timestamp.to_bytes(4, "big", signed=False))
+        payload_rsp.extend(local_timestamp.to_bytes(4, "big", signed=False))
+
+        self.create_catching_task(
+            super().command(TUYA_SET_TIME, payload_rsp, expect_reply=False)
+        )
+
+        return foundation.Status.SUCCESS
+
 
 class TuyaOnOff(OnOff, TuyaLocalCluster):
     """Tuya MCU OnOff cluster."""