Skip to content

Commit

Permalink
Convert Sinope light to new style AttributeDefs, add ZHA events (#3313)
Browse files Browse the repository at this point in the history
Co-authored-by: TheJulianJES <[email protected]>
  • Loading branch information
ckm2k1 and TheJulianJES authored Sep 24, 2024
1 parent e749877 commit 2bdd076
Show file tree
Hide file tree
Showing 3 changed files with 322 additions and 57 deletions.
152 changes: 149 additions & 3 deletions tests/test_sinope.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,32 @@
"""Tests for Sinope."""

from unittest import mock

import pytest
from zigpy.device import Device
import zigpy.types as t
from zigpy.zcl import foundation
from zigpy.zcl.clusters.general import DeviceTemperature
from zigpy.zcl.clusters.measurement import FlowMeasurement

from tests.common import ClusterListener
import zhaquirks.sinope.switch
import zhaquirks
from zhaquirks.const import COMMAND_BUTTON_DOUBLE, COMMAND_BUTTON_HOLD
from zhaquirks.sinope import SINOPE_MANUFACTURER_CLUSTER_ID
from zhaquirks.sinope.light import (
SinopeTechnologieslight,
SinopeTechnologiesManufacturerCluster,
)
from zhaquirks.sinope.switch import SinopeTechnologiesCalypso, SinopeTechnologiesValveG2

zhaquirks.setup()

ButtonAction = SinopeTechnologiesManufacturerCluster.Action

SINOPE_MANUFACTURER_ID = 4508 # 0x119C

@pytest.mark.parametrize("quirk", (zhaquirks.sinope.switch.SinopeTechnologiesCalypso,))

@pytest.mark.parametrize("quirk", (SinopeTechnologiesCalypso,))
async def test_sinope_device_temp(zigpy_device_from_quirk, quirk):
"""Test that device temperature is multiplied."""
device = zigpy_device_from_quirk(quirk)
Expand All @@ -33,7 +49,7 @@ async def test_sinope_device_temp(zigpy_device_from_quirk, quirk):
assert dev_temp_listener.attribute_updates[1][1] == 25 # not modified


@pytest.mark.parametrize("quirk", (zhaquirks.sinope.switch.SinopeTechnologiesValveG2,))
@pytest.mark.parametrize("quirk", (SinopeTechnologiesValveG2,))
async def test_sinope_flow_measurement(zigpy_device_from_quirk, quirk):
"""Test that flow measurement measured value is divided."""
device = zigpy_device_from_quirk(quirk)
Expand All @@ -57,3 +73,133 @@ async def test_sinope_flow_measurement(zigpy_device_from_quirk, quirk):
== flow_measurement_other_attr_id
)
assert flow_measurement_listener.attribute_updates[1][1] == 25 # not modified


def _get_packet_data(
command: foundation.GeneralCommand,
attr: foundation.Attribute | None = None,
dirc: foundation.Direction = foundation.Direction.Server_to_Client,
) -> bytes:
hdr = foundation.ZCLHeader.general(
1, command, SINOPE_MANUFACTURER_ID, dirc
).serialize()
if attr is not None:
cmd = foundation.GENERAL_COMMANDS[command].schema([attr]).serialize()
else:
cmd = b""
return t.SerializableBytes(hdr + cmd).serialize()


@pytest.mark.parametrize("quirk", (SinopeTechnologieslight,))
@pytest.mark.parametrize(
"press_type,exp_event",
(
(ButtonAction.Single_off, None),
(ButtonAction.Single_on, None),
(ButtonAction.Double_on, COMMAND_BUTTON_DOUBLE),
(ButtonAction.Double_off, COMMAND_BUTTON_DOUBLE),
(ButtonAction.Long_on, COMMAND_BUTTON_HOLD),
(ButtonAction.Long_off, COMMAND_BUTTON_HOLD),
# Should gracefully handle broken actions.
(t.uint8_t(0x00), None),
),
)
async def test_sinope_light_switch(
zigpy_device_from_quirk, quirk, press_type, exp_event
):
"""Test that button presses are sent as events."""
device: Device = zigpy_device_from_quirk(quirk)
cluster_id = SINOPE_MANUFACTURER_CLUSTER_ID
endpoint_id = 1

