Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Refactor Aqara Roller Driver E1 as v2 quirk to expose configuration and status entities #3686

Open
wants to merge 8 commits into
base: dev
Choose a base branch
from
Open

Refactor Aqara Roller Driver E1 as v2 quirk to expose configuration and status entities #3686

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
535e8ff
Refactor as v2 quirk
jeverley Jan 8, 2025
8140edb
Apply pre-commit auto fixes
pre-commit-ci[bot] Jan 8, 2025
cd98101
Update tests for v2 quirk
jeverley Jan 8, 2025
54ef750
Correct reference to attribute name
jeverley Jan 8, 2025
6e9b649
Update binary_sensor conversion test for charging status
jeverley Jan 8, 2025
babacac
Add mock_read for test handling stop command position read
jeverley Jan 8, 2025
a01e2ba
Validate analog output read occurs in stop command test
jeverley Jan 8, 2025
5498f26
Test non-defined enum to bool conversion values
jeverley Jan 8, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
134 changes: 100 additions & 34 deletions tests/test_xiaomi.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1388,15 +1388,12 @@ async def test_xiaomi_power_cluster_not_used(zigpy_device_from_quirk, caplog, qu
)


@pytest.mark.parametrize(
"quirk", (zhaquirks.xiaomi.aqara.roller_curtain_e1.RollerE1AQ,)
)
async def test_xiaomi_e1_roller_curtain_battery(zigpy_device_from_quirk, quirk):
def test_xiaomi_e1_roller_curtain_battery(zigpy_device_from_v2_quirk):
"""Test Aqara E1 roller curtain battery reporting."""
# Ideally, get a real Xiaomi "heartbeat" message to test.
# For now, fake the heartbeat message and check if battery parsing works.

device = zigpy_device_from_quirk(quirk)
device = zigpy_device_from_v2_quirk(LUMI, "lumi.curtain.acn002")

