|
2 | 2 |
|
3 | 3 | from unittest import mock |
4 | 4 |
|
| 5 | +import zigpy.types as t |
5 | 6 | from zigpy.zcl import ClusterType, foundation |
6 | 7 | from zigpy.zcl.clusters.smartenergy import Metering |
7 | 8 |
|
| 9 | +from tests.common import ClusterListener |
8 | 10 | import zhaquirks |
9 | 11 |
|
10 | 12 | zhaquirks.setup() |
@@ -137,3 +139,51 @@ async def test_frient_emi(zigpy_device_from_v2_quirk): |
137 | 139 | assert attr_data == b"\x00\x00%d\x00\x00\x00\x00\x00" |
138 | 140 |
|
139 | 141 | assert result == (foundation.Status.SUCCESS, "done") |
| 142 | + |
| 143 | + |
| 144 | +async def test_mfg_cluster_events(zigpy_device_from_v2_quirk): |
| 145 | + """Test Frient EMI Norwegian HAN ignoring incorrect divisor attribute reports.""" |
| 146 | + device = zigpy_device_from_v2_quirk("frient A/S", "EMIZB-132", endpoint_ids=[1, 2]) |
| 147 | + |
| 148 | + metering_cluster = device.endpoints[2].smartenergy_metering |
| 149 | + metering_listener = ClusterListener(metering_cluster) |
| 150 | + |
| 151 | + # divisor already fixed at 1000 |
| 152 | + assert metering_cluster.get(Metering.AttributeDefs.divisor.id) == 1000 |
| 153 | + |
| 154 | + # send incorrect divisor attribute report |
| 155 | + device.packet_received( |
| 156 | + t.ZigbeePacket( |
| 157 | + profile_id=260, |
| 158 | + cluster_id=Metering.cluster_id, |
| 159 | + src_ep=2, |
| 160 | + dst_ep=2, |
| 161 | + # XXX: get data from real device |
| 162 | + data=t.SerializableBytes(b'\x1c\x00\x00\x01\n\x02\x03"\x00\x02\x00'), |
| 163 | + ) |
| 164 | + ) |
| 165 | + |
| 166 | + # attribute_updated event should not be emitted |
| 167 | + assert len(metering_listener.attribute_updates) == 0 |
| 168 | + |
| 169 | + # divisor should still be fixed at 1000 |
| 170 | + assert metering_cluster.get(Metering.AttributeDefs.divisor.id) == 1000 |
| 171 | + |
| 172 | + # send current_summ_delivered attribute report |
| 173 | + device.packet_received( |
| 174 | + t.ZigbeePacket( |
| 175 | + profile_id=260, |
| 176 | + cluster_id=Metering.cluster_id, |
| 177 | + src_ep=2, |
| 178 | + dst_ep=2, |
| 179 | + data=t.SerializableBytes( |
| 180 | + b"\x1c\x00\x00\x01\n\x00\x00%\xd2\x04\x00\x00\x00\x00" |
| 181 | + ), |
| 182 | + ) |
| 183 | + ) |
| 184 | + |
| 185 | + # attribute_updated event should be emitted |
| 186 | + assert len(metering_listener.attribute_updates) == 1 |
| 187 | + assert ( |
| 188 | + metering_cluster.get(Metering.AttributeDefs.current_summ_delivered.id) == 1234 |
| 189 | + ) |
0 commit comments