Skip to content

Commit

Permalink
Merge branch 'dev' into dev
Browse files Browse the repository at this point in the history
  • Loading branch information
jpizquierdo authored Sep 25, 2024
2 parents edbd58a + 2bdd076 commit 9772ab6
Show file tree
Hide file tree
Showing 34 changed files with 2,871 additions and 437 deletions.
11 changes: 8 additions & 3 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
@@ -1,18 +1,23 @@
ci:
autofix_commit_msg: "Apply pre-commit auto fixes"
autoupdate_commit_msg: "Auto-update pre-commit hooks"
skip: [mypy]

repos:
- repo: https://github.com/codespell-project/codespell
rev: v2.2.6
rev: v2.3.0
hooks:
- id: codespell
additional_dependencies: [tomli]
args: ["--toml", "pyproject.toml"]

- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.9.0
rev: v1.11.2
hooks:
- id: mypy

- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.3.2
rev: v0.6.7
hooks:
- id: ruff
args: ["--fix", "--exit-non-zero-on-fix", "--config", "pyproject.toml"]
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ readme = "README.md"
license = {text = "Apache License Version 2.0"}
requires-python = ">=3.12"
dependencies = [
"zigpy>=0.65.2",
"zigpy>=0.66.0",
]

[tool.setuptools.packages.find]
Expand Down
2 changes: 1 addition & 1 deletion requirements_test.txt
Original file line number Diff line number Diff line change
Expand Up @@ -13,5 +13,5 @@ pytest-sugar
pytest-timeout
pytest-asyncio
pytest>=7.1.3
zigpy>=0.65.3
zigpy>=0.66.0
ruff==0.0.261
5 changes: 3 additions & 2 deletions tests/test_danfoss.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from unittest import mock

from zigpy.quirks import CustomCluster
import zigpy.types as t
from zigpy.zcl import foundation
from zigpy.zcl.clusters.hvac import Thermostat
from zigpy.zcl.foundation import WriteAttributesStatusRecord, ZCLAttributeDef
Expand Down Expand Up @@ -161,8 +162,8 @@ async def test_customized_standardcluster(zigpy_device_from_quirk):
) == [[4545, 345], [5433, 45355]]

mock_attributes = {
656: ZCLAttributeDef(is_manufacturer_specific=True),
56454: ZCLAttributeDef(is_manufacturer_specific=False),
656: ZCLAttributeDef(type=t.uint8_t, is_manufacturer_specific=True),
56454: ZCLAttributeDef(type=t.uint8_t, is_manufacturer_specific=False),
}

danfoss_thermostat_cluster.attributes = mock_attributes
Expand Down
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
Loading

0 comments on commit 9772ab6

Please sign in to comment.