basic_cluster = device.endpoints[1].basic
ClusterListener(basic_cluster)
Expand Down Expand Up @@ -1588,33 +1585,63 @@ async def test_xiaomi_e1_driver_light_level(


@pytest.mark.parametrize(
"command, value",
"command, value, read_analog_output",
[
(WindowCovering.ServerCommandDefs.up_open.id, 1),
(WindowCovering.ServerCommandDefs.down_close.id, 0),
(WindowCovering.ServerCommandDefs.stop.id, 2),
(WindowCovering.ServerCommandDefs.up_open.id, 1, False),
(WindowCovering.ServerCommandDefs.down_close.id, 0, False),
(WindowCovering.ServerCommandDefs.stop.id, 2, True),
],
)
async def test_xiaomi_e1_roller_commands_1(zigpy_device_from_quirk, command, value):
async def test_xiaomi_e1_roller_commands_1(
zigpy_device_from_v2_quirk, command, value, read_analog_output
):
"""Test Aqara E1 roller commands for basic movement functions using MultistateOutput Cluster."""
device = zigpy_device_from_quirk(
zhaquirks.xiaomi.aqara.roller_curtain_e1.RollerE1AQ
)
device = zigpy_device_from_v2_quirk(LUMI, "lumi.curtain.acn002")

window_covering_cluster = device.endpoints[1].window_covering
analog_cluster = device.endpoints[1].analog_output
multistate_cluster = device.endpoints[1].multistate_output
multistate_cluster._write_attributes = mock.AsyncMock(
return_value=(
[foundation.WriteAttributesStatusRecord(foundation.Status.SUCCESS)],
)
)
attr_id = MultistateOutput.AttributeDefs.present_value.id
multistate_attr_id = MultistateOutput.AttributeDefs.present_value.id
analog_attr_id = AnalogOutput.AttributeDefs.present_value.id

# test command
await window_covering_cluster.command(command)
assert multistate_cluster._write_attributes.call_count == 1
assert multistate_cluster._write_attributes.call_args[0][0][0].attrid == attr_id
assert multistate_cluster._write_attributes.call_args[0][0][0].value.value == value
# fake read response for attributes: return 1 for all attributes
# this is needed because the stop command issues a read on the AnalogOutput cluster
def mock_read(attributes, manufacturer=None):
records = [
foundation.ReadAttributeRecord(
attr, foundation.Status.SUCCESS, foundation.TypeValue(None, 1)
)
for attr in attributes
]
return (records,)

patch_analog_cluster_read = mock.patch.object(
analog_cluster, "_read_attributes", mock.AsyncMock(side_effect=mock_read)
)
with patch_analog_cluster_read:
# test command
await window_covering_cluster.command(command)
assert multistate_cluster._write_attributes.call_count == 1
assert (
multistate_cluster._write_attributes.call_args[0][0][0].attrid
== multistate_attr_id
)
assert (
multistate_cluster._write_attributes.call_args[0][0][0].value.value == value
)
# test that the analog output read occurs
if read_analog_output:
assert len(analog_cluster._read_attributes.mock_calls) == 1
assert analog_cluster._read_attributes.mock_calls[0][1][0] == [
analog_attr_id
]
else:
assert len(analog_cluster._read_attributes.mock_calls) == 0


@pytest.mark.parametrize(
Expand All @@ -1623,11 +1650,9 @@ async def test_xiaomi_e1_roller_commands_1(zigpy_device_from_quirk, command, val
(WindowCovering.ServerCommandDefs.go_to_lift_percentage.id, 60),
],
)
async def test_xiaomi_e1_roller_commands_2(zigpy_device_from_quirk, command, value):
async def test_xiaomi_e1_roller_commands_2(zigpy_device_from_v2_quirk, command, value):
"""Test Aqara E1 roller commands for go to lift percentage using AnalogOutput cluster."""
device = zigpy_device_from_quirk(
zhaquirks.xiaomi.aqara.roller_curtain_e1.RollerE1AQ
)
device = zigpy_device_from_v2_quirk(LUMI, "lumi.curtain.acn002")

window_covering_cluster = device.endpoints[1].window_covering
analog_cluster = device.endpoints[1].analog_output
Expand All @@ -1647,6 +1672,47 @@ async def test_xiaomi_e1_roller_commands_2(zigpy_device_from_quirk, command, val
)


@pytest.mark.parametrize(
"attribute_id, device_value, converted_value",
[
(
zhaquirks.xiaomi.aqara.roller_curtain_e1.XiaomiAqaraRollerE1.AttributeDefs.charging.id,
zhaquirks.xiaomi.aqara.roller_curtain_e1.AqaraRollerDriverCharging.true,
t.Bool.true,
),
(
zhaquirks.xiaomi.aqara.roller_curtain_e1.XiaomiAqaraRollerE1.AttributeDefs.charging.id,
zhaquirks.xiaomi.aqara.roller_curtain_e1.AqaraRollerDriverCharging.false,
t.Bool.false,
),
(
zhaquirks.xiaomi.aqara.roller_curtain_e1.XiaomiAqaraRollerE1.AttributeDefs.charging.id,
3,
None,
),
],
)
async def test_xiaomi_e1_roller_bool_value_conversion(
zigpy_device_from_v2_quirk, attribute_id, device_value, converted_value
):
"""Test remap of Aqara values to binary_sensor bool values."""
device = zigpy_device_from_v2_quirk(LUMI, "lumi.curtain.acn002")

opple_cluster = device.endpoints[1].opple_cluster
opple_listener = ClusterListener(opple_cluster)

# update aqara attribute
opple_cluster.update_attribute(attribute_id, device_value)
if converted_value is None:
assert len(opple_listener.attribute_updates) == 0
return
assert len(opple_listener.attribute_updates) == 1

# confirm aqara device value has been remapped to the expected bool value
assert opple_listener.attribute_updates[0][0] == attribute_id
assert opple_listener.attribute_updates[0][1] == converted_value


@pytest.mark.parametrize("endpoint", [(1), (2)])
async def test_aqara_t2_relay(zigpy_device_from_quirk, endpoint):
"""Test Aqara T2 relay."""
Expand Down Expand Up @@ -1774,11 +1840,11 @@ async def test_aqara_fp1e_sensor(
expected_motion_status,
):
"""Test Aqara FP1E sensor."""
quirk = zigpy_device_from_v2_quirk("aqara", "lumi.sensor_occupy.agl1")
device = zigpy_device_from_v2_quirk("aqara", "lumi.sensor_occupy.agl1")

opple_cluster = quirk.endpoints[1].opple_cluster
ias_cluster = quirk.endpoints[1].ias_zone
occupancy_cluster = quirk.endpoints[1].occupancy
opple_cluster = device.endpoints[1].opple_cluster
ias_cluster = device.endpoints[1].ias_zone
occupancy_cluster = device.endpoints[1].occupancy

opple_listener = ClusterListener(opple_cluster)
ias_listener = ClusterListener(ias_cluster)
Expand Down Expand Up @@ -1809,15 +1875,15 @@ async def test_aqara_fp1e_sensor(
def test_h1_wireless_remotes(zigpy_device_from_v2_quirk):
"""Test Aqara H1 wireless remote quirk adds missing endpoints."""
# create device with endpoint 1 only and verify we don't get a KeyError
quirk = zigpy_device_from_v2_quirk(LUMI, "lumi.remote.b28ac1")
device = zigpy_device_from_v2_quirk(LUMI, "lumi.remote.b28ac1")

# verify the quirk adds endpoints 2 and 3
assert 2 in quirk.endpoints
assert 3 in quirk.endpoints
assert 2 in device.endpoints
assert 3 in device.endpoints

# verify the quirk adds the correct clusters to the new endpoints
assert OnOff.cluster_id in quirk.endpoints[2].out_clusters
assert OnOff.cluster_id in quirk.endpoints[3].out_clusters
assert OnOff.cluster_id in device.endpoints[2].out_clusters
assert OnOff.cluster_id in device.endpoints[3].out_clusters

assert MultistateInput.cluster_id in quirk.endpoints[2].in_clusters
assert MultistateInput.cluster_id in quirk.endpoints[3].in_clusters
assert MultistateInput.cluster_id in device.endpoints[2].in_clusters
assert MultistateInput.cluster_id in device.endpoints[3].in_clusters
Loading
Loading