import asyncio import contextlib import importlib.metadata import logging from unittest.mock import AsyncMock, MagicMock, PropertyMock, call, patch, sentinel import pytest import zigpy.backups import zigpy.config import zigpy.device import zigpy.exceptions import zigpy.types as zigpy_t import zigpy.zdo.types as zdo_t import bellows.config as config from bellows.exception import ControllerError, EzspError import bellows.ezsp as ezsp from bellows.ezsp.v9.commands import GetTokenDataRsp import bellows.types import bellows.types as t import bellows.types.struct import bellows.uart as uart import bellows.zigbee.application from bellows.zigbee.application import ControllerApplication import bellows.zigbee.device from bellows.zigbee.util import map_rssi_to_energy from tests.common import mock_ezsp_commands APP_CONFIG = { config.CONF_DEVICE: { zigpy.config.CONF_DEVICE_PATH: "/dev/null", zigpy.config.CONF_DEVICE_BAUDRATE: 115200, }, zigpy.config.CONF_DATABASE: None, zigpy.config.CONF_STARTUP_ENERGY_SCAN: False, } @pytest.fixture def ieee(init=0): return t.EUI64(map(t.uint8_t, range(init, init + 8))) @pytest.fixture def make_app(monkeypatch, ieee): def inner(config, **kwargs): app_cfg = {**APP_CONFIG, **config} app = ControllerApplication(app_cfg) app._ezsp = _create_app_for_startup( app, nwk_type=t.EmberNodeType.COORDINATOR, ieee=ieee, **kwargs ) monkeypatch.setattr(bellows.zigbee.application, "APS_ACK_TIMEOUT", 0.05) app._ctrl_event.set() app._in_flight_msg = asyncio.Semaphore() app.handle_message = MagicMock() app.packet_received = MagicMock() return app return inner @pytest.fixture def app(make_app): return make_app({}) @pytest.fixture def aps(): f = t.EmberApsFrame() f.profileId = 99 f.sourceEndpoint = 8 f.clusterId = 6 f.destinationEndpoint = 8 f.sequence = 100 return f def _create_app_for_startup( app, nwk_type, ieee, auto_form=False, init=bellows.types.sl_Status.OK, ezsp_version=8, board_info=True, network_state=t.EmberNetworkStatus.JOINED_NETWORK, ): nwk_params = bellows.types.struct.EmberNetworkParameters( extendedPanId=t.ExtendedPanId.convert("aa:bb:cc:dd:ee:ff:aa:bb"), panId=t.EmberPanId(0x55AA), radioTxPower=0, radioChannel=25, joinMethod=t.EmberJoinMethod.USE_MAC_ASSOCIATION, nwkManagerId=t.EmberNodeId(0x0000), nwkUpdateId=1, channels=t.Channels.ALL_CHANNELS, ) app._in_flight_msg = None gateway = AsyncMock() ezsp_mock = ezsp.EZSP(device_config={}) ezsp_mock._gw = gateway ezsp_mock._ezsp_version = ezsp_version ezsp_mock._protocol = ezsp.EZSP._BY_VERSION[ezsp_version]( cb_handler=ezsp_mock.handle_callback, gateway=ezsp_mock._gw, ) ezsp_mock.start_ezsp() ezsp_mock.connect = AsyncMock() ezsp_mock.close = AsyncMock(wraps=ezsp_mock.close) ezsp_mock.startup_reset = AsyncMock() ezsp_mock.can_burn_userdata_custom_eui64 = AsyncMock(return_value=True) ezsp_mock.can_rewrite_custom_eui64 = AsyncMock(return_value=True) ezsp_mock.update_policies = AsyncMock() ezsp_mock.wait_for_stack_status = MagicMock() ezsp_mock.wait_for_stack_status.return_value.__enter__ = AsyncMock( return_value=t.EmberStatus.NETWORK_UP ) if board_info: ezsp_mock.get_board_info = AsyncMock( return_value=("Mock Manufacturer", "Mock board", "Mock version") ) else: ezsp_mock.get_board_info = AsyncMock(side_effect=EzspError("Not supported")) proto = ezsp_mock._protocol mock_ezsp_commands(proto) def form_network(): proto.getNetworkParameters.return_value = [ t.EmberStatus.SUCCESS, t.EmberNodeType.COORDINATOR, nwk_params, ] app.form_network = AsyncMock(side_effect=form_network) proto.initialize_network = AsyncMock( return_value=init, spec=proto.initialize_network ) proto.write_nwk_frame_counter = AsyncMock(spec=proto.write_nwk_frame_counter) proto.write_aps_frame_counter = AsyncMock(spec=proto.write_aps_frame_counter) proto.send_broadcast = AsyncMock( spec=proto.send_broadcast, return_value=[t.sl_Status.OK, 0x12] ) proto.write_link_keys = AsyncMock(spec=proto.write_link_keys) proto.write_child_data = AsyncMock(spec=proto.write_child_data) proto.add_transient_link_key = AsyncMock( return_value=t.sl_Status.OK, spec=proto.add_transient_link_key ) proto.get_network_key = AsyncMock( return_value=zigpy.state.Key( key=t.KeyData(b"ActualNetworkKey"), tx_counter=t.uint32_t(0x12345678), rx_counter=t.uint32_t(0x00000000), seq=t.uint8_t(1), partner_ieee=t.EUI64.convert("ff:ff:ff:ff:ff:ff:ff:ff"), ), proto=proto.get_network_key, ) proto.get_tc_link_key = AsyncMock( return_value=zigpy.state.Key( key=t.KeyData(b"thehashedlinkkey"), tx_counter=t.uint32_t(0x87654321), rx_counter=t.uint32_t(0x00000000), seq=t.uint8_t(0), partner_ieee=t.EUI64.convert("ff:ff:ff:ff:ff:ff:ff:ff"), ), proto=proto.get_tc_link_key, ) proto.factory_reset = AsyncMock(proto=proto.factory_reset) proto.set_extended_timeout = AsyncMock(proto=proto.set_extended_timeout) proto.read_link_keys = MagicMock() proto.read_link_keys.return_value.__aiter__.return_value = [ zigpy.state.Key( key=t.KeyData(b"test_link_key_01"), tx_counter=12345, rx_counter=67890, seq=1, partner_ieee=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"), ), zigpy.state.Key( key=t.KeyData(b"test_link_key_02"), tx_counter=54321, rx_counter=98765, seq=2, partner_ieee=t.EUI64.convert("11:22:33:44:55:66:77:88"), ), ] proto.read_child_data = MagicMock() proto.read_child_data.return_value.__aiter__.return_value = [ ( 0x1234, t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"), t.EmberNodeType.END_DEVICE, ), ( 0x5678, t.EUI64.convert("11:22:33:44:55:66:77:88"), t.EmberNodeType.END_DEVICE, ), ] proto.read_address_table = MagicMock() proto.read_address_table.return_value.__aiter__.return_value = [ (0xABCD, t.EUI64.convert("ab:cd:00:11:22:33:44:55")), (0xDCBA, t.EUI64.convert("dc:ba:00:11:22:33:44:55")), ] if "getTokenData" in proto.COMMANDS: proto.getTokenData.return_value = GetTokenDataRsp( status=t.EmberStatus.ERR_FATAL ) proto.addEndpoint.return_value = [t.EmberStatus.SUCCESS] proto.setManufacturerCode.return_value = [t.EmberStatus.SUCCESS] proto.setConfigurationValue.return_value = [t.EmberStatus.SUCCESS] proto.getNetworkParameters.return_value = [ t.EmberStatus.SUCCESS, nwk_type, nwk_params, ] proto.startScan.return_value = [[c, 1] for c in range(11, 26 + 1)] proto.setPolicy.return_value = [t.EmberStatus.SUCCESS] proto.getMfgToken.return_value = (b"Some token\xff",) proto.getNodeId.return_value = [t.EmberNodeId(0x0000)] proto.getEui64.return_value = [ieee] proto.getValue.return_value = (0, b"\x01" * 6) proto.setValue.return_value = (t.EmberStatus.SUCCESS,) proto.readCounters.return_value = ([0x0000] * len(t.EmberCounterType),) async def mock_leave(*args, **kwargs): app._ezsp.handle_callback("stackStatusHandler", [t.EmberStatus.NETWORK_DOWN]) return [t.EmberStatus.SUCCESS] proto.leaveNetwork.side_effect = mock_leave proto.getConfigurationValue.return_value = [t.EmberStatus.SUCCESS, 1] proto.networkState.return_value = [network_state] proto.setInitialSecurityState.return_value = [t.EmberStatus.SUCCESS] proto.formNetwork.return_value = [t.EmberStatus.SUCCESS] proto.getCurrentSecurityState.return_value = ( t.EmberStatus.SUCCESS, t.EmberCurrentSecurityState( bitmask=( t.EmberCurrentSecurityBitmask.GLOBAL_LINK_KEY | t.EmberCurrentSecurityBitmask.HAVE_TRUST_CENTER_LINK_KEY | 224 ), trustCenterLongAddress=ieee, ), ) proto.getMulticastTableEntry.return_value = ( t.EmberStatus.SUCCESS, t.EmberMulticastTableEntry(multicastId=0x0000, endpoint=0, networkIndex=0), ) proto.setMulticastTableEntry.return_value = [t.EmberStatus.SUCCESS] return ezsp_mock @contextlib.contextmanager def mock_for_startup( app, ieee, nwk_type=t.EmberNodeType.COORDINATOR, auto_form=False, init=0, ezsp_version=4, board_info=True, network_state=t.EmberNetworkStatus.JOINED_NETWORK, ): ezsp_mock = _create_app_for_startup( app, nwk_type, ieee, auto_form, init, ezsp_version, board_info, network_state ) with patch("bellows.ezsp.EZSP", return_value=ezsp_mock), patch( "zigpy.device.Device._initialize", new=AsyncMock() ): yield ezsp_mock async def _test_startup( app, nwk_type, ieee, auto_form=False, init=0, ezsp_version=4, board_info=True, network_state=t.EmberNetworkStatus.JOINED_NETWORK, ): with mock_for_startup( app, ieee, nwk_type, auto_form, init, ezsp_version, board_info, network_state ) as ezsp_mock: await app.startup(auto_form=auto_form) assert ezsp_mock._protocol.addEndpoint.call_count >= 2 async def test_startup(app, ieee): await _test_startup(app, t.EmberNodeType.COORDINATOR, ieee) async def test_startup_ezsp_ver7(app, ieee): app.state.counters["ezsp_counters"] = MagicMock() await _test_startup(app, t.EmberNodeType.COORDINATOR, ieee, ezsp_version=7) assert app.state.counters["ezsp_counters"].reset.call_count == 1 async def test_startup_ezsp_ver8(app, ieee): app.state.counters["ezsp_counters"] = MagicMock() ieee_1 = t.EUI64.convert("11:22:33:44:55:66:77:88") dev_1 = app.add_device(ieee_1, 0x1234) dev_1.relays = [ t.EmberNodeId(0x2222), ] await _test_startup(app, t.EmberNodeType.COORDINATOR, ieee, ezsp_version=8) assert app.state.counters["ezsp_counters"].reset.call_count == 1 async def test_startup_no_status(app, ieee): """Test when NCP is not a coordinator and not auto forming.""" with pytest.raises(zigpy.exceptions.NetworkNotFormed): await _test_startup( app, t.EmberNodeType.UNKNOWN_DEVICE, ieee, auto_form=False, init=1 ) async def test_startup_status_not_joined(app, ieee): """Test when NCP is a coordinator but isn't a part of a network.""" with pytest.raises(zigpy.exceptions.NetworkNotFormed): await _test_startup( app, t.EmberNodeType.COORDINATOR, ieee, auto_form=False, init=bellows.types.sl_Status.NOT_JOINED, network_state=t.EmberNetworkStatus.NO_NETWORK, ) async def test_startup_status_unknown(app, ieee): """Test when NCP is a coordinator but stack init fails.""" with pytest.raises(zigpy.exceptions.ControllerException): await _test_startup( app, t.EmberNodeType.COORDINATOR, ieee, auto_form=False, init=bellows.types.sl_Status.FAIL, network_state=t.EmberNetworkStatus.NO_NETWORK, ) async def test_startup_no_status_form(app, ieee): """Test when NCP is not a coordinator but allow auto forming.""" await _test_startup( app, t.EmberNodeType.UNKNOWN_NODE_TYPE, ieee, auto_form=True, init=1 ) async def test_startup_end(app, ieee): """Test when NCP is a End Device and not auto forming.""" with pytest.raises(zigpy.exceptions.NetworkNotFormed): await _test_startup( app, t.EmberNodeType.SLEEPY_END_DEVICE, ieee, auto_form=False ) async def test_startup_end_form(app, ieee): """Test when NCP is a End Device but allow auto forming.""" await _test_startup(app, t.EmberNodeType.SLEEPY_END_DEVICE, ieee, auto_form=True) async def test_startup_no_board_info(app, ieee, caplog): """Test when NCP does not support `get_board_info`.""" with caplog.at_level(logging.INFO): await _test_startup(app, t.EmberNodeType.COORDINATOR, ieee, board_info=False) assert "EZSP Radio does not support getMfgToken command" in caplog.text @pytest.fixture def aps_frame(): return t.EmberApsFrame( profileId=0x1234, clusterId=0x5678, sourceEndpoint=0x9A, destinationEndpoint=0xBC, options=t.EmberApsOption.APS_OPTION_NONE, groupId=0x0000, sequence=0xDE, ) def _handle_incoming_aps_frame(app, aps_frame, type): app.ezsp_callback_handler( "incomingMessageHandler", list( dict( type=type, apsFrame=aps_frame, lastHopLqi=123, lastHopRssi=-45, sender=0xABCD, bindingIndex=56, addressIndex=78, message=b"test message", ).values() ), ) def test_frame_handler_unicast(app, aps_frame): _handle_incoming_aps_frame( app, aps_frame, type=t.EmberIncomingMessageType.INCOMING_UNICAST ) assert app.packet_received.call_count == 1 packet = app.packet_received.mock_calls[0].args[0] assert packet.profile_id == 0x1234 assert packet.cluster_id == 0x5678 assert packet.src_ep == 0x9A assert packet.dst_ep == 0xBC assert packet.tsn == 0xDE assert packet.src.addr_mode == zigpy_t.AddrMode.NWK assert packet.src.address == 0xABCD assert packet.dst.addr_mode == zigpy_t.AddrMode.NWK assert packet.dst.address == app.state.node_info.nwk assert packet.data.serialize() == b"test message" assert packet.lqi == 123 assert packet.rssi == -45 assert ( app.state.counters[bellows.zigbee.application.COUNTERS_CTRL][ bellows.zigbee.application.COUNTER_RX_UNICAST ] == 1 ) def test_frame_handler_broadcast(app, aps_frame): _handle_incoming_aps_frame( app, aps_frame, type=t.EmberIncomingMessageType.INCOMING_BROADCAST ) assert app.packet_received.call_count == 1 packet = app.packet_received.mock_calls[0].args[0] assert packet.profile_id == 0x1234 assert packet.cluster_id == 0x5678 assert packet.src_ep == 0x9A assert packet.dst_ep == 0xBC assert packet.tsn == 0xDE assert packet.src.addr_mode == zigpy_t.AddrMode.NWK assert packet.src.address == 0xABCD assert packet.dst.addr_mode == zigpy_t.AddrMode.Broadcast assert packet.dst.address == zigpy_t.BroadcastAddress.ALL_ROUTERS_AND_COORDINATOR assert packet.data.serialize() == b"test message" assert packet.lqi == 123 assert packet.rssi == -45 assert ( app.state.counters[bellows.zigbee.application.COUNTERS_CTRL][ bellows.zigbee.application.COUNTER_RX_BCAST ] == 1 ) def test_frame_handler_multicast(app, aps_frame): aps_frame.groupId = 0xEF12 _handle_incoming_aps_frame( app, aps_frame, type=t.EmberIncomingMessageType.INCOMING_MULTICAST ) assert app.packet_received.call_count == 1 packet = app.packet_received.mock_calls[0].args[0] assert packet.profile_id == 0x1234 assert packet.cluster_id == 0x5678 assert packet.src_ep == 0x9A assert packet.dst_ep == 0xBC assert packet.tsn == 0xDE assert packet.src.addr_mode == zigpy_t.AddrMode.NWK assert packet.src.address == 0xABCD assert packet.dst.addr_mode == zigpy_t.AddrMode.Group assert packet.dst.address == 0xEF12 assert packet.data.serialize() == b"test message" assert packet.lqi == 123 assert packet.rssi == -45 assert ( app.state.counters[bellows.zigbee.application.COUNTERS_CTRL][ bellows.zigbee.application.COUNTER_RX_MCAST ] == 1 ) def test_frame_handler_ignored(app, aps_frame): _handle_incoming_aps_frame( app, aps_frame, type=t.EmberIncomingMessageType.INCOMING_BROADCAST_LOOPBACK ) assert app.packet_received.call_count == 0 @pytest.mark.parametrize( "msg_type", ( t.EmberIncomingMessageType.INCOMING_BROADCAST, t.EmberIncomingMessageType.INCOMING_MULTICAST, t.EmberIncomingMessageType.INCOMING_UNICAST, 0xFF, ), ) def test_send_failure(app, aps, ieee, msg_type): req = app._pending[(0xBEED, 254)] = MagicMock() app.ezsp_callback_handler( "messageSentHandler", [msg_type, 0xBEED, aps, 254, t.EmberStatus.SUCCESS, b""] ) assert req.result.set_exception.call_count == 0 assert req.result.set_result.call_count == 1 assert req.result.set_result.call_args[0][0][0] is bellows.types.sl_Status.OK def test_dup_send_failure(app, aps, ieee): req = app._pending[(0xBEED, 254)] = MagicMock() req.result.set_result.side_effect = asyncio.InvalidStateError() app.ezsp_callback_handler( "messageSentHandler", [ t.EmberIncomingMessageType.INCOMING_UNICAST, 0xBEED, aps, 254, sentinel.status, b"", ], ) assert req.result.set_exception.call_count == 0 assert req.result.set_result.call_count == 1 def test_send_failure_unexpected(app, aps, ieee): app.ezsp_callback_handler( "messageSentHandler", [ t.EmberIncomingMessageType.INCOMING_BROADCAST_LOOPBACK, 0xBEED, aps, 257, 1, b"", ], ) def test_send_success(app, aps, ieee): req = app._pending[(0xBEED, 253)] = MagicMock() app.ezsp_callback_handler( "messageSentHandler", [ t.EmberIncomingMessageType.INCOMING_MULTICAST_LOOPBACK, 0xBEED, aps, 253, t.EmberStatus.SUCCESS, b"", ], ) assert req.result.set_exception.call_count == 0 assert req.result.set_result.call_count == 1 assert req.result.set_result.call_args[0][0][0] is bellows.types.sl_Status.OK def test_unexpected_send_success(app, aps, ieee): app.ezsp_callback_handler( "messageSentHandler", [t.EmberIncomingMessageType.INCOMING_MULTICAST, 0xBEED, aps, 253, 0, b""], ) def test_dup_send_success(app, aps, ieee): req = app._pending[(0xBEED, 253)] = MagicMock() req.result.set_result.side_effect = asyncio.InvalidStateError() app.ezsp_callback_handler( "messageSentHandler", [t.EmberIncomingMessageType.INCOMING_MULTICAST, 0xBEED, aps, 253, 0, b""], ) assert req.result.set_exception.call_count == 0 assert req.result.set_result.call_count == 1 async def test_join_handler(app, ieee): # 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, ], ) 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 def test_leave_handler(app, ieee): app.handle_join = MagicMock() app.devices[ieee] = MagicMock() app.ezsp_callback_handler( "trustCenterJoinHandler", [1, ieee, t.EmberDeviceUpdate.DEVICE_LEFT, None, None] ) assert ieee in app.devices assert app.handle_join.call_count == 0 async def test_force_remove(app, ieee): app._ezsp._protocol.removeDevice.return_value = [t.EmberStatus.SUCCESS] dev = MagicMock() await app.force_remove(dev) def test_sequence(app): for i in range(1000): seq = app.get_sequence() assert seq >= 0 assert seq < 256 async def test_permit_ncp(app): app._ezsp._protocol.permitJoining.return_value = [t.EmberStatus.SUCCESS] await app.permit_ncp(60) assert app._ezsp._protocol.permitJoining.mock_calls == [call(60)] async def test_permit_with_link_key(app, ieee): 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, ) 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"), ) ] async def test_permit_with_link_key_failure(app, ieee): app._ezsp._protocol.add_transient_link_key.return_value = t.EmberStatus.ERR_FATAL with patch("zigpy.application.ControllerApplication.permit") as permit_mock: with pytest.raises(ControllerError): 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, ) assert permit_mock.await_count == 0 assert app._ezsp._protocol.add_transient_link_key.await_count == 1 @pytest.fixture def packet(): return zigpy_t.ZigbeePacket( src=zigpy_t.AddrModeAddress(addr_mode=zigpy_t.AddrMode.NWK, address=0x0000), src_ep=0x12, dst=zigpy_t.AddrModeAddress(addr_mode=zigpy_t.AddrMode.NWK, address=0x1234), dst_ep=0x34, tsn=0x56, profile_id=0x7890, cluster_id=0xABCD, data=zigpy_t.SerializableBytes(b"some data"), tx_options=zigpy_t.TransmitOptions.ACK, radius=0, ) async def _test_send_packet_unicast( app, packet, *, statuses=(bellows.types.sl_Status.OK,), options=( t.EmberApsOption.APS_OPTION_RETRY | t.EmberApsOption.APS_OPTION_ENABLE_ROUTE_DISCOVERY ), ): def send_unicast(*args, **kwargs): nonlocal statuses status = statuses[0] statuses = statuses[1:] if not statuses: asyncio.get_running_loop().call_later( 0.01, app.ezsp_callback_handler, "messageSentHandler", list( dict( type=t.EmberOutgoingMessageType.OUTGOING_DIRECT, indexOrDestination=0x1234, apsFrame=sentinel.aps, messageTag=sentinel.msg_tag, status=status, message=b"", ).values() ), ) return [status, 0x12] app._ezsp.send_unicast = AsyncMock( side_effect=send_unicast, spec=app._ezsp.send_unicast ) app.get_sequence = MagicMock(return_value=sentinel.msg_tag) expected_unicast_calls = len(statuses) await app.send_packet(packet) assert app._ezsp.send_unicast.call_count == expected_unicast_calls assert app._ezsp.send_unicast.mock_calls[-1] == ( call( nwk=t.EmberNodeId(0x1234), aps_frame=t.EmberApsFrame( profileId=packet.profile_id, clusterId=packet.cluster_id, sourceEndpoint=packet.src_ep, destinationEndpoint=packet.dst_ep, options=options, groupId=0x0000, sequence=packet.tsn, ), message_tag=sentinel.msg_tag, data=b"some data", ) ) assert len(app._pending) == 0 async def test_send_packet_unicast(app, packet): await _test_send_packet_unicast(app, packet) async def test_send_packet_unicast_not_running(app, packet): app.controller_event.clear() with pytest.raises(ControllerError): await _test_send_packet_unicast(app, packet) async def test_send_packet_unicast_ieee_fallback(app, packet, caplog): ieee = zigpy_t.EUI64.convert("aa:bb:cc:dd:11:22:33:44") packet.dst = zigpy_t.AddrModeAddress(addr_mode=zigpy_t.AddrMode.IEEE, address=ieee) app.add_device(nwk=0x1234, ieee=ieee) with caplog.at_level(logging.WARNING): await _test_send_packet_unicast(app, packet) assert "IEEE addressing is not supported" in caplog.text async def test_send_packet_unicast_ieee_no_fallback(app, packet, caplog): ieee = zigpy_t.EUI64.convert("aa:bb:cc:dd:11:22:33:44") packet.dst = zigpy_t.AddrModeAddress(addr_mode=zigpy_t.AddrMode.IEEE, address=ieee) with pytest.raises(ValueError): await _test_send_packet_unicast(app, packet) assert app._ezsp.send_unicast.call_count == 0 async def test_send_packet_unicast_source_route(make_app, packet): app = make_app({zigpy.config.CONF_SOURCE_ROUTING: True}) app._ezsp._protocol.set_source_route = AsyncMock( return_value=t.sl_Status.OK, spec=app._ezsp._protocol.set_source_route ) packet.source_route = [0x0001, 0x0002] await _test_send_packet_unicast( app, packet, options=( t.EmberApsOption.APS_OPTION_RETRY | t.EmberApsOption.APS_OPTION_ENABLE_ADDRESS_DISCOVERY ), ) app._ezsp._protocol.set_source_route.assert_called_once_with( nwk=packet.dst.address, relays=[0x0001, 0x0002], ) async def test_send_packet_unicast_extended_timeout(app, ieee, packet): app.add_device(nwk=packet.dst.address, ieee=ieee) await _test_send_packet_unicast( app, packet.replace(extended_timeout=True), ) assert app._ezsp._protocol.set_extended_timeout.mock_calls == [ call(nwk=packet.dst.address, ieee=ieee, extended_timeout=True) ] @patch("bellows.zigbee.application.RETRY_DELAYS", [0.01, 0.01, 0.01]) async def test_send_packet_unicast_retries_success(app, packet): await _test_send_packet_unicast( app, packet, statuses=( bellows.types.sl_Status.ALLOCATION_FAILED, bellows.types.sl_Status.ALLOCATION_FAILED, bellows.types.sl_Status.OK, ), ) async def test_send_packet_unicast_unexpected_failure(app, packet): with pytest.raises(zigpy.exceptions.DeliveryError): await _test_send_packet_unicast( app, packet, statuses=(t.EmberStatus.ERR_FATAL,) ) @patch("bellows.zigbee.application.RETRY_DELAYS", [0.01, 0.01, 0.01]) async def test_send_packet_unicast_retries_failure(app, packet): with pytest.raises(zigpy.exceptions.DeliveryError): await _test_send_packet_unicast( app, packet, statuses=( bellows.types.sl_Status.ALLOCATION_FAILED, bellows.types.sl_Status.ALLOCATION_FAILED, bellows.types.sl_Status.ALLOCATION_FAILED, ), ) async def test_send_packet_unicast_concurrency(app, packet, monkeypatch): monkeypatch.setattr(bellows.zigbee.application, "APS_ACK_TIMEOUT", 0.5) app._concurrent_requests_semaphore.max_value = 10 max_concurrency = 0 in_flight_requests = 0 async def send_message_sent_reply( type, indexOrDestination, apsFrame, messageTag, message ): nonlocal max_concurrency, in_flight_requests max_concurrency = max(max_concurrency, in_flight_requests) in_flight_requests -= 1 await asyncio.sleep(0.01) app.ezsp_callback_handler( "messageSentHandler", list( dict( type=type, indexOrDestination=indexOrDestination, apsFrame=apsFrame, messageTag=messageTag, status=t.EmberStatus.SUCCESS, message=b"", ).values() ), ) async def send_unicast(nwk, aps_frame, message_tag, data): nonlocal max_concurrency, in_flight_requests in_flight_requests += 1 max_concurrency = max(max_concurrency, in_flight_requests) asyncio.create_task( send_message_sent_reply( t.EmberOutgoingMessageType.OUTGOING_DIRECT, nwk, aps_frame, message_tag, data, ) ) return [bellows.types.sl_Status.OK, 0x12] app._ezsp.send_unicast = AsyncMock(side_effect=send_unicast) responses = await asyncio.gather(*[app.send_packet(packet) for _ in range(100)]) assert len(responses) == 100 assert max_concurrency == 10 assert in_flight_requests == 0 async def test_send_packet_broadcast(app, packet): packet.dst = zigpy_t.AddrModeAddress( addr_mode=zigpy_t.AddrMode.Broadcast, address=0xFFFE ) packet.radius = 30 app._ezsp.send_broadcast = AsyncMock( return_value=(bellows.types.named.sl_Status.OK, 0x12) ) app.get_sequence = MagicMock(return_value=sentinel.msg_tag) asyncio.get_running_loop().call_soon( app.ezsp_callback_handler, "messageSentHandler", list( dict( type=t.EmberOutgoingMessageType.OUTGOING_BROADCAST, indexOrDestination=0xFFFE, apsFrame=sentinel.aps, messageTag=sentinel.msg_tag, status=t.EmberStatus.SUCCESS, message=b"", ).values() ), ) await app.send_packet(packet) assert app._ezsp.send_broadcast.mock_calls == [ call( address=bellows.types.named.BroadcastAddress(0xFFFE), 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_RETRY | t.EmberApsOption.APS_OPTION_ENABLE_ROUTE_DISCOVERY ), groupId=0x0000, sequence=packet.tsn, ), radius=packet.radius, message_tag=sentinel.msg_tag, aps_sequence=packet.tsn, data=b"some data", ) ] assert len(app._pending) == 0 async def test_send_packet_broadcast_ignored_delivery_failure(app, packet): packet.dst = zigpy_t.AddrModeAddress( addr_mode=zigpy_t.AddrMode.Broadcast, address=0xFFFE ) packet.radius = 30 app.get_sequence = MagicMock(return_value=sentinel.msg_tag) asyncio.get_running_loop().call_soon( app.ezsp_callback_handler, "messageSentHandler", list( dict( type=t.EmberOutgoingMessageType.OUTGOING_BROADCAST, indexOrDestination=0xFFFE, apsFrame=sentinel.aps, messageTag=sentinel.msg_tag, status=t.EmberStatus.DELIVERY_FAILED, message=b"", ).values() ), ) # Does not throw an error await app.send_packet(packet) assert app._ezsp.send_broadcast.mock_calls == [ call( address=bellows.types.named.BroadcastAddress(0xFFFE), 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_RETRY | t.EmberApsOption.APS_OPTION_ENABLE_ROUTE_DISCOVERY ), groupId=0x0000, sequence=packet.tsn, ), radius=packet.radius, message_tag=sentinel.msg_tag, aps_sequence=packet.tsn, data=b"some data", ) ] assert len(app._pending) == 0 async def test_send_packet_multicast(app, packet): packet.dst = zigpy_t.AddrModeAddress( addr_mode=zigpy_t.AddrMode.Group, address=0x1234 ) packet.radius = 5 packet.non_member_radius = 6 app._ezsp._protocol.send_multicast = AsyncMock( return_value=(bellows.types.sl_Status.OK, 0x12), spec=app._ezsp._protocol.send_multicast, ) app.get_sequence = MagicMock(return_value=sentinel.msg_tag) asyncio.get_running_loop().call_soon( app.ezsp_callback_handler, "messageSentHandler", list( dict( type=t.EmberOutgoingMessageType.OUTGOING_MULTICAST, indexOrDestination=0x1234, apsFrame=sentinel.aps, messageTag=sentinel.msg_tag, status=t.EmberStatus.SUCCESS, message=b"", ).values() ), ) await app.send_packet(packet) assert app._ezsp.send_multicast.mock_calls == [ call( 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_RETRY | t.EmberApsOption.APS_OPTION_ENABLE_ROUTE_DISCOVERY ), groupId=0x1234, sequence=packet.tsn, ), radius=packet.radius, non_member_radius=packet.non_member_radius, message_tag=sentinel.msg_tag, data=b"some data", ) ] assert len(app._pending) == 0 def test_is_controller_running(app): ezsp_running = PropertyMock(return_value=False) with patch.object(type(app._ezsp), "is_ezsp_running", ezsp_running): app._ctrl_event.clear() assert app.is_controller_running is False app._ctrl_event.set() assert app.is_controller_running is False assert ezsp_running.call_count == 1 ezsp_running = PropertyMock(return_value=True) with patch.object(type(app._ezsp), "is_ezsp_running", ezsp_running): app._ctrl_event.clear() assert app.is_controller_running is False app._ctrl_event.set() assert app.is_controller_running is True assert ezsp_running.call_count == 1 def test_reset_frame(app): app.connection_lost = MagicMock(spec_set=app.connection_lost) app.ezsp_callback_handler("_reset_controller_application", (sentinel.error,)) assert app.connection_lost.mock_calls == [call(sentinel.error)] @pytest.mark.parametrize("ezsp_version", (4, 7)) async def test_watchdog(make_app, monkeypatch, ezsp_version): from bellows.zigbee import application app = make_app({}, ezsp_version=ezsp_version) monkeypatch.setattr(application.ControllerApplication, "_watchdog_period", 0.01) monkeypatch.setattr(application, "EZSP_COUNTERS_CLEAR_IN_WATCHDOG_PERIODS", 2) nop_success = 7 async def nop_mock(): nonlocal nop_success if nop_success: nop_success -= 1 if nop_success % 3: raise EzspError else: return ([0] * 10,) raise asyncio.TimeoutError app._ezsp._protocol.getValue.return_value = [t.EmberStatus.SUCCESS, b"\xFE"] app._ezsp._protocol.nop.side_effect = nop_mock app._ezsp._protocol.readCounters.side_effect = nop_mock app._ezsp._protocol.readAndClearCounters.side_effect = nop_mock app._ctrl_event.set() app.connection_lost = MagicMock() for i in range(nop_success): await app._watchdog_feed() # Fail four times in a row to exhaust the watchdog buffer await app._watchdog_feed() await app._watchdog_feed() await app._watchdog_feed() await app._watchdog_feed() # The last time will throw a real error with pytest.raises(asyncio.TimeoutError): await app._watchdog_feed() if ezsp_version == 4: assert app._ezsp._protocol.nop.await_count > 4 else: assert app._ezsp._protocol.readCounters.await_count >= 4 async def test_watchdog_counters(app, monkeypatch, caplog): from bellows.zigbee import application monkeypatch.setattr(application.ControllerApplication, "_watchdog_period", 0.01) nop_success = 3 async def counters_mock(): nonlocal nop_success if nop_success: nop_success -= 1 if nop_success % 2: raise EzspError else: return ([0, 1, 2, 3],) raise asyncio.TimeoutError app._ezsp._protocol.getValue = AsyncMock( return_value=[t.EmberStatus.SUCCESS, b"\xFE"] ) app._ezsp._protocol.readCounters = AsyncMock(side_effect=counters_mock) app._ezsp._protocol.nop = AsyncMock(side_effect=EzspError) app._handle_reset_request = MagicMock() app._ctrl_event.set() caplog.set_level(logging.DEBUG, "bellows.zigbee.application") await app._watchdog_feed() assert app._ezsp._protocol.readCounters.await_count != 0 assert app._ezsp._protocol.nop.await_count == 0 # don't do counters on older firmwares app._ezsp._ezsp_version = 4 app._ezsp._protocol.readCounters.reset_mock() await app._watchdog_feed() assert app._ezsp._protocol.readCounters.await_count == 0 assert app._ezsp._protocol.nop.await_count != 0 async def test_ezsp_value_counter(app, monkeypatch): from bellows.zigbee import application monkeypatch.setattr(application.ControllerApplication, "_watchdog_period", 0.01) nop_success = 3 async def counters_mock(): nonlocal nop_success if nop_success: nop_success -= 1 if nop_success % 2: raise EzspError else: return {t.EmberCounterType(i): v for i, v in enumerate([0, 1, 2, 3])} raise asyncio.TimeoutError app._ezsp.read_counters = AsyncMock(side_effect=counters_mock) app._ezsp.nop = AsyncMock(side_effect=EzspError) app._ezsp.getValue = AsyncMock( return_value=(t.EzspStatus.ERROR_OUT_OF_MEMORY, b"\x20") ) app._handle_reset_request = MagicMock() app._ctrl_event.set() await app._watchdog_feed() assert app._ezsp.read_counters.await_count != 0 assert app._ezsp.nop.await_count == 0 cnt = t.EmberCounterType assert ( app.state.counters[application.COUNTERS_EZSP][ cnt.COUNTER_MAC_RX_BROADCAST.name[8:] ] == 0 ) assert ( app.state.counters[application.COUNTERS_EZSP][ cnt.COUNTER_MAC_TX_BROADCAST.name[8:] ] == 1 ) assert ( app.state.counters[application.COUNTERS_EZSP][ cnt.COUNTER_MAC_RX_UNICAST.name[8:] ] == 2 ) assert ( app.state.counters[application.COUNTERS_EZSP][ cnt.COUNTER_MAC_TX_UNICAST_SUCCESS.name[8:] ] == 3 ) assert ( app.state.counters[application.COUNTERS_EZSP].get( application.COUNTER_EZSP_BUFFERS ) is None ) assert ( app.state.counters[application.COUNTERS_CTRL][application.COUNTER_WATCHDOG] == 0 ) # Ezsp Value success app._ezsp.getValue = AsyncMock(return_value=(t.EzspStatus.SUCCESS, b"\x20")) nop_success = 3 await app._watchdog_feed() assert ( app.state.counters[application.COUNTERS_EZSP][application.COUNTER_EZSP_BUFFERS] == 0x20 ) async def test_shutdown(app): ezsp = app._ezsp await app.shutdown() assert app.controller_event.is_set() is False assert ezsp.close.call_count == 1 @pytest.fixture def coordinator(app, ieee): dev = zigpy.device.Device(app, ieee, 0x0000) dev.endpoints[1] = bellows.zigbee.device.EZSPEndpoint(dev, 1, MagicMock()) dev.model = dev.endpoints[1].model dev.manufacturer = dev.endpoints[1].manufacturer return dev async def test_ezsp_add_to_group(coordinator): coordinator.application._multicast = MagicMock() mc = coordinator.application._multicast mc.subscribe = AsyncMock(return_value=t.EmberStatus.SUCCESS) grp_id = 0x2345 assert grp_id not in coordinator.endpoints[1].member_of ret = await coordinator.add_to_group(grp_id) assert ret is None assert mc.subscribe.call_count == 1 assert mc.subscribe.call_args[0][0] == grp_id assert grp_id in coordinator.endpoints[1].member_of async def test_ezsp_add_to_group_ep(coordinator): coordinator.application._multicast = MagicMock() mc = coordinator.application._multicast mc.subscribe = AsyncMock(return_value=t.EmberStatus.SUCCESS) grp_id = 0x2345 assert grp_id not in coordinator.endpoints[1].member_of await coordinator.endpoints[1].add_to_group(grp_id) assert mc.subscribe.call_count == 1 assert mc.subscribe.call_args[0][0] == grp_id assert grp_id in coordinator.endpoints[1].member_of mc.reset_mock() await coordinator.endpoints[1].add_to_group(grp_id) assert mc.subscribe.call_count == 0 async def test_ezsp_add_to_group_fail(coordinator): coordinator.application._multicast = MagicMock() mc = coordinator.application._multicast mc.subscribe = AsyncMock(return_value=t.EmberStatus.ERR_FATAL) grp_id = 0x2345 assert grp_id not in coordinator.endpoints[1].member_of with pytest.raises(ValueError): await coordinator.add_to_group(grp_id) assert mc.subscribe.call_count == 1 assert mc.subscribe.call_args[0][0] == grp_id assert grp_id not in coordinator.endpoints[1].member_of async def test_ezsp_add_to_group_ep_fail(coordinator): coordinator.application._multicast = MagicMock() mc = coordinator.application._multicast mc.subscribe = AsyncMock(return_value=t.EmberStatus.ERR_FATAL) grp_id = 0x2345 assert grp_id not in coordinator.endpoints[1].member_of with pytest.raises(ValueError): await coordinator.endpoints[1].add_to_group(grp_id) assert mc.subscribe.call_count == 1 assert mc.subscribe.call_args[0][0] == grp_id assert grp_id not in coordinator.endpoints[1].member_of async def test_ezsp_remove_from_group(coordinator): coordinator.application._multicast = MagicMock() mc = coordinator.application._multicast mc.unsubscribe = AsyncMock(return_value=t.EmberStatus.SUCCESS) grp_id = 0x2345 grp = coordinator.application.groups.add_group(grp_id) grp.add_member(coordinator.endpoints[1]) assert grp_id in coordinator.endpoints[1].member_of await coordinator.remove_from_group(grp_id) assert mc.unsubscribe.call_count == 1 assert mc.unsubscribe.call_args[0][0] == grp_id assert grp_id not in coordinator.endpoints[1].member_of async def test_ezsp_remove_from_group_ep(coordinator): coordinator.application._multicast = MagicMock() mc = coordinator.application._multicast mc.unsubscribe = AsyncMock(return_value=t.EmberStatus.SUCCESS) grp_id = 0x2345 grp = coordinator.application.groups.add_group(grp_id) grp.add_member(coordinator.endpoints[1]) assert grp_id in coordinator.endpoints[1].member_of await coordinator.endpoints[1].remove_from_group(grp_id) assert mc.unsubscribe.call_count == 1 assert mc.unsubscribe.call_args[0][0] == grp_id assert grp_id not in coordinator.endpoints[1].member_of mc.reset_mock() await coordinator.endpoints[1].remove_from_group(grp_id) assert mc.subscribe.call_count == 0 async def test_ezsp_remove_from_group_fail(coordinator): coordinator.application._multicast = MagicMock() mc = coordinator.application._multicast mc.unsubscribe = AsyncMock(return_value=t.EmberStatus.ERR_FATAL) grp_id = 0x2345 grp = coordinator.application.groups.add_group(grp_id) grp.add_member(coordinator.endpoints[1]) assert grp_id in coordinator.endpoints[1].member_of with pytest.raises(ValueError): await coordinator.remove_from_group(grp_id) assert mc.unsubscribe.call_count == 1 assert mc.unsubscribe.call_args[0][0] == grp_id async def test_ezsp_remove_from_group_fail_ep(coordinator): coordinator.application._multicast = MagicMock() mc = coordinator.application._multicast mc.unsubscribe = AsyncMock(return_value=t.EmberStatus.ERR_FATAL) grp_id = 0x2345 grp = coordinator.application.groups.add_group(grp_id) grp.add_member(coordinator.endpoints[1]) assert grp_id in coordinator.endpoints[1].member_of with pytest.raises(ValueError): await coordinator.endpoints[1].remove_from_group(grp_id) assert mc.unsubscribe.call_count == 1 assert mc.unsubscribe.call_args[0][0] == grp_id def test_coordinator_model_manuf(coordinator): """Test we get coordinator manufacturer and model.""" assert coordinator.endpoints[1].manufacturer assert coordinator.endpoints[1].model def test_handle_route_record(app): """Test route record handling for an existing device.""" app.handle_relays = MagicMock(spec_set=app.handle_relays) app.ezsp_callback_handler( "incomingRouteRecordHandler", [sentinel.nwk, sentinel.ieee, sentinel.lqi, sentinel.rssi, sentinel.relays], ) app.handle_relays.assert_called_once_with(nwk=sentinel.nwk, relays=sentinel.relays) def test_handle_route_error(app): """Test route error handler.""" app.handle_relays = MagicMock(spec_set=app.handle_relays) app.ezsp_callback_handler( "incomingRouteErrorHandler", [sentinel.status, sentinel.nwk] ) app.handle_relays.assert_not_called() def test_handle_id_conflict(app, ieee): """Test handling of an ID conflict report.""" nwk = t.EmberNodeId(0x1234) app.add_device(ieee, nwk) app.handle_leave = MagicMock() app.ezsp_callback_handler("idConflictHandler", [nwk + 1]) assert app.handle_leave.call_count == 0 app.ezsp_callback_handler("idConflictHandler", [nwk]) assert app.handle_leave.call_count == 1 assert app.handle_leave.call_args[0][0] == nwk async def test_handle_no_such_device(app, ieee): """Test handling of an unknown device IEEE lookup.""" app._ezsp.lookupEui64ByNodeId = AsyncMock() p1 = patch.object( app._ezsp, "lookupEui64ByNodeId", AsyncMock(return_value=(t.EmberStatus.ERR_FATAL, ieee)), ) p2 = patch.object(app, "handle_join") with p1 as lookup_mock, p2 as handle_join_mock: await app._handle_no_such_device(sentinel.nwk) assert lookup_mock.mock_calls == [call(nodeId=sentinel.nwk)] assert handle_join_mock.call_count == 0 p1 = patch.object( app._ezsp, "lookupEui64ByNodeId", AsyncMock(return_value=(t.EmberStatus.SUCCESS, sentinel.ieee)), ) with p1 as lookup_mock, p2 as handle_join_mock: await app._handle_no_such_device(sentinel.nwk) assert lookup_mock.mock_calls == [call(nodeId=sentinel.nwk)] assert handle_join_mock.call_count == 1 assert handle_join_mock.call_args[0][0] == sentinel.nwk assert handle_join_mock.call_args[0][1] == sentinel.ieee async def test_cleanup_tc_link_key(app): """Test cleaning up tc link key.""" ezsp = app._ezsp ezsp.findKeyTableEntry = AsyncMock(side_effect=((0xFF,), (sentinel.index,))) ezsp.eraseKeyTableEntry = AsyncMock(return_value=(0x00,)) await app.cleanup_tc_link_key(sentinel.ieee) assert ezsp.findKeyTableEntry.mock_calls == [ call(address=sentinel.ieee, linkKey=True) ] assert ezsp.eraseKeyTableEntry.await_count == 0 assert ezsp.eraseKeyTableEntry.call_count == 0 ezsp.findKeyTableEntry.reset_mock() await app.cleanup_tc_link_key(sentinel.ieee2) assert ezsp.findKeyTableEntry.mock_calls == [ call(address=sentinel.ieee2, linkKey=True) ] 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", ( ("54:ef:44:00:00:00:00:11", 0x115F), ("04:cf:8c:00:00:00:00:11", 0x115F), ("01:22:33:00:00:00:00:11", None), ), ) async def test_set_mfg_id(ieee, expected_mfg_id, app): """Test setting manufacturer id based on IEEE address.""" app.handle_join = MagicMock() app.cleanup_tc_link_key = AsyncMock() app.ezsp_callback_handler( "trustCenterJoinHandler", [ 1, t.EUI64.convert(ieee), t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_JOIN, t.EmberJoinDecision.NO_ACTION, sentinel.parent, ], ) # preempt app.ezsp_callback_handler( "trustCenterJoinHandler", [ 1, t.EUI64.convert(ieee), t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_JOIN, t.EmberJoinDecision.NO_ACTION, sentinel.parent, ], ) await asyncio.sleep(0.20) if expected_mfg_id is not None: assert app._ezsp._protocol.setManufacturerCode.mock_calls == [ call(code=expected_mfg_id), call(code=bellows.zigbee.application.DEFAULT_MFG_ID), ] else: assert app._ezsp._protocol.setManufacturerCode.mock_calls == [] async def test_ensure_network_running_joined(app): ezsp = app._ezsp # Make initialization take two attempts ezsp.initialize_network = AsyncMock( side_effect=[ bellows.types.sl_Status.INVALID_PARAMETER, bellows.types.sl_Status.OK, ] ) ezsp.networkState = AsyncMock(return_value=[t.EmberNetworkStatus.JOINED_NETWORK]) rsp = await app._ensure_network_running() assert not rsp ezsp.initialize_network.assert_not_called() async def test_ensure_network_running_not_joined_failure(app): ezsp = app._ezsp ezsp.networkState = AsyncMock(return_value=[t.EmberNetworkStatus.NO_NETWORK]) ezsp.initialize_network = AsyncMock( return_value=bellows.types.sl_Status.INVALID_PARAMETER ) with pytest.raises(zigpy.exceptions.ControllerException): await app._ensure_network_running() ezsp.networkState.assert_called_once() ezsp.initialize_network.assert_called_once() async def test_ensure_network_running_not_joined_success(app): ezsp = app._ezsp ezsp.networkState = AsyncMock(return_value=[t.EmberNetworkStatus.NO_NETWORK]) ezsp.initialize_network = AsyncMock(return_value=bellows.types.sl_Status.OK) rsp = await app._ensure_network_running() assert rsp ezsp.networkState.assert_called_once() ezsp.initialize_network.assert_called_once() async def test_startup_coordinator_existing_groups_joined(app, ieee): """Coordinator joins groups loaded from the database.""" with mock_for_startup(app, ieee): await app.connect() db_device = app.add_device(ieee, 0x0000) db_ep = db_device.add_endpoint(1) app.groups.add_group(0x1234, "Group Name", suppress_event=True) app.groups[0x1234].add_member(db_ep, suppress_event=True) await app.start_network() assert app._ezsp._protocol.setMulticastTableEntry.mock_calls == [ call( 0, t.EmberMulticastTableEntry(multicastId=0x1234, endpoint=1, networkIndex=0), ) ] async def test_startup_new_coordinator_no_groups_joined(app, ieee): """Coordinator freshy added to the database has no groups to join.""" with mock_for_startup(app, ieee): await app.connect() await app.start_network() assert app._ezsp._protocol.setMulticastTableEntry.mock_calls == [] @pytest.mark.parametrize( "scan_results", [ # Normal [ -39, -30, -26, -33, -23, -53, -42, -46, -69, -75, -49, -67, -57, -81, -29, -40, ], # Maximum [10] * (26 - 11 + 1), # Minimum [-200] * (26 - 11 + 1), ], ) async def test_energy_scanning(app, scan_results): app._ezsp.startScan = AsyncMock( return_value=list(zip(range(11, 26 + 1), scan_results)) ) results = await app.energy_scan( channels=t.Channels.ALL_CHANNELS, duration_exp=2, count=1, ) assert len(app._ezsp.startScan.mock_calls) == 1 assert set(results.keys()) == set(t.Channels.ALL_CHANNELS) assert all(0 <= v <= 255 for v in results.values()) async def test_energy_scanning_partial(app): app._ezsp.startScan = AsyncMock( side_effect=[ [(11, 11), (12, 12), (13, 13), (14, 14), (15, 15), (16, 16)], [(17, 17)], # Channel that doesn't exist [], [(18, 18), (19, 19), (20, 20)], [(18, 18), (19, 19), (20, 20)], # Duplicate results [(21, 21), (22, 22), (23, 23), (24, 24), (25, 25), (26, 26)], ] ) results = await app.energy_scan( channels=t.Channels.from_channel_list([11, 13, 14, 15, 20, 25, 26]), duration_exp=2, count=1, ) assert len(app._ezsp.startScan.mock_calls) == 6 assert set(results.keys()) == {11, 13, 14, 15, 20, 25, 26} assert results == {c: map_rssi_to_energy(c) for c in [11, 13, 14, 15, 20, 25, 26]} async def test_connect_failure(app: ControllerApplication) -> None: """Test that a failure to connect propagates.""" ezsp = app._ezsp app._ezsp.write_config = AsyncMock(side_effect=OSError()) app._ezsp.connect = AsyncMock() app._ezsp = None with patch("bellows.ezsp.EZSP", return_value=ezsp): with pytest.raises(OSError): await app.connect() assert app._ezsp is None assert len(ezsp.close.mock_calls) == 1 async def test_repair_tclk_partner_ieee( app: ControllerApplication, ieee: zigpy_t.EUI64 ) -> None: """Test that EZSP is reset after repairing TCLK.""" app._reset = AsyncMock() with mock_for_startup(app, ieee), patch( "bellows.zigbee.repairs.fix_invalid_tclk_partner_ieee", AsyncMock(return_value=False), ): await app.connect() await app.start_network() assert len(app._reset.mock_calls) == 0 app._reset.reset_mock() with mock_for_startup(app, ieee), patch( "bellows.zigbee.repairs.fix_invalid_tclk_partner_ieee", AsyncMock(return_value=True), ): await app.start_network() assert len(app._reset.mock_calls) == 1 @pytest.fixture def zigpy_backup() -> zigpy.backups.NetworkBackup: return zigpy.backups.NetworkBackup( node_info=zigpy.state.NodeInfo( nwk=zigpy_t.NWK(0x0000), ieee=t.EUI64.convert("07:06:05:04:03:02:01:00"), logical_type=zdo_t.LogicalType.Coordinator, model="Mock board", manufacturer="Mock Manufacturer", version="Mock version", ), network_info=zigpy.state.NetworkInfo( extended_pan_id=zigpy_t.ExtendedPanId.convert("aa:bb:cc:dd:ee:ff:aa:bb"), pan_id=zigpy_t.PanId(0x55AA), nwk_update_id=1, nwk_manager_id=zigpy_t.NWK(0x0000), channel=t.uint8_t(25), channel_mask=t.Channels.ALL_CHANNELS, security_level=t.uint8_t(1), network_key=zigpy.state.Key( key=t.KeyData.convert( "41:63:74:75:61:6c:4e:65:74:77:6f:72:6b:4b:65:79" ), seq=1, tx_counter=305419896, ), tc_link_key=zigpy.state.Key( key=t.KeyData(b"ZigBeeAlliance09"), partner_ieee=t.EUI64.convert("07:06:05:04:03:02:01:00"), tx_counter=2271560481, ), key_table=[ zigpy.state.Key( key=t.KeyData.convert( "74:65:73:74:5f:6c:69:6e:6b:5f:6b:65:79:5f:30:31" ), tx_counter=12345, rx_counter=67890, seq=1, partner_ieee=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"), ), zigpy.state.Key( key=t.KeyData.convert( "74:65:73:74:5f:6c:69:6e:6b:5f:6b:65:79:5f:30:32" ), tx_counter=54321, rx_counter=98765, seq=2, partner_ieee=t.EUI64.convert("11:22:33:44:55:66:77:88"), ), ], children=[ t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"), t.EUI64.convert("11:22:33:44:55:66:77:88"), ], nwk_addresses={ t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"): zigpy_t.NWK(0x1234), t.EUI64.convert("dc:ba:00:11:22:33:44:55"): zigpy_t.NWK(0xDCBA), t.EUI64.convert("ab:cd:00:11:22:33:44:55"): zigpy_t.NWK(0xABCD), t.EUI64.convert("11:22:33:44:55:66:77:88"): zigpy_t.NWK(0x5678), }, stack_specific={"ezsp": {"hashed_tclk": b"thehashedlinkkey".hex()}}, source=f"bellows@{importlib.metadata.version('bellows')}", metadata={ "ezsp": { "stack_version": 8, "can_burn_userdata_custom_eui64": True, "can_rewrite_custom_eui64": True, } }, ), ) async def test_load_network_info( app: ControllerApplication, ieee: zigpy_t.EUI64, zigpy_backup: zigpy.backups.NetworkBackup, ) -> None: await app.load_network_info(load_devices=True) assert app.state.node_info == zigpy_backup.node_info assert app.state.network_info == zigpy_backup.network_info async def test_write_network_info( app: ControllerApplication, ieee: zigpy_t.EUI64, zigpy_backup: zigpy.backups.NetworkBackup, ) -> None: with patch.object(app, "_reset"): await app.write_network_info( node_info=zigpy_backup.node_info, network_info=zigpy_backup.network_info, ) assert app._ezsp._protocol.write_nwk_frame_counter.mock_calls == [ call(zigpy_backup.network_info.network_key.tx_counter) ] assert app._ezsp._protocol.write_aps_frame_counter.mock_calls == [ call(zigpy_backup.network_info.tc_link_key.tx_counter) ] assert app._ezsp._protocol.formNetwork.mock_calls == [ call( parameters=t.EmberNetworkParameters( panId=zigpy_backup.network_info.pan_id, extendedPanId=zigpy_backup.network_info.extended_pan_id, radioTxPower=t.uint8_t(8), radioChannel=zigpy_backup.network_info.channel, joinMethod=t.EmberJoinMethod.USE_MAC_ASSOCIATION, nwkManagerId=t.EmberNodeId(0x0000), nwkUpdateId=zigpy_backup.network_info.nwk_update_id, channels=zigpy_backup.network_info.channel_mask, ) ) ]