diff --git a/bellows/ezsp/__init__.py b/bellows/ezsp/__init__.py index a21138292..6992ea5b8 100644 --- a/bellows/ezsp/__init__.py +++ b/bellows/ezsp/__init__.py @@ -5,7 +5,7 @@ import asyncio from asyncio import timeout as asyncio_timeout import collections -from collections.abc import Callable, Generator +from collections.abc import AsyncIterator, Callable, Generator import contextlib import dataclasses import functools @@ -18,6 +18,7 @@ from bellows.ash import NcpFailure import bellows.config as conf from bellows.exception import ( + ControllerError, EzspError, InvalidCommandError, InvalidCommandPayload, @@ -52,6 +53,9 @@ NETWORK_OPS_TIMEOUT = 10 NETWORK_COORDINATOR_STARTUP_RESET_WAIT = 2 +WILDCARD_EUI64 = t.EUI64.convert("FF:FF:FF:FF:FF:FF:FF:FF") +WELL_KNOWN_LINK_KEY = t.KeyData(b"ZigBeeAlliance09") + class EZSP: _BY_VERSION = { @@ -628,6 +632,37 @@ async def set_source_routing(self, enabled: bool) -> None: ) ) + @contextlib.asynccontextmanager + async def allow_joins( + self, *, link_keys: dict[t.EUI64, t.KeyData] | None = None + ) -> AsyncIterator[None]: + """Allow joins with the well-known key, or only with the given link keys.""" + install_codes_only = link_keys is not None + + if link_keys is None: + link_keys = {WILDCARD_EUI64: WELL_KNOWN_LINK_KEY} + + try: + # Keys from a previous window may remain if its cleanup failed + await self._protocol.clear_transient_link_keys() + + for ieee, key in link_keys.items(): + status = await self._protocol.add_transient_link_key(ieee, key) + + if status != t.sl_Status.OK: + raise ControllerError(f"Failed to set link key for {ieee}") + + await self._protocol.set_join_policy( + allow_joins=True, install_codes_only=install_codes_only + ) + + yield + finally: + try: + await self._protocol.clear_transient_link_keys() + finally: + await self._protocol.set_join_policy(allow_joins=False) + def start_ezsp(self): """Mark EZSP as running.""" self._ezsp_event.set() diff --git a/bellows/ezsp/protocol.py b/bellows/ezsp/protocol.py index 841c1b3a8..b1b2d5b79 100644 --- a/bellows/ezsp/protocol.py +++ b/bellows/ezsp/protocol.py @@ -283,8 +283,11 @@ def __getattr__(self, name: str) -> Callable: return functools.partial(self.command, name) - async def pre_permit(self, time_s: int) -> None: - """Schedule task before allowing new joins.""" + @abc.abstractmethod + async def set_join_policy( + self, *, allow_joins: bool, install_codes_only: bool = False + ) -> None: + raise NotImplementedError @abc.abstractmethod async def add_transient_link_key( @@ -292,6 +295,10 @@ async def add_transient_link_key( ) -> t.sl_Status: raise NotImplementedError + @abc.abstractmethod + async def clear_transient_link_keys(self) -> None: + raise NotImplementedError + @abc.abstractmethod async def read_child_data( self, diff --git a/bellows/ezsp/v10/config.py b/bellows/ezsp/v10/config.py index d77116c0f..de47bb8c2 100644 --- a/bellows/ezsp/v10/config.py +++ b/bellows/ezsp/v10/config.py @@ -3,7 +3,7 @@ from bellows.config import cv_uint16 from bellows.types import EzspPolicyId -from ..v4.config import EZSP_POLICIES_SHARED +from ..v8.config import EZSP_POLICIES_SHARED from ..v9 import config as v9_config EZSP_SCHEMA = { diff --git a/bellows/ezsp/v11/config.py b/bellows/ezsp/v11/config.py index 995269353..3f969d74b 100644 --- a/bellows/ezsp/v11/config.py +++ b/bellows/ezsp/v11/config.py @@ -3,7 +3,7 @@ from bellows.config import cv_uint16 from bellows.types import EzspPolicyId -from ..v4.config import EZSP_POLICIES_SHARED +from ..v8.config import EZSP_POLICIES_SHARED from ..v10 import config as v10_config EZSP_SCHEMA = { diff --git a/bellows/ezsp/v12/config.py b/bellows/ezsp/v12/config.py index 36688809f..992da2497 100644 --- a/bellows/ezsp/v12/config.py +++ b/bellows/ezsp/v12/config.py @@ -3,7 +3,7 @@ from bellows.config import cv_uint16 from bellows.types import EzspPolicyId -from ..v4.config import EZSP_POLICIES_SHARED +from ..v8.config import EZSP_POLICIES_SHARED from ..v11 import config as v11_config EZSP_SCHEMA = { diff --git a/bellows/ezsp/v13/config.py b/bellows/ezsp/v13/config.py index f71825db3..396545567 100644 --- a/bellows/ezsp/v13/config.py +++ b/bellows/ezsp/v13/config.py @@ -3,7 +3,7 @@ from bellows.config import cv_uint16 from bellows.types import EzspPolicyId -from ..v4.config import EZSP_POLICIES_SHARED +from ..v8.config import EZSP_POLICIES_SHARED from ..v12 import config as v12_config EZSP_SCHEMA = { diff --git a/bellows/ezsp/v14/config.py b/bellows/ezsp/v14/config.py index c56d354c0..8d16a3c34 100644 --- a/bellows/ezsp/v14/config.py +++ b/bellows/ezsp/v14/config.py @@ -3,7 +3,7 @@ from bellows.config import cv_uint16 from bellows.types import EzspPolicyId -from ..v4.config import EZSP_POLICIES_SHARED +from ..v8.config import EZSP_POLICIES_SHARED from ..v13 import config as v13_config EZSP_SCHEMA = { diff --git a/bellows/ezsp/v16/config.py b/bellows/ezsp/v16/config.py index d502f1ec1..77c9c35a1 100644 --- a/bellows/ezsp/v16/config.py +++ b/bellows/ezsp/v16/config.py @@ -3,7 +3,7 @@ from bellows.config import cv_uint16 from bellows.types import EzspPolicyId -from ..v4.config import EZSP_POLICIES_SHARED +from ..v8.config import EZSP_POLICIES_SHARED from ..v14 import config as v14_config EZSP_SCHEMA = { diff --git a/bellows/ezsp/v17/config.py b/bellows/ezsp/v17/config.py index 45bf59e5d..64891d973 100644 --- a/bellows/ezsp/v17/config.py +++ b/bellows/ezsp/v17/config.py @@ -3,7 +3,7 @@ from bellows.config import cv_uint16 from bellows.types import EzspPolicyId -from ..v4.config import EZSP_POLICIES_SHARED +from ..v8.config import EZSP_POLICIES_SHARED from ..v16 import config as v16_config EZSP_SCHEMA = { diff --git a/bellows/ezsp/v18/config.py b/bellows/ezsp/v18/config.py index 167fa57aa..c4873e00c 100644 --- a/bellows/ezsp/v18/config.py +++ b/bellows/ezsp/v18/config.py @@ -3,7 +3,7 @@ from bellows.config import cv_uint16 from bellows.types import EzspPolicyId -from ..v4.config import EZSP_POLICIES_SHARED +from ..v8.config import EZSP_POLICIES_SHARED from ..v17 import config as v17_config EZSP_SCHEMA = { diff --git a/bellows/ezsp/v19/config.py b/bellows/ezsp/v19/config.py index 87252f279..f16469ab3 100644 --- a/bellows/ezsp/v19/config.py +++ b/bellows/ezsp/v19/config.py @@ -3,7 +3,7 @@ from bellows.config import cv_uint16 from bellows.types import EzspPolicyId -from ..v4.config import EZSP_POLICIES_SHARED +from ..v8.config import EZSP_POLICIES_SHARED from ..v18 import config as v18_config EZSP_SCHEMA = { diff --git a/bellows/ezsp/v4/__init__.py b/bellows/ezsp/v4/__init__.py index abbde484b..9f94bae6a 100644 --- a/bellows/ezsp/v4/__init__.py +++ b/bellows/ezsp/v4/__init__.py @@ -36,7 +36,9 @@ def _ezsp_frame_rx(self, data: bytes) -> tuple[int, int, bytes]: """Handler for received data frame.""" return data[0], data[2], data[3:] - async def pre_permit(self, time_s: int) -> None: + async def set_join_policy( + self, *, allow_joins: bool, install_codes_only: bool = False + ) -> None: pass async def add_transient_link_key( @@ -44,6 +46,9 @@ async def add_transient_link_key( ) -> t.sl_Status: return t.sl_Status.OK + async def clear_transient_link_keys(self) -> None: + pass + async def read_child_data( self, ) -> AsyncGenerator[tuple[t.NWK, t.EUI64, t.EmberNodeType]]: diff --git a/bellows/ezsp/v5/__init__.py b/bellows/ezsp/v5/__init__.py index 0a741a120..aeb2bc106 100644 --- a/bellows/ezsp/v5/__init__.py +++ b/bellows/ezsp/v5/__init__.py @@ -41,11 +41,8 @@ async def add_transient_link_key( (status,) = await self.addTransientLinkKey(partner=ieee, transientKey=key) return t.sl_Status.from_ember_status(status) - async def pre_permit(self, time_s: int) -> None: - """Add pre-shared TC Link key.""" - wild_card_ieee = t.EUI64.convert("FF:FF:FF:FF:FF:FF:FF:FF") - tc_link_key = t.KeyData(b"ZigBeeAlliance09") - await self.add_transient_link_key(wild_card_ieee, tc_link_key) + async def clear_transient_link_keys(self) -> None: + await self.clearTransientLinkKeys() async def read_address_table(self) -> AsyncGenerator[tuple[t.NWK, t.EUI64]]: (status, addr_table_size) = await self.getConfigurationValue( diff --git a/bellows/ezsp/v8/__init__.py b/bellows/ezsp/v8/__init__.py index b2b55dde2..ea7412e89 100644 --- a/bellows/ezsp/v8/__init__.py +++ b/bellows/ezsp/v8/__init__.py @@ -1,5 +1,4 @@ """"EZSP Protocol version 8 protocol handler.""" -import asyncio import logging import voluptuous @@ -36,18 +35,25 @@ def _ezsp_frame_rx(self, data: bytes) -> tuple[int, int, bytes]: return seq, frame_id, data - async def pre_permit(self, time_s: int) -> None: - """Temporarily change TC policy while allowing new joins.""" - await super().pre_permit(time_s) - await self.setPolicy( - policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, - decisionId=( - t.EzspDecisionBitmask.ALLOW_JOINS + async def set_join_policy( + self, *, allow_joins: bool, install_codes_only: bool = False + ) -> None: + if not allow_joins: + decision = self.tc_policy + elif install_codes_only: + decision = ( + self.tc_policy + | t.EzspDecisionBitmask.ALLOW_JOINS | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS - ), - ) - await asyncio.sleep(time_s + 2) + | t.EzspDecisionBitmask.JOINS_USE_INSTALL_CODE_KEY + ) + else: + decision = ( + self.tc_policy + | t.EzspDecisionBitmask.ALLOW_JOINS + | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS + ) + await self.setPolicy( - policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, - decisionId=self.tc_policy, + policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, decisionId=decision ) diff --git a/bellows/ezsp/v8/config.py b/bellows/ezsp/v8/config.py index 7ee0adf03..da38778bc 100644 --- a/bellows/ezsp/v8/config.py +++ b/bellows/ezsp/v8/config.py @@ -1,11 +1,30 @@ import voluptuous as vol from bellows.config import cv_optional_int, cv_uint16, extend_vol_schema -from bellows.types import EzspConfigId, EzspPolicyId +from bellows.types import EzspConfigId, EzspDecisionBitmask, EzspPolicyId -from ..v4.config import EZSP_POLICIES_SHARED +from ..v4 import config as v4_config from ..v7 import config as v7_config +# EZSPv8 replaces the trust center decisions with a decision bitmask +EZSP_POLICIES_SHARED = { + **{ + key: value + for key, value in v4_config.EZSP_POLICIES_SHARED.items() + if key != EzspPolicyId.TRUST_CENTER_POLICY.name + }, + # Same as the NCP's boot default and the SDK host framework. Trust Center rejoins + # are always accepted. New joins still need permit joining and the well-known + # transient key, which only exists while joins are allowed. + vol.Optional( + EzspPolicyId.TRUST_CENTER_POLICY.name, + default=( + EzspDecisionBitmask.ALLOW_JOINS + | EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS + ), + ): cv_uint16, +} + EZSP_SCHEMA = extend_vol_schema( v7_config.EZSP_SCHEMA, { diff --git a/bellows/ezsp/v9/config.py b/bellows/ezsp/v9/config.py index 9c678ff56..95b5a0019 100644 --- a/bellows/ezsp/v9/config.py +++ b/bellows/ezsp/v9/config.py @@ -3,8 +3,8 @@ from bellows.config import cv_uint16 from bellows.types import EzspPolicyId -from ..v4.config import EZSP_POLICIES_SHARED from ..v8 import config as v8_config +from ..v8.config import EZSP_POLICIES_SHARED EZSP_SCHEMA = { **v8_config.EZSP_SCHEMA, diff --git a/bellows/types/named.py b/bellows/types/named.py index 663915576..909361a1f 100644 --- a/bellows/types/named.py +++ b/bellows/types/named.py @@ -996,19 +996,20 @@ class EzspNetworkScanType(basic.enum8): class EmberJoinDecision(basic.enum8): - # Decision made by the trust center when a node attempts to join. + # The decision made by the Trust Center when a node attempts to join. - # Allow the node to join. The joining node should have a pre-configured - # key. The security data sent to it will be encrypted with that key. + # Allow the node to join. The node has the key. USE_PRECONFIGURED_KEY = 0x00 - # Allow the node to join. Send the necessary key (the Network Key in - # Standard Security mode, the Trust Center Master in High Security mode) - # in-the-clear to the joining device. + # Allow the node to join. Send the key to the node. SEND_KEY_IN_THE_CLEAR = 0x01 # Deny join. DENY_JOIN = 0x02 # Take no action. NO_ACTION = 0x03 + # Allow rejoins only. + # Note: only used by the SDK host framework, the NCP never reports it in + # `trustCenterJoinHandler`. + ALLOW_REJOINS_ONLY = 0x04 class EmberInitialSecurityBitmask(basic.bitmap16): @@ -2449,7 +2450,10 @@ class EzspPolicyId(basic.enum8): ZLL_POLICY = 0x08 # Controls whether Trust Center (insecure) rejoins for devices using the well-known # link key are accepted. If rejoining using the well-known key is allowed, it is - # disabled again after emAllowTcRejoinsUsingWellKnownKeyTimeoutSec seconds. + # disabled again after + # sli_zigbee_allow_tc_rejoins_using_well_known_key_timeout_sec seconds. + # Note: the timeout is set with `CONFIG_TC_REJOINS_USING_WELL_KNOWN_KEY_TIMEOUT_S`. + # A timeout of 0 never disables it. TC_REJOINS_USING_WELL_KNOWN_KEY_POLICY = 0x09 # Controls whether the ZigBee RF4CE stack will use standard profile-dependent # behavior during the discovery and pairing process. The profiles supported at the @@ -2470,14 +2474,28 @@ class EzspPolicyId(basic.enum8): class EzspDecisionBitmask(basic.bitmap8): """EZSP Decision bitmask.""" + # This is the policy decision bitmask that controls the trust center decision + # strategies. The bitmask is modified and extracted from the + # sl_zigbee_ezsp_decision_id_t for supporting bitmask operations. + # Note: used for `TRUST_CENTER_POLICY` from EZSPv8 onwards, replacing the trust + # center decisions of `EzspDecisionId`. The NCP boots with `ALLOW_JOINS | + # ALLOW_UNSECURED_REJOINS`. + # XXX: this is intentionally an 8-bit bitmap! The SDK source treats it as a 16-bit # value but it is always serialized to an 8-bit integer # Disallow joins and rejoins. DEFAULT_CONFIGURATION = 0x0000 # Send the network key to all joining devices. + # Note: EmberZNet 6.7.x (EZSPv8) also accepts unsecured rejoins with only this bit + # set. ALLOW_JOINS = 0x0001 # Send the network key to all rejoining devices. + # Note: from EmberZNet 6.9.0 (EZSPv8) onwards, unsecured rejoins are denied without + # it. + # Note: these are Trust Center rejoins (spec name). The network key is APS + # encrypted with the device's link key, devices using the well-known key are + # rejected unless `TC_REJOINS_USING_WELL_KNOWN_KEY_POLICY` is enabled. ALLOW_UNSECURED_REJOINS = 0x0002 # Send the network key in the clear. SEND_KEY_IN_CLEAR = 0x0004 @@ -2492,6 +2510,9 @@ class EzspDecisionBitmask(basic.bitmap8): class EzspDecisionId(basic.enum8): # Identifies a policy decision. + # Note: `TRUST_CENTER_POLICY` decisions, EZSPv4 to EZSPv7 only. EZSPv8 and later + # use `EzspDecisionBitmask` instead and no longer define these values. + # Send the network key in the clear to all joining and rejoining devices. ALLOW_JOINS = 0x00 # Send the network key in the clear to all joining devices. Rejoining @@ -2520,6 +2541,7 @@ class EzspDecisionId(basic.enum8): BDB_JOIN_USES_INSTALL_CODE_KEY = 0x06 # Delay sending the network key to a new joining device. DEFER_JOINS_REJOINS_HAVE_LINK_KEY = 0x07 + # BINDING_MODIFICATION_POLICY default decision. Do not allow the local # binding table to be changed by remote nodes. DISALLOW_BINDING_MODIFICATION = 0x10 @@ -2552,8 +2574,11 @@ class EzspDecisionId(basic.enum8): # TC_KEY_REQUEST_POLICY decision. When the Trust Center receives a request for a # Trust Center link key, it will reply to it with the corresponding key. ALLOW_TC_KEY_REQUESTS_AND_SEND_CURRENT_KEY = 0x51 - # TC_KEY_REQUEST_POLICY decision. When the Trust Center receives a request - # for a Trust Center link key, it will generate a key to send to the joiner. + # SL_ZIGBEE_EZSP_TC_KEY_REQUEST_POLICY decision. When the Trust Center + # receives a request for a Trust Center link key, it will generate a key to + # send to the joiner. After generation, the key will be added to the + # transient key table and After verification, this key will be added into the + # link key table ALLOW_TC_KEY_REQUEST_AND_GENERATE_NEW_KEY = 0x52 # APP_KEY_REQUEST_POLICY decision. When the Trust Center receives a request # for an application link key, it will be ignored. @@ -2590,11 +2615,22 @@ class EzspDecisionId(basic.enum8): class EmberDeviceUpdate(basic.enum8): - # The status of the device update. - + # The Status of the Update Device message sent to the Trust Center. + # The device may have joined or rejoined insecurely, rejoined securely, or + # left. MAC Security has been deprecated and therefore there is no secure + # join. + # These map to the actual values within the APS Command frame so they cannot + # be arbitrarily changed. + # Note: this is the Status field of the APS Update Device command, Zigbee R23.2 + # 4.4.11.2.4 and Table 4-14. The spec reserves 0x04 to 0x07. + + # Note: spec name "Standard Device Secured Rejoin". STANDARD_SECURITY_SECURED_REJOIN = 0x0 + # Note: spec name "Standard Device Unsecured Join". STANDARD_SECURITY_UNSECURED_JOIN = 0x1 DEVICE_LEFT = 0x2 + # Note: spec name "Standard Device Trust Center Rejoin", a NWK rejoin without NWK + # security. STANDARD_SECURITY_UNSECURED_REJOIN = 0x3 # Removed in EZSPv5 diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py index 9fe94ac58..f6996efeb 100644 --- a/bellows/zigbee/application.py +++ b/bellows/zigbee/application.py @@ -3,6 +3,8 @@ import asyncio from asyncio import timeout as asyncio_timeout from collections.abc import AsyncGenerator +import contextlib +import dataclasses from datetime import UTC, datetime import importlib.metadata import logging @@ -99,6 +101,14 @@ } +@dataclasses.dataclass +class JoinWindow: + link_keys: dict[t.EUI64, t.KeyData] | None + close_event: asyncio.Event + timer: asyncio.TimerHandle + task: asyncio.Task + + class ControllerApplication(zigpy.application.ControllerApplication): SCHEMA = CONFIG_SCHEMA @@ -116,6 +126,8 @@ def __init__(self, config: dict) -> None: self._ezsp = None self._multicast = None self._mfg_id_task: asyncio.Task | None = None + self._join_window: JoinWindow | None = None + self._join_window_lock = asyncio.Lock() self._pending_requests = {} self._watchdog_failures = 0 self._watchdog_feed_counter = 0 @@ -1005,7 +1017,12 @@ def _handle_tc_join_handler( if self._mfg_id_task and not self._mfg_id_task.done(): self._mfg_id_task.cancel() self._mfg_id_task = asyncio.create_task(self._reset_mfg_id(mfg_id)) - self.handle_join(nwk, ieee, parent_nwk) + + # Unsecured joins and rejoins are unauthenticated: anyone can send one for any + # EUI64 and the device may never receive the network key. The device's own + # network-secured Device_annce reports the join once it has the key. + if device_update_status == t.EmberDeviceUpdate.STANDARD_SECURITY_SECURED_REJOIN: + self.handle_join(nwk, ieee, parent_nwk) async def _reset_mfg_id(self, mfg_id: int) -> None: """Resets manufacturer id if was temporary overridden by a joining device.""" @@ -1335,10 +1352,61 @@ async def send_packet(self, packet: zigpy.types.ZigbeePacket) -> None: finally: del self._pending_requests[pending_tag] + async def _close_join_window(self) -> None: + if self._join_window is None: + return + + window = self._join_window + self._join_window = None + + window.timer.cancel() + window.close_event.set() + + # The window task always runs to completion so cleanup is never interrupted + await asyncio.wait([window.task]) + + async def _open_join_window( + self, time_s: int, link_keys: dict[t.EUI64, t.KeyData] | None = None + ) -> None: + loop = asyncio.get_running_loop() + window = self._join_window + + # Reopening the window would briefly reject joins, extend it instead + if ( + window is not None + and not window.close_event.is_set() + and window.link_keys == link_keys + ): + window.timer.cancel() + window.timer = loop.call_later(time_s, window.close_event.set) + return + + await self._close_join_window() + + stack = contextlib.AsyncExitStack() + await stack.enter_async_context(self._ezsp.allow_joins(link_keys=link_keys)) + close_event = asyncio.Event() + + async def join_window() -> None: + async with stack: + await close_event.wait() + + self._join_window = JoinWindow( + link_keys=link_keys, + close_event=close_event, + timer=loop.call_later(time_s, close_event.set), + task=self.create_task(join_window(), "join_window"), + ) + async def permit(self, time_s: int = 60, node: t.EmberNodeId = None) -> None: """Permit joining.""" - self.create_task(self._ezsp.pre_permit(time_s), "pre_permit") - await super().permit(time_s, node) + async with self._join_window_lock: + if time_s == 0: + await self._close_join_window() + else: + await self._open_join_window(time_s) + + await super().permit(time_s, node) def permit_ncp(self, time_s=60): assert 0 <= time_s <= 254 @@ -1348,22 +1416,9 @@ async def permit_with_link_key( self, node: t.EUI64, link_key: t.KeyData, time_s: int = 60 ) -> None: """Permits a new device to join with the given IEEE and link key.""" - - status = await self._ezsp.add_transient_link_key(node, link_key) - - if t.sl_Status.from_ember_status(status) != t.sl_Status.OK: - raise ControllerError("Failed to set link key") - - if self._ezsp.ezsp_version >= 8: - await self._ezsp.setPolicy( - policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, - decisionId=( - t.EzspDecisionBitmask.ALLOW_JOINS - | t.EzspDecisionBitmask.JOINS_USE_INSTALL_CODE_KEY - ), - ) - - return await super().permit(time_s) + async with self._join_window_lock: + await self._open_join_window(time_s, link_keys={node: link_key}) + await super().permit(time_s) async def _subscribe_to_multicast_group( self, group_id: zigpy.types.Group, endpoint_id: int diff --git a/tests/test_application.py b/tests/test_application.py index 326ef9733..1a721fce5 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -661,46 +661,55 @@ def test_unexpected_send_success(app, aps, ieee): ) -async def test_join_handler(app, ieee): +@pytest.mark.parametrize( + "status, decision, joins, cleans_up_tclk", + [ + # Only a secured rejoin proves the device holds the network key + ( + t.EmberDeviceUpdate.STANDARD_SECURITY_SECURED_REJOIN, + t.EmberJoinDecision.NO_ACTION, + True, + False, + ), + # Unsecured joins and rejoins are reported by the device's Device_annce + ( + t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_JOIN, + t.EmberJoinDecision.USE_PRECONFIGURED_KEY, + False, + True, + ), + ( + t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_REJOIN, + t.EmberJoinDecision.USE_PRECONFIGURED_KEY, + False, + False, + ), + ( + t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_JOIN, + t.EmberJoinDecision.DENY_JOIN, + False, + True, + ), + ( + t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_REJOIN, + t.EmberJoinDecision.DENY_JOIN, + False, + False, + ), + ], +) +async def test_join_handler(app, ieee, status, decision, joins, cleans_up_tclk): # Calls device.initialize, leaks a task app.handle_join = MagicMock() app.cleanup_tc_link_key = AsyncMock() app.ezsp_callback_handler( - "trustCenterJoinHandler", - [ - 1, - ieee, - t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_JOIN, - t.EmberJoinDecision.NO_ACTION, - sentinel.parent, - ], + "trustCenterJoinHandler", [1, ieee, status, decision, sentinel.parent] ) await asyncio.sleep(0) + assert ieee not in app.devices - assert app.handle_join.call_count == 1 - assert app.handle_join.call_args[0][0] == 1 - assert app.handle_join.call_args[0][1] == ieee - assert app.handle_join.call_args[0][2] is sentinel.parent - assert app.cleanup_tc_link_key.await_count == 1 - assert app.cleanup_tc_link_key.call_args[0][0] is ieee - - # cleanup TCLK, but no join handling - app.handle_join.reset_mock() - app.cleanup_tc_link_key.reset_mock() - app.ezsp_callback_handler( - "trustCenterJoinHandler", - [ - 1, - ieee, - t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_JOIN, - t.EmberJoinDecision.DENY_JOIN, - sentinel.parent, - ], - ) - await asyncio.sleep(0) - assert app.cleanup_tc_link_key.await_count == 1 - assert app.cleanup_tc_link_key.call_args[0][0] is ieee - assert app.handle_join.call_count == 0 + assert app.handle_join.mock_calls == [call(1, ieee, sentinel.parent)] * joins + assert app.cleanup_tc_link_key.mock_calls == [call(ieee)] * cleans_up_tclk def test_leave_handler(app, ieee): @@ -732,25 +741,162 @@ async def test_permit_ncp(app): assert app._ezsp._protocol.permitJoining.mock_calls == [call(60)] -async def test_permit_with_link_key(app, ieee): +POLICY_OPEN = call( + policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, + decisionId=( + t.EzspDecisionBitmask.ALLOW_JOINS + | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS + ), +) +POLICY_INSTALL_CODES_ONLY = call( + policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, + decisionId=( + t.EzspDecisionBitmask.ALLOW_JOINS + | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS + | t.EzspDecisionBitmask.JOINS_USE_INSTALL_CODE_KEY + ), +) +POLICY_CLOSED = call(policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, decisionId=0) + + +async def test_permit(app): + proto = app._ezsp._protocol + + def check_trust_center_ready(*args, **kwargs): + # The Trust Center must be ready before the network opens + assert proto.add_transient_link_key.mock_calls == [ + call(ezsp.WILDCARD_EUI64, ezsp.WELL_KNOWN_LINK_KEY) + ] + assert proto.setPolicy.mock_calls == [POLICY_OPEN] + + with patch( + "zigpy.application.ControllerApplication.permit", + side_effect=check_trust_center_ready, + ) as permit_mock: + await app.permit(10) + + assert permit_mock.mock_calls == [call(10, None)] + assert proto.clearTransientLinkKeys.mock_calls == [call()] + with patch("zigpy.application.ControllerApplication.permit") as permit_mock: - await app.permit_with_link_key( - ieee, - zigpy_t.KeyData.convert("11:22:33:44:55:66:77:88:11:22:33:44:55:66:77:88"), - 60, - ) + await app.permit(0) - assert permit_mock.await_count == 1 - assert app._ezsp._protocol.add_transient_link_key.mock_calls == [ - call( - ieee, - zigpy_t.KeyData.convert("11:22:33:44:55:66:77:88:11:22:33:44:55:66:77:88"), - ) + assert permit_mock.mock_calls == [call(0, None)] + assert proto.clearTransientLinkKeys.mock_calls == [call(), call()] + assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED] + + +async def test_permit_window_expires(app): + proto = app._ezsp._protocol + + with patch("zigpy.application.ControllerApplication.permit"): + await app.permit(0.01) + await asyncio.wait([app._join_window.task]) + + assert proto.clearTransientLinkKeys.mock_calls == [call(), call()] + assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED] + + # An expired window is reopened, not extended + with patch("zigpy.application.ControllerApplication.permit"): + await app.permit(10) + + assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED, POLICY_OPEN] + + +async def test_permit_extend(app): + proto = app._ezsp._protocol + + with patch("zigpy.application.ControllerApplication.permit") as permit_mock: + await app.permit(60) + window = app._join_window + await app.permit(0.01) + + # The open window is extended, joins are never rejected in between + assert app._join_window is window + assert permit_mock.mock_calls == [call(60, None), call(0.01, None)] + assert proto.clearTransientLinkKeys.mock_calls == [call()] + assert proto.setPolicy.mock_calls == [POLICY_OPEN] + + # The window now closes on the timer of the last call + await asyncio.wait([window.task]) + + assert proto.clearTransientLinkKeys.mock_calls == [call(), call()] + assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED] + + +async def test_permit_switch_link_keys(app, ieee): + proto = app._ezsp._protocol + key = zigpy_t.KeyData.convert("11:22:33:44:55:66:77:88:11:22:33:44:55:66:77:88") + + with patch("zigpy.application.ControllerApplication.permit"): + await app.permit(60) + await app.permit_with_link_key(ieee, key, 60) + await app.permit_with_link_key(ieee, key, 60) + + # A window with other keys is replaced, one with the same keys is extended + assert proto.setPolicy.mock_calls == [ + POLICY_OPEN, + POLICY_CLOSED, + POLICY_INSTALL_CODES_ONLY, ] + with patch("zigpy.application.ControllerApplication.permit"): + await app.permit(60) + + assert proto.setPolicy.mock_calls == [ + POLICY_OPEN, + POLICY_CLOSED, + POLICY_INSTALL_CODES_ONLY, + POLICY_CLOSED, + POLICY_OPEN, + ] + + +async def test_permit_concurrent(app): + proto = app._ezsp._protocol + + async def add_transient_link_key(ieee, key): + # Let the other permit call run while this window is opening + await asyncio.sleep(0) + return t.sl_Status.OK + + proto.add_transient_link_key.side_effect = add_transient_link_key + + with patch("zigpy.application.ControllerApplication.permit") as permit_mock: + await asyncio.gather(app.permit(60), app.permit(10)) + + # The second call waits for the first window to open and then extends it + assert permit_mock.mock_calls == [call(60, None), call(10, None)] + assert proto.setPolicy.mock_calls == [POLICY_OPEN] + + with patch("zigpy.application.ControllerApplication.permit"): + await app.permit(0) + + assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED] + + +async def test_permit_with_link_key(app, ieee): + proto = app._ezsp._protocol + key = zigpy_t.KeyData.convert("11:22:33:44:55:66:77:88:11:22:33:44:55:66:77:88") + + with patch("zigpy.application.ControllerApplication.permit") as permit_mock: + await app.permit_with_link_key(ieee, key, 60) + + assert permit_mock.mock_calls == [call(60)] + assert proto.add_transient_link_key.mock_calls == [call(ieee, key)] + assert proto.setPolicy.mock_calls == [POLICY_INSTALL_CODES_ONLY] + + # The policy is restored once the window closes + with patch("zigpy.application.ControllerApplication.permit"): + await app.permit(0) + + assert proto.clearTransientLinkKeys.mock_calls == [call(), call()] + assert proto.setPolicy.mock_calls == [POLICY_INSTALL_CODES_ONLY, POLICY_CLOSED] + async def test_permit_with_link_key_failure(app, ieee): - app._ezsp._protocol.add_transient_link_key.return_value = t.EmberStatus.ERR_FATAL + proto = app._ezsp._protocol + proto.add_transient_link_key.return_value = t.sl_Status.FAIL with patch("zigpy.application.ControllerApplication.permit") as permit_mock: with pytest.raises(ControllerError): @@ -762,8 +908,12 @@ async def test_permit_with_link_key_failure(app, ieee): 60, ) - assert permit_mock.await_count == 0 - assert app._ezsp._protocol.add_transient_link_key.await_count == 1 + assert len(permit_mock.mock_calls) == 0 + assert proto.add_transient_link_key.await_count == 1 + + # Joins were never allowed + assert proto.clearTransientLinkKeys.mock_calls == [call(), call()] + assert proto.setPolicy.mock_calls == [POLICY_CLOSED] @pytest.fixture @@ -1963,14 +2113,6 @@ async def test_cleanup_tc_link_key(app): assert ezsp.eraseKeyTableEntry.mock_calls == [call(index=sentinel.index)] -# @patch("zigpy.application.ControllerApplication.permit", new=AsyncMock()) -async def test_permit(app): - """Test permit method.""" - await app.permit(10) - await asyncio.sleep(0.1) - assert app._ezsp._protocol.add_transient_link_key.await_count == 1 - - @patch("bellows.zigbee.application.MFG_ID_RESET_DELAY", new=0.01) @pytest.mark.parametrize( "ieee, expected_mfg_id", diff --git a/tests/test_ezsp.py b/tests/test_ezsp.py index bccf17b39..77a017af2 100644 --- a/tests/test_ezsp.py +++ b/tests/test_ezsp.py @@ -12,12 +12,13 @@ from bellows import config, uart from bellows.ash import NcpFailure from bellows.exception import ( + ControllerError, EzspError, InvalidCommandError, InvalidCommandPayload, InvalidTxPower, ) -from bellows.ezsp import EZSP, EZSP_LATEST, xncp +from bellows.ezsp import EZSP, EZSP_LATEST, WELL_KNOWN_LINK_KEY, WILDCARD_EUI64, xncp from bellows.ezsp.v9.commands import GetTokenDataRsp import bellows.types as t @@ -463,6 +464,109 @@ async def test_set_disable_source_routing(ezsp_f): ) +@pytest.mark.parametrize( + "link_keys, expected_keys, install_codes_only", + [ + (None, {WILDCARD_EUI64: WELL_KNOWN_LINK_KEY}, False), + ( + { + t.EUI64.convert("00:11:22:33:44:55:66:77"): t.KeyData( + b"0123456789abcdef" + ) + }, + { + t.EUI64.convert("00:11:22:33:44:55:66:77"): t.KeyData( + b"0123456789abcdef" + ) + }, + True, + ), + ], +) +async def test_allow_joins(ezsp_f, link_keys, expected_keys, install_codes_only): + proto = ezsp_f._protocol + proto.add_transient_link_key = AsyncMock(return_value=t.sl_Status.OK) + proto.set_join_policy = AsyncMock() + proto.clear_transient_link_keys = AsyncMock() + + async with ezsp_f.allow_joins(link_keys=link_keys): + assert proto.add_transient_link_key.mock_calls == [ + call(ieee, key) for ieee, key in expected_keys.items() + ] + assert proto.set_join_policy.mock_calls == [ + call(allow_joins=True, install_codes_only=install_codes_only) + ] + + # Leftover keys from a previous window are cleared before the new ones + assert proto.clear_transient_link_keys.mock_calls == [call()] + + assert proto.clear_transient_link_keys.mock_calls == [call(), call()] + assert proto.set_join_policy.mock_calls == [ + call(allow_joins=True, install_codes_only=install_codes_only), + call(allow_joins=False), + ] + + +async def test_allow_joins_clear_keys_failure(ezsp_f): + proto = ezsp_f._protocol + proto.add_transient_link_key = AsyncMock(return_value=t.sl_Status.OK) + proto.set_join_policy = AsyncMock() + proto.clear_transient_link_keys = AsyncMock(side_effect=[None, TimeoutError()]) + + with pytest.raises(TimeoutError): + async with ezsp_f.allow_joins(): + pass + + # The policy is restored even though the keys could not be cleared + assert proto.set_join_policy.mock_calls == [ + call(allow_joins=True, install_codes_only=False), + call(allow_joins=False), + ] + + +async def test_allow_joins_link_key_failure(ezsp_f): + proto = ezsp_f._protocol + proto.add_transient_link_key = AsyncMock(return_value=t.sl_Status.FAIL) + proto.set_join_policy = AsyncMock() + proto.clear_transient_link_keys = AsyncMock() + + link_keys = { + t.EUI64.convert("00:11:22:33:44:55:66:77"): t.KeyData(b"0123456789abcdef") + } + + with pytest.raises(ControllerError): + async with ezsp_f.allow_joins(link_keys=link_keys): + pytest.fail("Joins must not be allowed") # pragma: no cover + + # Joins are never allowed but everything is still cleaned up + assert proto.clear_transient_link_keys.mock_calls == [call(), call()] + assert proto.set_join_policy.mock_calls == [call(allow_joins=False)] + + +@pytest.mark.parametrize( + "version, default", + [(v, t.EzspDecisionId.ALLOW_PRECONFIGURED_KEY_JOINS) for v in (4, 5, 6, 7)] + + [ + ( + v, + t.EzspDecisionBitmask.ALLOW_JOINS + | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS, + ) + for v in EZSP._BY_VERSION + if v >= 8 + ], +) +def test_trust_center_policy_default(version, default): + schema = EZSP._BY_VERSION[version].SCHEMAS[config.CONF_EZSP_POLICIES] + + policies = schema({}) + assert policies[t.EzspPolicyId.TRUST_CENTER_POLICY.name] == default + + # The user's configuration takes precedence + policies = schema({t.EzspPolicyId.TRUST_CENTER_POLICY.name: 0x02}) + assert policies[t.EzspPolicyId.TRUST_CENTER_POLICY.name] == 0x02 + + async def test_leave_network_error(ezsp_f): """Test EZSP leaveNetwork command failure.""" diff --git a/tests/test_ezsp_v10.py b/tests/test_ezsp_v10.py index 3ae6063d9..2b950b53a 100644 --- a/tests/test_ezsp_v10.py +++ b/tests/test_ezsp_v10.py @@ -1,4 +1,4 @@ -from unittest.mock import AsyncMock, MagicMock, call, patch +from unittest.mock import AsyncMock, MagicMock, call import pytest @@ -33,20 +33,6 @@ def test_ezsp_frame_rx(ezsp_f): assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234] -async def test_pre_permit(ezsp_f): - """Test pre permit.""" - p1 = patch.object(ezsp_f, "setPolicy", new=AsyncMock()) - p2 = patch.object( - ezsp_f, - "addTransientLinkKey", - new=AsyncMock(return_value=[t.EmberStatus.SUCCESS]), - ) - with p1 as pre_permit_mock, p2 as tclk_mock: - await ezsp_f.pre_permit(-1.9) - assert pre_permit_mock.await_count == 2 - assert tclk_mock.await_count == 1 - - async def test_write_child_data(ezsp_f) -> None: ezsp_f.setChildData.return_value = [t.EmberStatus.SUCCESS] diff --git a/tests/test_ezsp_v13.py b/tests/test_ezsp_v13.py index 3646e4d9a..a53dcbc3f 100644 --- a/tests/test_ezsp_v13.py +++ b/tests/test_ezsp_v13.py @@ -1,4 +1,4 @@ -from unittest.mock import AsyncMock, MagicMock, call, patch +from unittest.mock import MagicMock, call import pytest import zigpy.exceptions @@ -33,18 +33,22 @@ def test_ezsp_frame_rx(ezsp_f): assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234] -async def test_pre_permit(ezsp_f): - """Test pre permit.""" - p1 = patch.object(ezsp_f, "setPolicy", new=AsyncMock()) - p2 = patch.object( - ezsp_f, - "importTransientKey", - new=AsyncMock(return_value=[t.sl_Status.OK]), +async def test_add_transient_link_key(ezsp_f) -> None: + ezsp_f.importTransientKey.return_value = [t.sl_Status.OK] + + status = await ezsp_f.add_transient_link_key( + ieee=t.EUI64.convert("00:11:22:33:44:55:66:77"), + key=t.KeyData(b"ZigBeeAlliance09"), ) - with p1 as pre_permit_mock, p2 as tclk_mock: - await ezsp_f.pre_permit(-1.9) - assert pre_permit_mock.await_count == 2 - assert tclk_mock.await_count == 1 + + assert status == t.sl_Status.OK + assert ezsp_f.importTransientKey.mock_calls == [ + call( + eui64=t.EUI64.convert("00:11:22:33:44:55:66:77"), + plaintext_key=t.KeyData(b"ZigBeeAlliance09"), + flags=t.SecurityManagerContextFlags.NONE, + ) + ] async def test_read_link_keys(ezsp_f): diff --git a/tests/test_ezsp_v4.py b/tests/test_ezsp_v4.py index e8f688210..6b855b0b9 100644 --- a/tests/test_ezsp_v4.py +++ b/tests/test_ezsp_v4.py @@ -34,9 +34,24 @@ def test_ezsp_frame_rx(ezsp_f): assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234] -async def test_pre_permit(ezsp_f): - """Test pre permit.""" - await ezsp_f.pre_permit(1.9) +@pytest.mark.parametrize( + "kwargs", + [ + {"allow_joins": True}, + {"allow_joins": True, "install_codes_only": True}, + {"allow_joins": False}, + ], +) +async def test_set_join_policy(ezsp_f, kwargs) -> None: + # It's a no-op + await ezsp_f.set_join_policy(**kwargs) + assert len(ezsp_f.setPolicy.mock_calls) == 0 + + +async def test_clear_transient_link_keys(ezsp_f) -> None: + # It's a no-op + await ezsp_f.clear_transient_link_keys() + assert len(ezsp_f.clearTransientLinkKeys.mock_calls) == 0 async def test_read_child_data(ezsp_f): diff --git a/tests/test_ezsp_v5.py b/tests/test_ezsp_v5.py index 5f078365a..aad1922f1 100644 --- a/tests/test_ezsp_v5.py +++ b/tests/test_ezsp_v5.py @@ -32,19 +32,28 @@ def test_ezsp_frame_rx(ezsp_f): assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234] -async def test_pre_permit(ezsp_f): - """Test pre permit.""" +async def test_add_transient_link_key(ezsp_f) -> None: ezsp_f.addTransientLinkKey.return_value = [t.EmberStatus.SUCCESS] - await ezsp_f.pre_permit(1) + status = await ezsp_f.add_transient_link_key( + ieee=t.EUI64.convert("00:11:22:33:44:55:66:77"), + key=t.KeyData(b"ZigBeeAlliance09"), + ) + + assert status == t.sl_Status.OK assert ezsp_f.addTransientLinkKey.mock_calls == [ call( - partner=t.EUI64.convert("ff:ff:ff:ff:ff:ff:ff:ff"), + partner=t.EUI64.convert("00:11:22:33:44:55:66:77"), transientKey=t.KeyData(b"ZigBeeAlliance09"), ) ] +async def test_clear_transient_link_keys(ezsp_f) -> None: + await ezsp_f.clear_transient_link_keys() + assert ezsp_f.clearTransientLinkKeys.mock_calls == [call()] + + async def test_read_address_table(ezsp_f): def get_addr_table_node_id(addressTableIndex): return ( diff --git a/tests/test_ezsp_v8.py b/tests/test_ezsp_v8.py index e3d93de8c..d815ef83e 100644 --- a/tests/test_ezsp_v8.py +++ b/tests/test_ezsp_v8.py @@ -32,26 +32,39 @@ def test_ezsp_frame_rx(ezsp_f): assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234] -async def test_pre_permit(ezsp_f): - """Test pre permit.""" - ezsp_f.addTransientLinkKey.return_value = (t.EmberStatus.SUCCESS,) +@pytest.mark.parametrize( + "kwargs, decision", + [ + ( + {"allow_joins": True}, + t.EzspDecisionBitmask.DEFER_JOINS + | t.EzspDecisionBitmask.ALLOW_JOINS + | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS, + ), + ( + {"allow_joins": True, "install_codes_only": True}, + t.EzspDecisionBitmask.DEFER_JOINS + | t.EzspDecisionBitmask.ALLOW_JOINS + | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS + | t.EzspDecisionBitmask.JOINS_USE_INSTALL_CODE_KEY, + ), + # Closing restores the configured policy + ({"allow_joins": False}, t.EzspDecisionBitmask.DEFER_JOINS), + ], +) +async def test_set_join_policy(ezsp_f, kwargs, decision) -> None: + ezsp_f.setPolicy.return_value = [t.EmberStatus.SUCCESS] + + # Join windows build on top of the user's policy + await ezsp_f.update_policies( + {t.EzspPolicyId.TRUST_CENTER_POLICY.name: t.EzspDecisionBitmask.DEFER_JOINS} + ) + ezsp_f.setPolicy.reset_mock() + + await ezsp_f.set_join_policy(**kwargs) - await ezsp_f.pre_permit(-1.9) assert ezsp_f.setPolicy.mock_calls == [ - call( - policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, - decisionId=( - t.EzspDecisionBitmask.ALLOW_JOINS - | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS - ), - ), - call(policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, decisionId=0), - ] - assert ezsp_f.addTransientLinkKey.mock_calls == [ - call( - partner=t.EUI64.convert("ff:ff:ff:ff:ff:ff:ff:ff"), - transientKey=t.KeyData(b"ZigBeeAlliance09"), - ) + call(policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, decisionId=decision) ] diff --git a/tests/test_ezsp_v9.py b/tests/test_ezsp_v9.py index 9219282a1..a50895c7b 100644 --- a/tests/test_ezsp_v9.py +++ b/tests/test_ezsp_v9.py @@ -1,4 +1,4 @@ -from unittest.mock import AsyncMock, MagicMock, call, patch +from unittest.mock import MagicMock, call import pytest @@ -31,20 +31,6 @@ def test_ezsp_frame_rx(ezsp_f): assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234] -async def test_pre_permit(ezsp_f): - """Test pre permit.""" - p1 = patch.object(ezsp_f, "setPolicy", new=AsyncMock()) - p2 = patch.object( - ezsp_f, - "addTransientLinkKey", - new=AsyncMock(return_value=[t.EmberStatus.SUCCESS]), - ) - with p1 as pre_permit_mock, p2 as tclk_mock: - await ezsp_f.pre_permit(-1.9) - assert pre_permit_mock.await_count == 2 - assert tclk_mock.await_count == 1 - - async def test_write_child_data(ezsp_f) -> None: ezsp_f.setChildData.return_value = [t.EmberStatus.SUCCESS]