class Listener:
zha_send_event = mock.MagicMock()

cluster_listener = Listener()
device.endpoints[endpoint_id].in_clusters[cluster_id].add_listener(cluster_listener)

attr = foundation.Attribute(
attrid=0x54, # "action_report" attribute
value=foundation.TypeValue(
type=t.enum8(0x30),
value=press_type,
),
)
data = _get_packet_data(foundation.GeneralCommand.Report_Attributes, attr)
device.handle_message(260, cluster_id, endpoint_id, endpoint_id, data)

if exp_event is None:
assert cluster_listener.zha_send_event.call_count == 0
else:
assert cluster_listener.zha_send_event.call_count == 1
assert cluster_listener.zha_send_event.call_args == mock.call(
exp_event,
{
"attribute_id": 84,
"attribute_name": "action_report",
"value": press_type.value,
},
)


@pytest.mark.parametrize("quirk", (SinopeTechnologieslight,))
async def test_sinope_light_switch_non_action_report(zigpy_device_from_quirk, quirk):
"""Test commands not handled by custom handler.
Make sure that non attribute report commands and attribute reports that don't
concern action_report are passed through to base class.
"""

device: Device = zigpy_device_from_quirk(quirk)
cluster_id = SINOPE_MANUFACTURER_CLUSTER_ID
endpoint_id = 1

class Listener:
zha_send_event = mock.MagicMock()

cluster_listener = Listener()
device.endpoints[endpoint_id].in_clusters[cluster_id].add_listener(cluster_listener)

# read attributes general command
data = _get_packet_data(foundation.GeneralCommand.Read_Attributes)
device.handle_message(260, cluster_id, endpoint_id, endpoint_id, data)
# no ZHA events emitted because we only handle Report_Attributes
assert cluster_listener.zha_send_event.call_count == 0

# report attributes command, but not "action_report"
attr = foundation.Attribute(
attrid=0x10, # "on_intensity" attribute
value=foundation.TypeValue(
type=t.int16s(0x29), value=t.int16s(50)
), # 0x29 = t.int16s
)
data = _get_packet_data(foundation.GeneralCommand.Report_Attributes, attr)
device.handle_message(260, cluster_id, endpoint_id, endpoint_id, data)
# ZHA event emitted because we pass non "action_report"
# reports to the base class handler.
assert cluster_listener.zha_send_event.call_count == 1


@pytest.mark.parametrize("quirk", (SinopeTechnologieslight,))
async def test_sinope_light_switch_reporting(zigpy_device_from_quirk, quirk):
"""Test that configuring reporting for action_report works."""
device: Device = zigpy_device_from_quirk(quirk)

manu_cluster = device.endpoints[1].in_clusters[SINOPE_MANUFACTURER_CLUSTER_ID]

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 manu_cluster.bind()
await manu_cluster.configure_reporting(
SinopeTechnologiesManufacturerCluster.AttributeDefs.action_report.id,
3600,
10800,
1,
)

assert len(request_mock.mock_calls) == 1
assert len(bind_mock.mock_calls) == 1
2 changes: 1 addition & 1 deletion zhaquirks/sinope/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@

SINOPE = "Sinope Technologies"
SINOPE_MANUFACTURER_CLUSTER_ID = 0xFF01
ATTRIBUTE_ACTION = "actionReport"
ATTRIBUTE_ACTION = "action_report"

LIGHT_DEVICE_TRIGGERS = {
(SHORT_PRESS, TURN_ON): {
Expand Down
Loading

0 comments on commit 2bdd076

Please sign in to comment.