diff --git a/tests/common.py b/tests/common.py index 319cd19dff..16fe56b4b7 100644 --- a/tests/common.py +++ b/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) diff --git a/tests/test_tuya.py b/tests/test_tuya.py index 210122270a..74b5d87b13 100644 --- a/tests/test_tuya.py +++ b/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) diff --git a/tests/test_tuya_mcu.py b/tests/test_tuya_mcu.py index c910c7ae12..fc79c74f2b 100644 --- a/tests/test_tuya_mcu.py +++ b/tests/test_tuya_mcu.py @@ -1,5 +1,6 @@ """Tests for Tuya quirks.""" +import datetime from unittest import mock import pytest @@ -7,19 +8,21 @@ 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(): diff --git a/zhaquirks/tuya/mcu/__init__.py b/zhaquirks/tuya/mcu/__init__.py index 76c10554cf..aecd941e14 100644 --- a/zhaquirks/tuya/mcu/__init__.py +++ b/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,6 +14,7 @@ TUYA_MCU_COMMAND, TUYA_MCU_VERSION_RSP, TUYA_SET_DATA, + TUYA_SET_TIME, Data, NoManufacturerCluster, PowerOnState, @@ -20,6 +22,7 @@ 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."""