diff --git a/bellows/ezsp/protocol.py b/bellows/ezsp/protocol.py index 4b0853602..4d16017b8 100644 --- a/bellows/ezsp/protocol.py +++ b/bellows/ezsp/protocol.py @@ -332,6 +332,11 @@ async def write_child_data(self, children: dict[t.EUI64, t.NWK]) -> None: async def initialize_network(self) -> t.sl_Status: raise NotImplementedError + @abc.abstractmethod + async def clear_gp_tables(self) -> None: + """Clear the Green Power proxy and sink tables of the NCP, where it has them.""" + raise NotImplementedError + @abc.abstractmethod async def factory_reset(self) -> None: raise NotImplementedError @@ -363,7 +368,10 @@ async def send_multicast( non_member_radius: t.uint8_t, message_tag: t.uint8_t, data: bytes, + alias: t.NWK | None = None, + alias_seq: t.uint8_t | None = None, ) -> tuple[t.sl_Status, t.uint8_t]: + """Send a multicast, as `alias` with the NWK sequence `alias_seq` if given.""" raise NotImplementedError @abc.abstractmethod @@ -375,7 +383,10 @@ async def send_broadcast( message_tag: t.uint8_t, aps_sequence: t.uint8_t, data: bytes, + alias: t.NWK | None = None, + alias_seq: t.uint8_t | None = None, ) -> tuple[t.sl_Status, t.uint8_t]: + """Send a broadcast, as `alias` with the NWK sequence `alias_seq` if given.""" raise NotImplementedError @abc.abstractmethod diff --git a/bellows/ezsp/v14/__init__.py b/bellows/ezsp/v14/__init__.py index f06d4cf06..3076088f0 100644 --- a/bellows/ezsp/v14/__init__.py +++ b/bellows/ezsp/v14/__init__.py @@ -122,13 +122,16 @@ async def send_multicast( non_member_radius: t.uint8_t, message_tag: t.uint8_t, data: bytes, + alias: t.NWK | None = None, + alias_seq: t.uint8_t | None = None, ) -> tuple[t.sl_Status, t.uint8_t]: status, sequence = await self.sendMulticast( aps_frame=aps_frame, hops=radius, broadcast_addr=t.BroadcastAddress.RX_ON_WHEN_IDLE, - alias=NULL_NODE_ID, - sequence=0x00, # NWK sequence, only used with an alias + alias=NULL_NODE_ID if alias is None else alias, + # NWK sequence, only used with an alias + sequence=0x00 if alias_seq is None else alias_seq, message_tag=message_tag, message=data, ) @@ -143,13 +146,16 @@ async def send_broadcast( message_tag: t.uint8_t, aps_sequence: t.uint8_t, data: bytes, + alias: t.NWK | None = None, + alias_seq: t.uint8_t | None = None, ) -> tuple[t.sl_Status, t.uint8_t]: # `aps_sequence` is not used status, sequence = await self.sendBroadcast( - alias=NULL_NODE_ID, + alias=NULL_NODE_ID if alias is None else alias, destination=address, - sequence=0x00, # NWK sequence, only used with an alias + # NWK sequence, only used with an alias + sequence=0x00 if alias_seq is None else alias_seq, aps_frame=aps_frame, radius=radius, message_tag=message_tag, diff --git a/bellows/ezsp/v14/commands.py b/bellows/ezsp/v14/commands.py index 311f462bb..26a9e4521 100644 --- a/bellows/ezsp/v14/commands.py +++ b/bellows/ezsp/v14/commands.py @@ -1,4 +1,5 @@ from zigpy.types import EUI64, NWK, BroadcastAddress, Struct, StructField +import zigpy.zgp.types as zgp_t import bellows.types as t @@ -377,6 +378,26 @@ class GetTokenDataRsp(Struct): "status": t.EmberStatus, }, ), + "gpepIncomingMessageHandler": ( + 0x00C5, + {}, + { + # The GP status codes migrate to their own enum, not to `sl_Status` + "status": t.sl_GpStatus, + "gpdLink": zgp_t.GPPGPDLink, + "sequenceNumber": t.uint8_t, + "addr": t.EmberGpAddress, + "gpdfSecurityLevel": t.EmberGpSecurityLevel, + "gpdfSecurityKeyType": t.EmberGpKeyType, + "autoCommissioning": t.Bool, + "bidirectionalInfo": t.EmberGpBidirectionalInfo, + "gpdSecurityFrameCounter": t.uint32_t, + "gpdCommandId": zgp_t.GPDCommandID, + "mic": t.uint32_t, + "proxyTableIndex": t.uint8_t, + "gpdCommandPayload": t.LVBytes, + }, + ), # Gained an `options` argument, which the NCP reads even if we don't send it "leaveNetwork": ( 0x0020, diff --git a/bellows/ezsp/v17/__init__.py b/bellows/ezsp/v17/__init__.py index 1f6d98329..8a3897f94 100644 --- a/bellows/ezsp/v17/__init__.py +++ b/bellows/ezsp/v17/__init__.py @@ -1,13 +1,18 @@ """"EZSP Protocol version 17 protocol handler.""" from __future__ import annotations +import logging + import voluptuous as vol import bellows.config +from bellows.exception import InvalidCommandError from . import commands, config from ..v16 import EZSPv16 +LOGGER = logging.getLogger(__name__) + class EZSPv17(EZSPv16): """EZSP Version 17 Protocol version handler.""" @@ -18,3 +23,10 @@ class EZSPv17(EZSPv16): bellows.config.CONF_EZSP_CONFIG: vol.Schema(config.EZSP_SCHEMA), bellows.config.CONF_EZSP_POLICIES: vol.Schema(config.EZSP_POLICIES_SCH), } + + async def clear_gp_tables(self) -> None: + try: + await self.gpSinkTableClearAll() + await self.gpClearProxyTable() + except InvalidCommandError: + LOGGER.debug("The NCP has no Green Power tables") diff --git a/bellows/ezsp/v17/commands.py b/bellows/ezsp/v17/commands.py index 47cdc2c50..4cb4d3d4a 100644 --- a/bellows/ezsp/v17/commands.py +++ b/bellows/ezsp/v17/commands.py @@ -1,3 +1,5 @@ +import zigpy.zgp.types as zgp_t + import bellows.types as t from ..v16.commands import COMMANDS as COMMANDS_v16 @@ -29,19 +31,22 @@ {}, { "status": t.sl_GpStatus, - "gpdLink": t.uint8_t, + "gpdLink": zgp_t.GPPGPDLink, "sequenceNumber": t.uint8_t, "addr": t.EmberGpAddress, "gpdfSecurityLevel": t.EmberGpSecurityLevel, "gpdfSecurityKeyType": t.EmberGpKeyType, "autoCommissioning": t.Bool, - "bidirectionalInfo": t.uint8_t, + "bidirectionalInfo": t.EmberGpBidirectionalInfo, "gpdSecurityFrameCounter": t.uint32_t, "gpdCommandId": t.uint8_t, "mic": t.uint32_t, "proxyTableIndex": t.uint8_t, "gpdCommandPayload": t.LVBytes, - "packetInfo": t.SlRxPacketInfo, + # XXX: `packetInfo` is the final field according to the protocol YAML + # definitions and all SDK code but it is never actually appended by the + # firmware. + # "packetInfo": t.SlRxPacketInfo, }, ), } diff --git a/bellows/ezsp/v4/__init__.py b/bellows/ezsp/v4/__init__.py index 46ae33c33..d937444af 100644 --- a/bellows/ezsp/v4/__init__.py +++ b/bellows/ezsp/v4/__init__.py @@ -124,6 +124,10 @@ async def initialize_network(self) -> t.sl_Status: ) return t.sl_Status.from_ember_status(init_status) + async def clear_gp_tables(self) -> None: + # No Green Power table clearing before EZSPv6 + pass + async def factory_reset(self) -> None: await self.clearKeyTable() @@ -163,7 +167,13 @@ async def send_multicast( non_member_radius: t.uint8_t, message_tag: t.uint8_t, data: bytes, + alias: t.NWK | None = None, + alias_seq: t.uint8_t | None = None, ) -> tuple[t.sl_Status, t.uint8_t]: + if alias is not None: + # TODO: `sendMulticastWithAlias`, from EZSPv6 + raise NotImplementedError("Aliased multicasts are not supported") + status, sequence = await self.sendMulticast( apsFrame=aps_frame, hops=radius, @@ -182,9 +192,24 @@ async def send_broadcast( message_tag: t.uint8_t, aps_sequence: t.uint8_t, data: bytes, + alias: t.NWK | None = None, + alias_seq: t.uint8_t | None = None, ) -> tuple[t.sl_Status, t.uint8_t]: # `aps_sequence` is not used + if alias is not None: + status, sequence = await self.proxyBroadcast( + source=t.EmberNodeId(alias), + destination=address, + nwkSequence=t.uint8_t(0 if alias_seq is None else alias_seq), + apsFrame=aps_frame, + radius=radius, + messageTag=message_tag, + messageContents=data, + ) + + return t.sl_Status.from_ember_status(status), sequence + status, sequence = await self.sendBroadcast( destination=address, apsFrame=aps_frame, diff --git a/bellows/ezsp/v4/commands.py b/bellows/ezsp/v4/commands.py index 64e4e4cb5..60821ce1a 100644 --- a/bellows/ezsp/v4/commands.py +++ b/bellows/ezsp/v4/commands.py @@ -1,3 +1,5 @@ +import zigpy.zgp.types as zgp_t + import bellows.types as t COMMANDS = { @@ -1991,20 +1993,18 @@ 0x00C5, {}, { + # The NCP sends the legacy GP status codes, `EmberStatus` 0x78-0x7F, and + # `SUCCESS` for a GPDF it authenticated itself "status": t.EmberStatus, - "gpdLink": t.uint8_t, + "gpdLink": zgp_t.GPPGPDLink, "sequenceNumber": t.uint8_t, - "addrType": t.uint8_t, - "addr": t.uint32_t, - "applicationId": t.uint8_t, - "address": t.EUI64, # TODO: `sourceId` if appId == 0 else `gpdIeeeAddress` - "endpoint": t.uint8_t, + "addr": t.EmberGpAddress, "gpdfSecurityLevel": t.EmberGpSecurityLevel, "gpdfSecurityKeyType": t.EmberGpKeyType, "autoCommissioning": t.Bool, - "bidirectionalInfo": t.Bool, + "bidirectionalInfo": t.EmberGpBidirectionalInfo, "gpdSecurityFrameCounter": t.uint32_t, - "gpdCommandId": t.uint8_t, + "gpdCommandId": zgp_t.GPDCommandID, "mic": t.uint32_t, "proxyTableIndex": t.uint8_t, "gpdCommandPayload": t.LVBytes, diff --git a/bellows/ezsp/v6/__init__.py b/bellows/ezsp/v6/__init__.py index 729357edb..05dfff24e 100644 --- a/bellows/ezsp/v6/__init__.py +++ b/bellows/ezsp/v6/__init__.py @@ -4,6 +4,7 @@ import voluptuous import bellows.config +from bellows.exception import InvalidCommandError import bellows.types as t from . import commands, config @@ -27,3 +28,41 @@ async def initialize_network(self) -> t.sl_Status: networkInitBitmask=t.EmberNetworkInitBitmask.NETWORK_INIT_NO_OPTIONS ) return t.sl_Status.from_ember_status(init_status) + + async def clear_gp_tables(self) -> None: + # The proxy table has no bulk clear before EZSPv17 + try: + await self.gpClearSinkTable() + except InvalidCommandError: + LOGGER.debug("The NCP has no Green Power sink table") + + async def send_multicast( + self, + aps_frame: t.EmberApsFrame, + radius: t.uint8_t, + non_member_radius: t.uint8_t, + message_tag: t.uint8_t, + data: bytes, + alias: t.NWK | None = None, + alias_seq: t.uint8_t | None = None, + ) -> tuple[t.sl_Status, t.uint8_t]: + if alias is None: + return await super().send_multicast( + aps_frame=aps_frame, + radius=radius, + non_member_radius=non_member_radius, + message_tag=message_tag, + data=data, + ) + + status, sequence = await self.sendMulticastWithAlias( + apsFrame=aps_frame, + hops=radius, + nonmemberRadius=non_member_radius, + alias=t.EmberNodeId(alias), + nwkSequence=t.uint8_t(0 if alias_seq is None else alias_seq), + messageTag=message_tag, + messageContents=data, + ) + + return t.sl_Status.from_ember_status(status), sequence diff --git a/bellows/ezsp/v6/commands.py b/bellows/ezsp/v6/commands.py index 11228416d..37d460fab 100644 --- a/bellows/ezsp/v6/commands.py +++ b/bellows/ezsp/v6/commands.py @@ -186,12 +186,12 @@ "sendMulticastWithAlias": ( 0x3A, { - "hops": t.EmberApsFrame, + "apsFrame": t.EmberApsFrame, + "hops": t.uint8_t, "nonmemberRadius": t.uint8_t, - "alias": t.uint8_t, - "nwkSequence": t.uint16_t, + "alias": t.EmberNodeId, + "nwkSequence": t.uint8_t, "messageTag": t.uint8_t, - "messageLength": t.uint8_t, "messageContents": t.LVBytes, }, { diff --git a/bellows/ezsp/v7/__init__.py b/bellows/ezsp/v7/__init__.py index 9e26d9043..9c6d7af1d 100644 --- a/bellows/ezsp/v7/__init__.py +++ b/bellows/ezsp/v7/__init__.py @@ -7,6 +7,7 @@ import voluptuous import bellows.config +from bellows.exception import InvalidCommandError import bellows.types as t from . import commands, config @@ -25,6 +26,13 @@ class EZSPv7(EZSPv6): bellows.config.CONF_EZSP_POLICIES: voluptuous.Schema(config.EZSP_POLICIES_SCH), } + async def clear_gp_tables(self) -> None: + # `gpClearSinkTable` is renamed in EZSPv7 + try: + await self.gpSinkTableClearAll() + except InvalidCommandError: + LOGGER.debug("The NCP has no Green Power sink table") + async def read_child_data( self, ) -> AsyncGenerator[tuple[t.NWK, t.EUI64, t.EmberNodeType]]: diff --git a/bellows/types/named.py b/bellows/types/named.py index aad9c4f0e..663915576 100644 --- a/bellows/types/named.py +++ b/bellows/types/named.py @@ -58,6 +58,16 @@ class EmberGpKeyType(basic.uint8_t): pass +class EmberGpBidirectionalInfo(basic.bitmap8): + # Bidirectional information of a received GPDF. + + # The incoming GPDF has its rxAfterTx bit set: the GPD briefly opens its receiver + # after the frame and can be sent a GPDF in response. + RX_AFTER_TX = 0x01 + # The transmit queue of the green power stub is able to hold an outgoing GPDF. + TX_QUEUE_AVAILABLE = 0x02 + + class Bool(basic.enum8): # Boolean type with values true and false. diff --git a/bellows/types/struct.py b/bellows/types/struct.py index 1253f1d5a..929b542d4 100644 --- a/bellows/types/struct.py +++ b/bellows/types/struct.py @@ -1,6 +1,7 @@ from __future__ import annotations from zigpy.types import Struct as EzspStruct, StructField +from zigpy.zgp.types import ApplicationID, SrcID from . import basic, named @@ -371,14 +372,23 @@ class EmberTokTypeStackZllSecurity(EzspStruct): class EmberGpAddress(EzspStruct): - # A GP address structure. - # The GPD's EUI64. - gpdIeeeAddress: named.EUI64 - # The GPD's source ID. - sourceId: basic.uint32_t - # The GPD Application ID. + """sl_zigbee_gp_address_t: GPD address used by the GP callbacks.""" + applicationId: basic.uint8_t - # The GPD endpoint. + + # 8-byte union. When applicationId == 0 (SrcID), the first 4 bytes hold the 32-bit + # source ID (little-endian) and the rest is padding; when applicationId == 2 + # (IEEE), the 8 bytes are the GPD EUI64. + source_id: SrcID = StructField( + requires=lambda a: a.applicationId == ApplicationID.SrcID + ) + source_id_padding: basic.uint32_t = StructField( + requires=lambda a: a.applicationId == ApplicationID.SrcID + ) + gpd_eui64: named.EUI64 = StructField( + requires=lambda a: a.applicationId == ApplicationID.IEEE + ) + endpoint: basic.uint8_t @@ -498,8 +508,6 @@ class EmberGpSinkListEntry(EzspStruct): class EmberGpProxyTableEntry(EzspStruct): """The internal representation of a proxy table entry.""" - # The link key to be used to secure this pairing link. - securityLinkKey: named.KeyData # Internal status of the proxy table entry. status: named.EmberGpProxyTableEntryStatus # The tunneling options diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py index 11b3eb0f2..a638cb5df 100644 --- a/bellows/zigbee/application.py +++ b/bellows/zigbee/application.py @@ -22,6 +22,7 @@ import zigpy.types import zigpy.util import zigpy.zdo.types as zdo_t +import zigpy.zgp.types as zgp_t import bellows from bellows.config import ( @@ -80,6 +81,23 @@ LIB_VERSION = importlib.metadata.version("bellows") LOGGER = logging.getLogger(__name__) +GP_SECURITY_STATUSES = { + # The NCP reports a GPDF it authenticated itself as `SUCCESS`, not `MATCH` + (t.EmberStatus, t.EmberStatus.SUCCESS): zgp_t.SecurityStatus.SecuritySuccess, + (t.EmberStatus, t.EmberStatus.MATCH): zgp_t.SecurityStatus.SecuritySuccess, + (t.EmberStatus, t.EmberStatus.PASS_UNPROCESSED): zgp_t.SecurityStatus.Unprocessed, + (t.EmberStatus, t.EmberStatus.NO_SECURITY): zgp_t.SecurityStatus.NoSecurity, + (t.EmberStatus, t.EmberStatus.COUNTER_FAILURE): zgp_t.SecurityStatus.CounterFailure, + (t.EmberStatus, t.EmberStatus.AUTH_FAILURE): zgp_t.SecurityStatus.AuthFailure, + (t.EmberStatus, t.EmberStatus.UNPROCESSED): zgp_t.SecurityStatus.Unprocessed, + (t.sl_GpStatus, t.sl_GpStatus.OK): zgp_t.SecurityStatus.SecuritySuccess, + (t.sl_GpStatus, t.sl_GpStatus.MATCH): zgp_t.SecurityStatus.SecuritySuccess, + (t.sl_GpStatus, t.sl_GpStatus.UNPROCESSED): zgp_t.SecurityStatus.Unprocessed, + (t.sl_GpStatus, t.sl_GpStatus.PASS_UNPROCESSED): zgp_t.SecurityStatus.Unprocessed, + (t.sl_GpStatus, t.sl_GpStatus.NO_SECURITY): zgp_t.SecurityStatus.NoSecurity, + (t.sl_GpStatus, t.sl_GpStatus.AUTH_FAILURE): zgp_t.SecurityStatus.AuthFailure, +} + class ControllerApplication(zigpy.application.ControllerApplication): SCHEMA = CONFIG_SCHEMA @@ -606,6 +624,10 @@ async def write_network_info( async def reset_network_info(self): await self._ezsp.factory_reset() + # Green Power tables left by another stack would make the NCP forward and + # security process GPDFs on its own, zigpy keeps the Sink Table itself + await self._ezsp.clear_gp_tables() + # Reset the custom EUI64 await self._ezsp.reset_custom_eui64() @@ -675,6 +697,38 @@ def ezsp_callback_handler(self, frame_name, args): address_index=address_index, message=message, ) + elif frame_name == "gpepIncomingMessageHandler": + ( + status, + gpdLink, + sequenceNumber, + addr, + gpdfSecurityLevel, + gpdfSecurityKeyType, + autoCommissioning, + bidirectionalInfo, + gpdSecurityFrameCounter, + gpdCommandId, + mic, + proxyTableIndex, + gpdCommandPayload, + ) = args + + self._handle_gp_frame( + status, + gpdLink, + sequenceNumber, + addr, + gpdfSecurityLevel, + gpdfSecurityKeyType, + autoCommissioning, + bidirectionalInfo, + gpdSecurityFrameCounter, + gpdCommandId, + mic, + proxyTableIndex, + gpdCommandPayload, + ) elif frame_name == "messageSentHandler": if self._ezsp.ezsp_version >= 14: ( @@ -770,6 +824,96 @@ def _handle_frame( ) ) + def _handle_gp_frame( + self, + status: t.sl_GpStatus, + gpd_link: zgp_t.GPPGPDLink, + sequence_number: t.uint8_t, + addr: t.EmberGpAddress, + gpdf_security_level: t.EmberGpSecurityLevel, + gpdf_security_key_type: t.EmberGpKeyType, + auto_commissioning: t.Bool, + bidirectional_info: t.EmberGpBidirectionalInfo, + gpd_security_frame_counter: t.uint32_t, + gpd_command_id: t.uint8_t, + mic: t.uint32_t, + proxy_table_index: t.uint8_t, + gpd_command_payload: t.LVBytes, + ) -> None: + security_status = GP_SECURITY_STATUSES[type(status), status] + + if security_status in ( + zgp_t.SecurityStatus.AuthFailure, + zgp_t.SecurityStatus.CounterFailure, + ): + LOGGER.debug( + "Dropping GP frame that failed security processing: %s", status + ) + return + + security_level = zgp_t.SecurityLevel(gpdf_security_level) + unprocessed = security_status is zgp_t.SecurityStatus.Unprocessed + + self.gp_packet_received( + zigpy.types.ZigbeeGpPacket( + application_id=zgp_t.ApplicationID(addr.applicationId), + src_id=( + addr.source_id + if addr.applicationId == zgp_t.ApplicationID.SrcID + else None + ), + ieee=( + addr.gpd_eui64 + if addr.applicationId == zgp_t.ApplicationID.IEEE + else None + ), + # The endpoint is only sent by the NCP for IEEE-addressed GPDs + endpoint=( + addr.endpoint + if addr.applicationId == zgp_t.ApplicationID.IEEE + else None + ), + command_id=zgp_t.GPDCommandID(gpd_command_id), + payload=zigpy.types.SerializableBytes(gpd_command_payload), + # An unprotected GPDF has no security frame counter and the NCP reports + # 0xFFFFFFFF, its MAC sequence number takes the place (A.3.3.4.1) + frame_counter=( + t.uint32_t(sequence_number) + if security_level is zgp_t.SecurityLevel.NoSecurity + else gpd_security_frame_counter + ), + security_level=security_level, + # The NCP only knows the key type if it performed the security + # processing itself (spec Table 5). Otherwise this field carries the + # 1-bit `SecurityKey` sub-field off the air, which is not a key type: + # an `IndividualKey` GPD shows up here as `NWKKey`. An unprotected GPDF + # has no key, which zigpy expects as `NoKey`. + security_key_type=( + zgp_t.SecurityKeyType(gpdf_security_key_type) + if security_status is zgp_t.SecurityStatus.SecuritySuccess + else ( + zgp_t.SecurityKeyType.NoKey + if security_status is zgp_t.SecurityStatus.NoSecurity + else None + ) + ), + security_status=security_status, + # A GPDF the NCP did not process is verified and decrypted by zigpy, + # which needs the MIC and the header fields it authenticates + mic=mic if unprocessed else None, + auto_commissioning=bool(auto_commissioning), + rx_after_tx=( + t.EmberGpBidirectionalInfo.RX_AFTER_TX in bidirectional_info + ), + individual_key=bool(gpdf_security_key_type) if unprocessed else None, + # The NCP quantizes the LQI into the two-bit link quality of the GPP-GPD + # link field with the link cost thresholds of the stack (<48, <64, <80, + # and above), so only the band is known. Report its upper bound. + lqi=[47, 63, 79, 255][gpd_link.link_quality], + rssi=gpd_link.rssi_dbm, + ) + ) + def _handle_frame_sent( self, message_type: t.EmberIncomingMessageType, @@ -1027,6 +1171,18 @@ async def send_packet(self, packet: zigpy.types.ZigbeePacket) -> None: if zigpy.types.TransmitOptions.APS_Encryption in packet.tx_options: aps_frame.options |= t.EmberApsOption.APS_OPTION_ENCRYPTION + if packet.alias_src is not None: + if packet.dst.addr_mode not in ( + zigpy.types.AddrMode.Group, + zigpy.types.AddrMode.Broadcast, + ): + raise ValueError( + f"Only broadcasts and multicasts can be sent with an alias: {packet}" + ) + + # The alias NWK sequence number is also the APS counter + aps_frame.options |= t.EmberApsOption.APS_OPTION_USE_ALIAS_SEQUENCE_NUMBER + extended_timeout = packet.extended_timeout # EmberZNet requires retrying to enable APS ACKs @@ -1131,6 +1287,8 @@ async def send_packet(self, packet: zigpy.types.ZigbeePacket) -> None: non_member_radius=packet.non_member_radius, message_tag=message_tag, data=data, + alias=packet.alias_src, + alias_seq=packet.alias_seq, ) elif packet.dst.addr_mode == zigpy.types.AddrMode.Broadcast: status, _ = await self._ezsp.send_broadcast( @@ -1140,6 +1298,8 @@ async def send_packet(self, packet: zigpy.types.ZigbeePacket) -> None: message_tag=message_tag, aps_sequence=packet.tsn, data=data, + alias=packet.alias_src, + alias_seq=packet.alias_seq, ) if status != t.sl_Status.OK: diff --git a/pyproject.toml b/pyproject.toml index c7c907386..02c88e477 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,7 +17,7 @@ dependencies = [ "click", "click-log>=0.2.1", "voluptuous", - "zigpy>=2.1.0", + "zigpy>=2.4.0", "serialx>=1.13.0" ] diff --git a/tests/test_application.py b/tests/test_application.py index 4bfce5f05..ed5ce537a 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -11,6 +11,7 @@ import zigpy.exceptions import zigpy.types as zigpy_t import zigpy.zdo.types as zdo_t +import zigpy.zgp.types as zgp_t from bellows.ash import NcpFailure import bellows.config as config @@ -22,6 +23,8 @@ ) import bellows.ezsp as ezsp from bellows.ezsp.v9.commands import GetTokenDataRsp +from bellows.ezsp.v13.commands import COMMANDS as COMMANDS_v13 +from bellows.ezsp.v19.commands import COMMANDS as COMMANDS_v19 from bellows.ezsp.xncp import ( MAX_XNCP_PAYLOAD_LENGTH, FirmwareFeatures, @@ -230,6 +233,7 @@ def form_network(): ) proto.factory_reset = AsyncMock(proto=proto.factory_reset) + proto.clear_gp_tables = AsyncMock(spec=proto.clear_gp_tables) proto.set_extended_timeout = AsyncMock(proto=proto.set_extended_timeout) proto.read_link_keys = MagicMock() @@ -1356,6 +1360,8 @@ async def test_send_packet_broadcast(app, packet): message_tag=MSG_TAG, aps_sequence=packet.tsn, data=b"some data", + alias=None, + alias_seq=None, ) ] @@ -1404,6 +1410,8 @@ async def test_send_packet_broadcast_ignored_delivery_failure(app, packet): message_tag=MSG_TAG, aps_sequence=packet.tsn, data=b"some data", + alias=None, + alias_seq=None, ) ] @@ -1454,12 +1462,90 @@ async def test_send_packet_multicast(app, packet): non_member_radius=packet.non_member_radius, message_tag=MSG_TAG, data=b"some data", + alias=None, + alias_seq=None, ) ] assert len(app._pending_requests) == 0 +@pytest.mark.parametrize( + ("dst", "send", "message_type", "aps_group"), + [ + ( + zigpy_t.AddrModeAddress( + addr_mode=zigpy_t.AddrMode.Broadcast, address=0xFFFD + ), + "send_broadcast", + t.EmberOutgoingMessageType.OUTGOING_BROADCAST_WITH_ALIAS, + 0x0000, + ), + ( + zigpy_t.AddrModeAddress(addr_mode=zigpy_t.AddrMode.Group, address=0x1234), + "send_multicast", + t.EmberOutgoingMessageType.OUTGOING_MULTICAST_WITH_ALIAS, + 0x1234, + ), + ], +) +async def test_send_packet_with_alias(app, packet, dst, send, message_type, aps_group): + """An aliased send uses the alias sequence number as the APS counter too.""" + packet.dst = dst + packet.tx_options = zigpy_t.TransmitOptions.NONE + packet.alias_src = zigpy_t.NWK(0xABCD) + packet.alias_seq = zigpy_t.uint8_t(0x00) + + setattr( + app._ezsp._protocol, + send, + AsyncMock(return_value=(bellows.types.sl_Status.OK, 0x12)), + ) + app.get_sequence = MagicMock(return_value=MSG_TAG) + + asyncio.get_running_loop().call_soon( + app.ezsp_callback_handler, + "messageSentHandler", + list( + dict( + type=message_type, + indexOrDestination=dst.address, + apsFrame=sentinel.aps, + messageTag=MSG_TAG, + status=t.EmberStatus.SUCCESS, + message=b"", + ).values() + ), + ) + + await app.send_packet(packet) + + [sent] = getattr(app._ezsp, send).mock_calls + assert sent.kwargs["alias"] == 0xABCD + assert sent.kwargs["alias_seq"] == 0x00 + assert sent.kwargs["aps_frame"] == t.EmberApsFrame( + profileId=packet.profile_id, + clusterId=packet.cluster_id, + sourceEndpoint=packet.src_ep, + destinationEndpoint=packet.dst_ep, + options=( + t.EmberApsOption.APS_OPTION_ENABLE_ROUTE_DISCOVERY + | t.EmberApsOption.APS_OPTION_USE_ALIAS_SEQUENCE_NUMBER + ), + groupId=aps_group, + sequence=packet.tsn, + ) + + +async def test_send_packet_unicast_with_alias(app, packet): + """Only broadcasts and multicasts can be sent as an alias.""" + packet.alias_src = zigpy_t.NWK(0xABCD) + packet.alias_seq = zigpy_t.uint8_t(0x00) + + with pytest.raises(ValueError, match="alias"): + await app.send_packet(packet) + + def test_is_controller_running(app): ezsp_running = PropertyMock(return_value=False) with patch.object(type(app._ezsp), "is_ezsp_running", ezsp_running): @@ -2459,6 +2545,7 @@ async def test_write_network_info( assert app._ezsp._protocol.write_aps_frame_counter.mock_calls == [ call(zigpy_backup.network_info.tc_link_key.tx_counter) ] + assert app._ezsp._protocol.clear_gp_tables.mock_calls == [call()] assert app._ezsp._protocol.formNetwork.mock_calls == [ call( parameters=t.EmberNetworkParameters( @@ -3021,3 +3108,152 @@ async def test_multicast_group_subscription_xncp(app: ControllerApplication) -> # The multicast table was never touched assert len(app._ezsp._protocol.setMulticastTableEntry.mock_calls) == 0 + + +@pytest.mark.parametrize( + ("frame", "packet"), + [ + # EnOcean PTM-216Z button press, protected but not processed by the NCP + ( + "03da1b0043f4550143f45501d3020100001b000000696fd093d0ff0101", + zigpy_t.ZigbeeGpPacket( + application_id=zgp_t.ApplicationID.SrcID, + src_id=zgp_t.SrcID(0x0155F443), + ieee=None, + endpoint=None, + command_id=zgp_t.GPDCommandID.Press8BitVector, + payload=zigpy_t.SerializableBytes(b"\x01"), + frame_counter=27, + security_level=zgp_t.SecurityLevel.FullFrameCounterAndMIC, + # The NCP did no security processing so it has no key type to + # report: the `1` it sends is the on-air `SecurityKey` bit, not + # `NWKKey` + security_key_type=None, + security_status=zgp_t.SecurityStatus.Unprocessed, + mic=0xD093D06F, + auto_commissioning=False, + rx_after_tx=False, + individual_key=True, + lqi=255, + rssi=-58, + ), + ), + # Moes 3-gang switch commissioning, unprotected: the NCP reports 0xFFFFFFFF + # as the frame counter and the MAC sequence number takes its place + ( + "06da700091e30d6d91e30d6dd300000000ffffffffe0ffffffffff2b0289f3b9acc88b1df3" + "fe4e503c0be422f6f3530a9c4ed0700600007165726837706f78404750303030318d", + zigpy_t.ZigbeeGpPacket( + application_id=zgp_t.ApplicationID.SrcID, + src_id=zgp_t.SrcID(0x6D0DE391), + ieee=None, + endpoint=None, + command_id=zgp_t.GPDCommandID.CommissioningRequest, + payload=zigpy_t.SerializableBytes( + bytes.fromhex( + "0289f3b9acc88b1df3fe4e503c0be422f6f3530a9c4ed07006000071657268" + "37706f78404750303030318d" + ) + ), + frame_counter=0x70, + security_level=zgp_t.SecurityLevel.NoSecurity, + security_key_type=zgp_t.SecurityKeyType.NoKey, + security_status=zgp_t.SecurityStatus.NoSecurity, + mic=None, + auto_commissioning=False, + rx_after_tx=False, + individual_key=None, + lqi=255, + rssi=-58, + ), + ), + ], +) +def test_frame_handler_gp( + app: ControllerApplication, frame: str, packet: zigpy_t.ZigbeeGpPacket +) -> None: + """Test that an incoming GP frame is parsed into a `ZigbeeGpPacket`.""" + args, rest = t.deserialize_dict( + bytes.fromhex(frame), COMMANDS_v19["gpepIncomingMessageHandler"][2] + ) + assert not rest + + with patch.object(app, "gp_packet_received") as gp_packet_received: + app.ezsp_callback_handler("gpepIncomingMessageHandler", list(args.values())) + + assert gp_packet_received.mock_calls == [call(packet)] + + +def test_frame_handler_gp_legacy_success(app: ControllerApplication) -> None: + """A GPDF authenticated by a pre-EZSPv14 NCP is reported as `SUCCESS`.""" + # A captured EnOcean PTM-216Z frame, as the NCP reports it with a keyed proxy table + # entry. Its status is changed from `UNPROCESSED` to `SUCCESS`. + args, rest = t.deserialize_dict( + bytes.fromhex("00d727004af455014af455010002010000270000006a9a5f2634ff0100"), + COMMANDS_v13["gpepIncomingMessageHandler"][2], + ) + assert not rest + assert args["status"] == t.EmberStatus.SUCCESS + + with patch.object(app, "gp_packet_received") as gp_packet_received: + app.ezsp_callback_handler("gpepIncomingMessageHandler", list(args.values())) + + assert gp_packet_received.mock_calls == [ + call( + zigpy_t.ZigbeeGpPacket( + application_id=zgp_t.ApplicationID.SrcID, + src_id=zgp_t.SrcID(0x0155F44A), + ieee=None, + endpoint=None, + command_id=zgp_t.GPDCommandID.Release8BitVector, + payload=zigpy_t.SerializableBytes(b"\x00"), + frame_counter=0x27, + security_level=zgp_t.SecurityLevel.FullFrameCounterAndMIC, + security_key_type=zgp_t.SecurityKeyType.NWKKey, + security_status=zgp_t.SecurityStatus.SecuritySuccess, + mic=None, + auto_commissioning=False, + rx_after_tx=False, + individual_key=None, + lqi=255, + rssi=-64, + ) + ) + ] + + +def test_frame_handler_gp_unprotected_device(app: ControllerApplication) -> None: + """An unprotected GPDF reaches a GPD commissioned without security.""" + device = zigpy.device.GreenPowerDevice( + app, application_id=zgp_t.ApplicationID.SrcID, src_id=zgp_t.SrcID(0x6D0DE391) + ) + device.security_level = zgp_t.SecurityLevel.NoSecurity + device.security_key_type = zgp_t.SecurityKeyType.NoKey + app.devices[device.ieee] = device + + # The Moes 3-gang switch, now sending a toggle instead of a commissioning request + args, rest = t.deserialize_dict( + bytes.fromhex("06da700091e30d6d91e30d6dd300000000ffffffff22ffffffffff00"), + COMMANDS_v19["gpepIncomingMessageHandler"][2], + ) + assert not rest + + app.ezsp_callback_handler("gpepIncomingMessageHandler", list(args.values())) + + assert device.frame_counter == 0x70 + + +def test_frame_handler_gp_security_failure(app: ControllerApplication) -> None: + """A frame the NCP could not authenticate is dropped.""" + # The same EnOcean PTM-216Z frame, reported as an authentication failure + frame = bytes.fromhex("07da1b0043f4550143f45501d3020100001b000000696fd093d0ff0101") + args, rest = t.deserialize_dict( + frame, COMMANDS_v19["gpepIncomingMessageHandler"][2] + ) + assert not rest + assert args["status"] == t.sl_GpStatus.AUTH_FAILURE + + with patch.object(app, "gp_packet_received") as gp_packet_received: + app.ezsp_callback_handler("gpepIncomingMessageHandler", list(args.values())) + + assert len(gp_packet_received.mock_calls) == 0 diff --git a/tests/test_ezsp_v13.py b/tests/test_ezsp_v13.py index 59ae03d6c..3646e4d9a 100644 --- a/tests/test_ezsp_v13.py +++ b/tests/test_ezsp_v13.py @@ -249,3 +249,34 @@ def test_check_key_context(ezsp_f) -> None: {"status": t.sl_Status.OK}, b"", ) + + +@pytest.mark.parametrize( + "payload", + [ + # BJ6716U + "7ccdec0086f8710186f871010000000000ffffffffe0ffffffffff2e02c5f21ce9ae2f9e4f85f15de37c1ccbd943870013911aec1d000004111011121314151617226062636465666768", + # Moes 3-gang switch + "7cd83100c79b05aac79b05aa0000000000ffffffffe0ffffffffff2b0289f35cb8146efae86ac759171d7401c62fd5b7743df6310600007165726837706f7840475030303031d7", + "7fde3a00c79b05aac79b05aa00030100003a060000c7883a7806ff00", + "7fdf4300c79b05aac79b05aa00030100004306000009ac1475c6ff00", + "7fdf4c00c79b05aac79b05aa00030100004c060000a0a8597872ff00", + # EnOcean PTM-216Z (DB) + "7fda22004af455014af45501000201000022000000695b9b778fff0110", + "7fdb23004af455014af455010002010000230000006a47aa5aadff0100", + "7fd924004af455014af45501000201000024000000697daf2c79ff0101", + "7fd725004af455014af455010002010000250000006a7ccc5856ff0100", + "7fd826004af455014af4550100020100002600000069f51855c0ff0104", + "7fd727004af455014af455010002010000270000006a9a5f2634ff0100", + ], +) +def test_gpep_incoming_frame(payload: str, ezsp_f) -> None: + """Parse incoming GP frames.""" + + ezsp_frame = ( + bytes([0x42, 0x00, 0x01]) # seq + control bytes + + t.uint16_t(0x00C5).serialize() # frame_id LE + + bytes.fromhex(payload) + ) + + ezsp_f(ezsp_frame) diff --git a/tests/test_ezsp_v14.py b/tests/test_ezsp_v14.py index 7bfcfa903..64f818eb1 100644 --- a/tests/test_ezsp_v14.py +++ b/tests/test_ezsp_v14.py @@ -571,6 +571,33 @@ async def test_send_multicast(ezsp_f) -> None: ] +async def test_send_multicast_with_alias(ezsp_f) -> None: + ezsp_f.sendMulticast.return_value = (t.sl_Status.OK, 0x0042) + status, message_tag = await ezsp_f.send_multicast( + aps_frame=t.EmberApsFrame(sequence=0x34), + radius=12, + non_member_radius=34, + message_tag=0x42, + data=b"hello", + alias=t.NWK(0xABCD), + alias_seq=t.uint8_t(0x56), + ) + + assert status == t.sl_Status.OK + assert message_tag == 0x42 + assert ezsp_f.sendMulticast.mock_calls == [ + call( + aps_frame=t.EmberApsFrame(sequence=0x34), + hops=12, + broadcast_addr=t.BroadcastAddress.RX_ON_WHEN_IDLE, + alias=0xABCD, + sequence=0x56, + message_tag=0x0042, + message=b"hello", + ) + ] + + async def test_send_broadcast(ezsp_f) -> None: ezsp_f.sendBroadcast.return_value = (t.sl_Status.OK, 0x0042) status, message_tag = await ezsp_f.send_broadcast( @@ -747,3 +774,31 @@ async def test_set_extended_timeout_no_address_table( ) ] assert ezsp_f.replaceAddressTableEntry.mock_calls == [] + + +async def test_send_broadcast_with_alias(ezsp_f) -> None: + ezsp_f.sendBroadcast.return_value = (t.sl_Status.OK, 0x0042) + status, message_tag = await ezsp_f.send_broadcast( + address=t.BroadcastAddress.RX_ON_WHEN_IDLE, + aps_frame=t.EmberApsFrame(), + radius=12, + message_tag=0x42, + aps_sequence=34, + data=b"hello", + alias=t.NWK(0xABCD), + alias_seq=t.uint8_t(0x56), + ) + + assert status == t.sl_Status.OK + assert message_tag == 0x42 + assert ezsp_f.sendBroadcast.mock_calls == [ + call( + alias=0xABCD, + destination=t.BroadcastAddress.RX_ON_WHEN_IDLE, + sequence=0x56, + aps_frame=t.EmberApsFrame(), + radius=12, + message_tag=0x42, + message=b"hello", + ) + ] diff --git a/tests/test_ezsp_v17.py b/tests/test_ezsp_v17.py index 8c7758681..4dec8b43a 100644 --- a/tests/test_ezsp_v17.py +++ b/tests/test_ezsp_v17.py @@ -1,9 +1,10 @@ -from unittest.mock import MagicMock +from unittest.mock import MagicMock, call import pytest import zigpy.exceptions import zigpy.state +from bellows.exception import InvalidCommandError import bellows.ezsp.v17 import bellows.types as t @@ -31,3 +32,22 @@ def test_ezsp_frame_rx(ezsp_f): assert ezsp_f._handle_callback.call_count == 1 assert ezsp_f._handle_callback.call_args[0][0] == "version" assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234] + + +async def test_clear_gp_tables(ezsp_f) -> None: + ezsp_f.gpSinkTableClearAll.return_value = [] + ezsp_f.gpClearProxyTable.return_value = [] + + await ezsp_f.clear_gp_tables() + + assert ezsp_f.gpSinkTableClearAll.mock_calls == [call()] + assert ezsp_f.gpClearProxyTable.mock_calls == [call()] + + +async def test_clear_gp_tables_unsupported(ezsp_f, caplog) -> None: + ezsp_f.gpSinkTableClearAll.side_effect = InvalidCommandError("No GP") + + await ezsp_f.clear_gp_tables() + + assert len(ezsp_f.gpClearProxyTable.mock_calls) == 0 + assert "no Green Power tables" in caplog.text diff --git a/tests/test_ezsp_v19.py b/tests/test_ezsp_v19.py index 5c716750b..45579f370 100644 --- a/tests/test_ezsp_v19.py +++ b/tests/test_ezsp_v19.py @@ -48,3 +48,35 @@ def test_get_token_info(version: int) -> None: assert result["token_info"].isIdx == t.Bool.true assert result["token_info"].size == 300 assert result["token_info"].arraySize == 5 + + +@pytest.mark.parametrize( + "payload", + [ + # Moes 3-gang switch + "06da700091e30d6d91e30d6dd300000000ffffffffe0ffffffffff2b0289f3b9acc88b1df3fe4e503c0be422f6f3530a9c4ed0700600007165726837706f78404750303030318d", + "03dd790091e30d6d91e30d6dd3030100007906000052e6f26ed3ff00", + # EnOcean PTM-216Z (DB) + "03da1b0043f4550143f45501d3020100001b000000696fd093d0ff0101", + # EnOcean PTM-216Z, direct commissioning on channel 11 (long press B0): + # the commissioning telegram, DeviceID GenericSwitch with switch information + "06dd760043f4550143f45501d300000000ffffffffe0ffffffffff1f0785f26761efc264c893701987babda7238ce9102e934a7600000010020504", + # the AI+B0 radio channel confirmation, sent as ShortPress2of2 with no payload + "03dc770043f4550143f45501d302010000770000006875db8e85ff00", + # subsequent press/release pairs, one per rocker contact + "03db780043f4550143f45501d30201000078000000692a661d3eff0101", + "03da790043f4550143f45501d302010000790000006a3861bf27ff0100", + "03d67c0043f4550143f45501d3020100007c0000006977b4fb0bff0102", + "03d87e0043f4550143f45501d3020100007e0000006999877c4dff0106", + ], +) +def test_gpep_incoming_frame(payload: str, ezsp_f) -> None: + """Parse incoming GP frames.""" + + ezsp_frame = ( + bytes([0x42, 0x00, 0x01]) # seq + control bytes + + t.uint16_t(0x00C5).serialize() # frame_id LE + + bytes.fromhex(payload) + ) + + ezsp_f(ezsp_frame) diff --git a/tests/test_ezsp_v4.py b/tests/test_ezsp_v4.py index 7a8826f8c..38eec1908 100644 --- a/tests/test_ezsp_v4.py +++ b/tests/test_ezsp_v4.py @@ -368,6 +368,50 @@ async def test_send_broadcast(ezsp_f) -> None: ] +async def test_send_multicast_with_alias(ezsp_f) -> None: + with pytest.raises(NotImplementedError): + await ezsp_f.send_multicast( + aps_frame=t.EmberApsFrame(), + radius=12, + non_member_radius=34, + message_tag=0x42, + data=b"hello", + alias=t.NWK(0xABCD), + alias_seq=t.uint8_t(0x56), + ) + + assert len(ezsp_f.sendMulticast.mock_calls) == 0 + + +async def test_send_broadcast_with_alias(ezsp_f) -> None: + ezsp_f.proxyBroadcast.return_value = (t.EmberStatus.SUCCESS, 0x42) + status, message_tag = await ezsp_f.send_broadcast( + address=t.BroadcastAddress.RX_ON_WHEN_IDLE, + aps_frame=t.EmberApsFrame(), + radius=12, + message_tag=0x42, + aps_sequence=34, + data=b"hello", + alias=t.NWK(0xABCD), + alias_seq=t.uint8_t(0x56), + ) + + assert status == t.sl_Status.OK + assert message_tag == 0x42 + assert len(ezsp_f.sendBroadcast.mock_calls) == 0 + assert ezsp_f.proxyBroadcast.mock_calls == [ + call( + source=0xABCD, + destination=t.BroadcastAddress.RX_ON_WHEN_IDLE, + nwkSequence=0x56, + apsFrame=t.EmberApsFrame(), + radius=12, + messageTag=0x42, + messageContents=b"hello", + ) + ] + + async def test_source_route(ezsp_f) -> None: ezsp_f.setSourceRoute.return_value = (t.EmberStatus.SUCCESS,) @@ -562,3 +606,8 @@ async def send_data(data: bytes) -> None: # All but the first queue up assert caplog.text.count("Send semaphore is locked, delaying before sending") == 3 assert caplog.text.count("s delay") == 3 + + +async def test_clear_gp_tables(ezsp_f) -> None: + """There is no Green Power table clearing before EZSPv6.""" + await ezsp_f.clear_gp_tables() diff --git a/tests/test_ezsp_v6.py b/tests/test_ezsp_v6.py index 228f3368b..c6728d781 100644 --- a/tests/test_ezsp_v6.py +++ b/tests/test_ezsp_v6.py @@ -3,6 +3,7 @@ import pytest import zigpy.state +from bellows.exception import InvalidCommandError import bellows.ezsp.v6 import bellows.types as t @@ -38,3 +39,65 @@ async def test_initialize_network(ezsp_f) -> None: assert ezsp_f.networkInit.mock_calls == [ call(networkInitBitmask=t.EmberNetworkInitBitmask.NETWORK_INIT_NO_OPTIONS) ] + + +async def test_send_multicast_with_alias(ezsp_f) -> None: + ezsp_f.sendMulticastWithAlias.return_value = (t.EmberStatus.SUCCESS, 0x42) + status, message_tag = await ezsp_f.send_multicast( + aps_frame=t.EmberApsFrame(), + radius=12, + non_member_radius=34, + message_tag=0x42, + data=b"hello", + alias=t.NWK(0xABCD), + alias_seq=t.uint8_t(0x56), + ) + + assert status == t.sl_Status.OK + assert message_tag == 0x42 + assert len(ezsp_f.sendMulticast.mock_calls) == 0 + assert ezsp_f.sendMulticastWithAlias.mock_calls == [ + call( + apsFrame=t.EmberApsFrame(), + hops=12, + nonmemberRadius=34, + alias=0xABCD, + nwkSequence=0x56, + messageTag=0x42, + messageContents=b"hello", + ) + ] + + +async def test_send_multicast_without_alias(ezsp_f) -> None: + ezsp_f.sendMulticast.return_value = (t.EmberStatus.SUCCESS, 0x42) + status, message_tag = await ezsp_f.send_multicast( + aps_frame=t.EmberApsFrame(), + radius=12, + non_member_radius=34, + message_tag=0x42, + data=b"hello", + ) + + assert status == t.sl_Status.OK + assert message_tag == 0x42 + assert len(ezsp_f.sendMulticastWithAlias.mock_calls) == 0 + assert len(ezsp_f.sendMulticast.mock_calls) == 1 + + +async def test_clear_gp_tables(ezsp_f) -> None: + """Only the sink table has a bulk clear before EZSPv17.""" + ezsp_f.gpClearSinkTable.return_value = [] + + await ezsp_f.clear_gp_tables() + + assert ezsp_f.gpClearSinkTable.mock_calls == [call()] + + +async def test_clear_gp_tables_unsupported(ezsp_f, caplog) -> None: + """An NCP built without Green Power rejects the command.""" + ezsp_f.gpClearSinkTable.side_effect = InvalidCommandError("No GP") + + await ezsp_f.clear_gp_tables() + + assert "no Green Power sink table" in caplog.text diff --git a/tests/test_ezsp_v7.py b/tests/test_ezsp_v7.py index d8819c53e..a0ee97797 100644 --- a/tests/test_ezsp_v7.py +++ b/tests/test_ezsp_v7.py @@ -1,8 +1,9 @@ -from unittest.mock import MagicMock +from unittest.mock import MagicMock, call import pytest from bellows.ash import DataFrame +from bellows.exception import InvalidCommandError import bellows.ezsp.v7 import bellows.types as t @@ -61,3 +62,19 @@ def get_child_data(index): t.EmberNodeType.SLEEPY_END_DEVICE, ) ] + + +async def test_clear_gp_tables(ezsp_f) -> None: + ezsp_f.gpSinkTableClearAll.return_value = [] + + await ezsp_f.clear_gp_tables() + + assert ezsp_f.gpSinkTableClearAll.mock_calls == [call()] + + +async def test_clear_gp_tables_unsupported(ezsp_f, caplog) -> None: + ezsp_f.gpSinkTableClearAll.side_effect = InvalidCommandError("No GP") + + await ezsp_f.clear_gp_tables() + + assert "no Green Power sink table" in caplog.text diff --git a/tests/test_ezsp_v9.py b/tests/test_ezsp_v9.py index 1412404f7..9219282a1 100644 --- a/tests/test_ezsp_v9.py +++ b/tests/test_ezsp_v9.py @@ -89,3 +89,40 @@ async def test_source_route(ezsp_f) -> None: # Nothing happens assert ezsp_f.setSourceRoute.mock_calls == [] + + +@pytest.mark.parametrize( + ("payload", "src_id"), + [ + # EnOcean PTM 215Z: commissioning, then presses + ( + "7cdf1b00e61c7701e61c77010000000000ffffffffe0ffffffffff2e02c5f210405e0513cb" + "556a00ee94019ee66218118edec91b03000004111011121314151617226062636465666768", + 0x01771CE6, + ), + ("7fe11c00e61c7701e61c770100020100001c03000068e3ff0356ff00", 0x01771CE6), + ("7fe11d00e61c7701e61c770100020100001d03000012be6d0773ff00", 0x01771CE6), + ("7fe22500e61c7701e61c770100020100002503000064b4e65552ff00", 0x01771CE6), + # Moes 3-gang switch: commissioning, then encrypted presses + ( + "7cdc7400425374f3425374f30000000000ffffffffe0ffffffffff2b0289f3a86f2ed35956" + "0f929469d868472c2639e1be6d66741a00007165726837706f78404750303030318e", + 0xF3745342, + ), + ("7fe97d00425374f3425374f300030100007d1a000045dc3f907bff00", 0xF3745342), + ("7fe98600425374f3425374f30003010000861a000088f258ed78ff00", 0xF3745342), + ], +) +def test_gpep_incoming_frame(payload: str, src_id: int, ezsp_f) -> None: + """Parse incoming GP frames.""" + ezsp_frame = ( + bytes([0x42, 0x00, 0x01]) # seq + control bytes + + t.uint16_t(0x00C5).serialize() # frame_id LE + + bytes.fromhex(payload) + ) + + ezsp_f(ezsp_frame) + + [(name, args, *_)] = [c.args for c in ezsp_f._handle_callback.mock_calls] + assert name == "gpepIncomingMessageHandler" + assert args[3].source_id == src_id