import asyncio from datetime import datetime, timezone import errno import logging from unittest import mock from unittest.mock import ANY, PropertyMock, call import pytest import zigpy.application import zigpy.config as conf from zigpy.exceptions import ( DeliveryError, NetworkNotFormed, NetworkSettingsInconsistent, TransientConnectionError, ) import zigpy.ota import zigpy.quirks import zigpy.types as t from zigpy.zcl import clusters, foundation import zigpy.zdo.types as zdo_t from .async_mock import AsyncMock, MagicMock, patch, sentinel from .conftest import ( NCP_IEEE, App, make_app, make_ieee, make_neighbor, make_neighbor_from_device, make_node_desc, ) @pytest.fixture def ieee(): return make_ieee() async def test_permit(app, ieee): app.devices[ieee] = MagicMock() app.devices[ieee].zdo.permit = AsyncMock() app.permit_ncp = AsyncMock() await app.permit(node=(1, 1, 1, 1, 1, 1, 1, 1)) assert app.devices[ieee].zdo.permit.call_count == 0 assert app.permit_ncp.call_count == 0 await app.permit(node=ieee) assert app.devices[ieee].zdo.permit.call_count == 1 assert app.permit_ncp.call_count == 0 await app.permit(node=NCP_IEEE) assert app.devices[ieee].zdo.permit.call_count == 1 assert app.permit_ncp.call_count == 1 async def test_permit_delivery_failure(app, ieee): def zdo_permit(*args, **kwargs): raise DeliveryError("Failed") app.devices[ieee] = MagicMock() app.devices[ieee].zdo.permit = zdo_permit app.permit_ncp = AsyncMock() await app.permit(node=ieee) assert app.permit_ncp.call_count == 0 async def test_permit_broadcast(app): app.permit_ncp = AsyncMock() app.send_packet = AsyncMock() await app.permit(time_s=30) assert app.send_packet.call_count == 1 assert app.permit_ncp.call_count == 1 assert app.send_packet.mock_calls[0].args[0].dst.addr_mode == t.AddrMode.Broadcast @patch("zigpy.device.Device.initialize", new_callable=AsyncMock) async def test_join_handler_skip(init_mock, app, ieee): node_desc = make_node_desc() app.handle_join(1, ieee, None) app.get_device(ieee).node_desc = node_desc app.handle_join(1, ieee, None) assert app.get_device(ieee).node_desc == node_desc async def test_join_handler_change_id(app, ieee): app.handle_join(1, ieee, None) app.handle_join(2, ieee, None) assert app.devices[ieee].nwk == 2 async def test_unknown_device_left(app, ieee): with patch.object(app, "listener_event", wraps=app.listener_event): app.handle_leave(0x1234, ieee) app.listener_event.assert_not_called() async def test_known_device_left(app, ieee): dev = app.add_device(ieee, 0x1234) with patch.object(app, "listener_event", wraps=app.listener_event): app.handle_leave(0x1234, ieee) app.listener_event.assert_called_once_with("device_left", dev) async def _remove( app, ieee, retval, zdo_reply=True, delivery_failure=True, has_node_desc=True ): async def leave(*args, **kwargs): if zdo_reply: return retval elif delivery_failure: raise DeliveryError("Error") else: raise asyncio.TimeoutError device = MagicMock() device.ieee = ieee device.zdo.leave.side_effect = leave if has_node_desc: device.node_desc = zdo_t.NodeDescriptor(1, 64, 142, 4388, 82, 255, 0, 255, 0) else: device.node_desc = None app.devices[ieee] = device await app.remove(ieee) for _i in range(1, 20): await asyncio.sleep(0) assert ieee not in app.devices async def test_remove(app, ieee): """Test remove with successful zdo status.""" with patch.object(app, "_remove_device", wraps=app._remove_device) as remove_device: await _remove(app, ieee, [0]) assert remove_device.await_count == 1 async def test_remove_with_failed_zdo(app, ieee): """Test remove with unsuccessful zdo status.""" with patch.object(app, "_remove_device", wraps=app._remove_device) as remove_device: await _remove(app, ieee, [1]) assert remove_device.await_count == 1 async def test_remove_nonexistent(app, ieee): with patch.object(app, "_remove_device", AsyncMock()) as remove_device: await app.remove(ieee) for _i in range(1, 20): await asyncio.sleep(0) assert ieee not in app.devices assert remove_device.await_count == 0 async def test_remove_with_unreachable_device(app, ieee): with patch.object(app, "_remove_device", wraps=app._remove_device) as remove_device: await _remove(app, ieee, [0], zdo_reply=False) assert remove_device.await_count == 1 async def test_remove_with_reply_timeout(app, ieee): with patch.object(app, "_remove_device", wraps=app._remove_device) as remove_device: await _remove(app, ieee, [0], zdo_reply=False, delivery_failure=False) assert remove_device.await_count == 1 async def test_remove_without_node_desc(app, ieee): with patch.object(app, "_remove_device", wraps=app._remove_device) as remove_device: await _remove(app, ieee, [0], has_node_desc=False) assert remove_device.await_count == 1 def test_add_device(app, ieee): app.add_device(ieee, 8) app.add_device(ieee, 9) assert app.get_device(ieee).nwk == 9 def test_get_device_nwk(app, ieee): dev = app.add_device(ieee, 8) assert app.get_device(nwk=8) is dev def test_get_device_ieee(app, ieee): dev = app.add_device(ieee, 8) assert app.get_device(ieee=ieee) is dev def test_get_device_both(app, ieee): dev = app.add_device(ieee, 8) assert app.get_device(ieee=ieee, nwk=8) is dev def test_get_device_missing(app, ieee): with pytest.raises(KeyError): app.get_device(nwk=8) def test_device_property(app): app.add_device(nwk=0x0000, ieee=NCP_IEEE) assert app._device is app.get_device(ieee=NCP_IEEE) def test_ieee(app): assert app.state.node_info.ieee def test_nwk(app): assert app.state.node_info.nwk is not None def test_config(app): assert app.config == app._config def test_deserialize(app, ieee): dev = MagicMock() app.deserialize(dev, 1, 1, b"") assert dev.deserialize.call_count == 1 @pytest.mark.filterwarnings("ignore::DeprecationWarning") async def test_handle_message_shim(app): dev = MagicMock() dev.nwk = 0x1234 app.packet_received = MagicMock(spec_set=app.packet_received) app.handle_message(dev, 260, 1, 2, 3, b"data") assert app.packet_received.mock_calls == [ call( t.ZigbeePacket( profile_id=260, cluster_id=1, src_ep=2, dst_ep=3, data=t.SerializableBytes(b"data"), src=t.AddrModeAddress( addr_mode=t.AddrMode.NWK, address=0x1234, ), dst=t.AddrModeAddress( addr_mode=t.AddrMode.NWK, address=0x0000, ), ) ) ] @patch("zigpy.device.Device.is_initialized", new_callable=PropertyMock) @patch("zigpy.quirks.handle_message_from_uninitialized_sender", new=MagicMock()) async def test_handle_message_uninitialized_dev(is_init_mock, app, ieee): dev = app.add_device(ieee, 0x1234) dev.packet_received = MagicMock() is_init_mock.return_value = False assert not dev.initializing def make_packet( profile_id: int, cluster_id: int, src_ep: int, dst_ep: int, data: bytes ) -> t.ZigbeePacket: return t.ZigbeePacket( profile_id=profile_id, cluster_id=cluster_id, src_ep=src_ep, dst_ep=dst_ep, data=t.SerializableBytes(data), src=t.AddrModeAddress( addr_mode=t.AddrMode.NWK, address=dev.nwk, ), dst=t.AddrModeAddress( addr_mode=t.AddrMode.NWK, address=0x0000, ), ) # Power Configuration cluster not allowed, no endpoints app.packet_received( make_packet(profile_id=260, cluster_id=0x0001, src_ep=1, dst_ep=1, data=b"test") ) assert dev.packet_received.call_count == 0 assert zigpy.quirks.handle_message_from_uninitialized_sender.call_count == 1 # Device should be completing initialization assert dev.initializing # ZDO is allowed app.packet_received( make_packet(profile_id=260, cluster_id=0x0000, src_ep=0, dst_ep=0, data=b"test") ) assert dev.packet_received.call_count == 1 # Endpoint is uninitialized but Basic attribute read responses still work ep = dev.add_endpoint(1) app.packet_received( make_packet(profile_id=260, cluster_id=0x0000, src_ep=1, dst_ep=1, data=b"test") ) assert dev.packet_received.call_count == 2 # Others still do not app.packet_received( make_packet(profile_id=260, cluster_id=0x0001, src_ep=1, dst_ep=1, data=b"test") ) assert dev.packet_received.call_count == 2 assert zigpy.quirks.handle_message_from_uninitialized_sender.call_count == 2 # They work after the endpoint is initialized ep.status = zigpy.endpoint.Status.ZDO_INIT app.packet_received( make_packet(profile_id=260, cluster_id=0x0001, src_ep=1, dst_ep=1, data=b"test") ) assert dev.packet_received.call_count == 3 assert zigpy.quirks.handle_message_from_uninitialized_sender.call_count == 2 def test_get_dst_address(app): r = app.get_dst_address(MagicMock()) assert r.addrmode == 3 assert r.endpoint == 1 def test_props(app): assert app.state.network_info.channel is not None assert app.state.network_info.channel_mask is not None assert app.state.network_info.extended_pan_id is not None assert app.state.network_info.pan_id is not None assert app.state.network_info.nwk_update_id is not None @pytest.mark.filterwarnings( "ignore::DeprecationWarning" ) # TODO: migrate `handle_message_from_uninitialized_sender` away from `handle_message` async def test_uninitialized_message_handlers(app, ieee): """Test uninitialized message handlers.""" handler_1 = MagicMock(return_value=None) handler_2 = MagicMock(return_value=True) zigpy.quirks.register_uninitialized_device_message_handler(handler_1) zigpy.quirks.register_uninitialized_device_message_handler(handler_2) device = app.add_device(ieee, 0x1234) app.handle_message(device, 0x0260, 0x0000, 0, 0, b"123abcd23") assert handler_1.call_count == 0 assert handler_2.call_count == 0 app.handle_message(device, 0x0260, 0x0000, 1, 1, b"123abcd23") assert handler_1.call_count == 1 assert handler_2.call_count == 1 handler_1.return_value = True app.handle_message(device, 0x0260, 0x0000, 1, 1, b"123abcd23") assert handler_1.call_count == 2 assert handler_2.call_count == 1 async def test_remove_parent_devices(app, make_initialized_device): """Test removing an end device with parents.""" end_device = make_initialized_device(app) end_device.node_desc.logical_type = zdo_t.LogicalType.EndDevice router_1 = make_initialized_device(app) router_1.node_desc.logical_type = zdo_t.LogicalType.Router router_2 = make_initialized_device(app) router_2.node_desc.logical_type = zdo_t.LogicalType.Router parent = make_initialized_device(app) app.topology.neighbors[router_1.ieee] = [ make_neighbor_from_device(router_2), make_neighbor_from_device(parent), ] app.topology.neighbors[router_2.ieee] = [ make_neighbor_from_device(parent), make_neighbor_from_device(router_1), ] app.topology.neighbors[parent.ieee] = [ make_neighbor_from_device(router_2), make_neighbor_from_device(router_1), make_neighbor_from_device(end_device), make_neighbor(ieee=make_ieee(123), nwk=0x9876), ] p1 = patch.object(end_device.zdo, "leave", AsyncMock()) p2 = patch.object(end_device.zdo, "request", AsyncMock()) p3 = patch.object(parent.zdo, "leave", AsyncMock()) p4 = patch.object(parent.zdo, "request", AsyncMock()) p5 = patch.object(router_1.zdo, "leave", AsyncMock()) p6 = patch.object(router_1.zdo, "request", AsyncMock()) p7 = patch.object(router_2.zdo, "leave", AsyncMock()) p8 = patch.object(router_2.zdo, "request", AsyncMock()) with p1, p2, p3, p4, p5, p6, p7, p8: await app.remove(end_device.ieee) for _i in range(1, 60): await asyncio.sleep(0) assert end_device.zdo.leave.await_count == 1 assert end_device.zdo.request.await_count == 0 assert router_1.zdo.leave.await_count == 0 assert router_1.zdo.request.await_count == 0 assert router_2.zdo.leave.await_count == 0 assert router_2.zdo.request.await_count == 0 assert parent.zdo.leave.await_count == 0 assert parent.zdo.request.await_count == 1 @patch("zigpy.device.Device.schedule_initialize", new_callable=MagicMock) @patch("zigpy.device.Device.schedule_group_membership_scan", new_callable=MagicMock) @patch("zigpy.device.Device.is_initialized", new_callable=PropertyMock) async def test_device_join_rejoin(is_init_mock, group_scan_mock, init_mock, app, ieee): app.listener_event = MagicMock() is_init_mock.return_value = False # First join is treated as a new join app.handle_join(0x0001, ieee, None) app.listener_event.assert_called_once_with("device_joined", ANY) app.listener_event.reset_mock() init_mock.assert_called_once() init_mock.reset_mock() # Second join with the same NWK is just a reset, not a join app.handle_join(0x0001, ieee, None) app.listener_event.assert_not_called() group_scan_mock.assert_not_called() # Since the device is still partially initialized, re-initialize it init_mock.assert_called_once() init_mock.reset_mock() # Another join with the same NWK but initialized will trigger a group re-scan is_init_mock.return_value = True app.handle_join(0x0001, ieee, None) is_init_mock.return_value = True app.listener_event.assert_not_called() group_scan_mock.assert_called_once() group_scan_mock.reset_mock() init_mock.assert_not_called() # Join with a different NWK but the same IEEE is a re-join app.handle_join(0x0002, ieee, None) app.listener_event.assert_called_once_with("device_joined", ANY) group_scan_mock.assert_not_called() init_mock.assert_called_once() async def test_get_device(app): """Test get_device.""" await app.startup() app.add_device(t.EUI64.convert("11:11:11:11:22:22:22:22"), 0x0000) dev_2 = app.add_device(app.state.node_info.ieee, 0x0000) app.add_device(t.EUI64.convert("11:11:11:11:22:22:22:33"), 0x0000) assert app.get_device(nwk=0x0000) is dev_2 async def test_probe_success(): config = {"path": "/dev/test"} with ( patch.object(App, "connect") as connect, patch.object(App, "disconnect") as disconnect, ): result = await App.probe(config) assert set(config.items()) <= set(result.items()) assert connect.await_count == 1 assert disconnect.await_count == 1 async def test_probe_failure(): config = {"path": "/dev/test"} with ( patch.object(App, "connect", side_effect=asyncio.TimeoutError) as connect, patch.object(App, "disconnect") as disconnect, ): result = await App.probe(config) assert result is False assert connect.await_count == 1 assert disconnect.await_count == 1 async def test_form_network(app): with patch.object(app, "write_network_info") as write1: await app.form_network() with patch.object(app, "write_network_info") as write2: await app.form_network() nwk_info1 = write1.mock_calls[0].kwargs["network_info"] node_info1 = write1.mock_calls[0].kwargs["node_info"] nwk_info2 = write2.mock_calls[0].kwargs["network_info"] node_info2 = write2.mock_calls[0].kwargs["node_info"] assert node_info1 == node_info2 # Critical network settings are randomized assert nwk_info1.extended_pan_id != nwk_info2.extended_pan_id assert nwk_info1.pan_id != nwk_info2.pan_id assert nwk_info1.network_key != nwk_info2.network_key # The well-known TCLK is used assert ( nwk_info1.tc_link_key.key == nwk_info2.tc_link_key.key == t.KeyData(b"ZigBeeAlliance09") ) assert nwk_info1.channel in (11, 15, 20, 25) @mock.patch("zigpy.util.pick_optimal_channel", mock.Mock(return_value=22)) async def test_form_network_find_best_channel(app): orig_start_network = app.start_network async def start_network(*args, **kwargs): start_network.await_count += 1 if start_network.await_count == 1: raise NetworkNotFormed return await orig_start_network(*args, **kwargs) start_network.await_count = 0 app.start_network = start_network with patch.object(app, "write_network_info") as write: with patch.object( app.backups, "create_backup", wraps=app.backups.create_backup ) as create_backup: await app.form_network() assert start_network.await_count == 2 # A temporary network will be formed first nwk_info1 = write.mock_calls[0].kwargs["network_info"] assert nwk_info1.channel == 11 # Then, after the scan, a better channel is chosen nwk_info2 = write.mock_calls[1].kwargs["network_info"] assert nwk_info2.channel == 22 # Only a single backup will be present assert create_backup.await_count == 1 async def test_startup_formed(): app = make_app({}) app.start_network = AsyncMock(wraps=app.start_network) app.form_network = AsyncMock() app.permit = AsyncMock() await app.startup(auto_form=False) assert app.start_network.await_count == 1 assert app.form_network.await_count == 0 assert app.permit.await_count == 1 async def test_startup_not_formed(): app = make_app({}) app.start_network = AsyncMock(wraps=app.start_network) app.form_network = AsyncMock() app.load_network_info = AsyncMock( side_effect=[NetworkNotFormed(), NetworkNotFormed(), None] ) app.permit = AsyncMock() app.backups.backups = [] app.backups.restore_backup = AsyncMock() with pytest.raises(NetworkNotFormed): await app.startup(auto_form=False) assert app.start_network.await_count == 0 assert app.form_network.await_count == 0 assert app.permit.await_count == 0 await app.startup(auto_form=True) assert app.start_network.await_count == 1 assert app.form_network.await_count == 1 assert app.permit.await_count == 1 assert app.backups.restore_backup.await_count == 0 async def test_startup_not_formed_with_backup(): app = make_app({}) app.start_network = AsyncMock(wraps=app.start_network) app.load_network_info = AsyncMock(side_effect=[NetworkNotFormed(), None]) app.permit = AsyncMock() app.backups.restore_backup = AsyncMock() app.backups.backups = [sentinel.OLD_BACKUP, sentinel.NEW_BACKUP] await app.startup(auto_form=True) assert app.start_network.await_count == 1 app.backups.restore_backup.assert_called_once_with(sentinel.NEW_BACKUP) async def test_startup_backup(): app = make_app({conf.CONF_NWK_BACKUP_ENABLED: True}) with patch("zigpy.backups.BackupManager.start_periodic_backups") as p: await app.startup() p.assert_called_once() async def test_startup_no_backup(): app = make_app({conf.CONF_NWK_BACKUP_ENABLED: False}) with patch("zigpy.backups.BackupManager.start_periodic_backups") as p: await app.startup() p.assert_not_called() def with_attributes(obj, **attrs): for k, v in attrs.items(): setattr(obj, k, v) return obj @pytest.mark.parametrize( "error", [ with_attributes(OSError("Network is unreachable"), errno=errno.ENETUNREACH), ConnectionRefusedError(), ], ) async def test_startup_failure_transient_error(error): app = make_app({conf.CONF_NWK_BACKUP_ENABLED: False}) with patch.object(app, "connect", side_effect=[error]): with pytest.raises(TransientConnectionError): await app.startup() @patch("zigpy.backups.BackupManager.from_network_state") @patch("zigpy.backups.BackupManager.most_recent_backup") async def test_initialize_compatible_backup( mock_most_recent_backup, mock_backup_from_state ): app = make_app({conf.CONF_NWK_VALIDATE_SETTINGS: True}) mock_backup_from_state.return_value.is_compatible_with.return_value = True await app.initialize() mock_backup_from_state.return_value.is_compatible_with.assert_called_once() mock_most_recent_backup.assert_called_once() @patch("zigpy.backups.BackupManager.from_network_state") @patch("zigpy.backups.BackupManager.most_recent_backup") async def test_initialize_incompatible_backup( mock_most_recent_backup, mock_backup_from_state ): app = make_app({conf.CONF_NWK_VALIDATE_SETTINGS: True}) mock_backup_from_state.return_value.is_compatible_with.return_value = False with pytest.raises(NetworkSettingsInconsistent) as exc: await app.initialize() mock_backup_from_state.return_value.is_compatible_with.assert_called_once() mock_most_recent_backup.assert_called_once() assert exc.value.old_state is mock_most_recent_backup() assert exc.value.new_state is mock_backup_from_state.return_value async def test_relays_received_device_exists(app): device = MagicMock() app._discover_unknown_device = AsyncMock(spec_set=app._discover_unknown_device) app.get_device = MagicMock(spec_set=app.get_device, return_value=device) app.handle_relays(nwk=0x1234, relays=[0x5678, 0xABCD]) app.get_device.assert_called_once_with(nwk=0x1234) assert device.relays == [0x5678, 0xABCD] assert app._discover_unknown_device.call_count == 0 async def test_relays_received_device_does_not_exist(app): app._discover_unknown_device = AsyncMock(spec_set=app._discover_unknown_device) app.get_device = MagicMock(wraps=app.get_device) app.handle_relays(nwk=0x1234, relays=[0x5678, 0xABCD]) app.get_device.assert_called_once_with(nwk=0x1234) app._discover_unknown_device.assert_called_once_with(nwk=0x1234) async def test_request_concurrency(): current_concurrency = 0 peak_concurrency = 0 class SlowApp(App): async def send_packet(self, packet): nonlocal current_concurrency, peak_concurrency async with self._limit_concurrency(): current_concurrency += 1 peak_concurrency = max(peak_concurrency, current_concurrency) await asyncio.sleep(0.1) current_concurrency -= 1 if packet % 10 == 7: # Fail randomly raise DeliveryError("Failure") app = make_app({conf.CONF_MAX_CONCURRENT_REQUESTS: 16}, app_base=SlowApp) assert current_concurrency == 0 assert peak_concurrency == 0 await asyncio.gather( *[app.send_packet(i) for i in range(100)], return_exceptions=True ) assert current_concurrency == 0 assert peak_concurrency == 16 @pytest.fixture def device(): device = MagicMock() device.nwk = 0xABCD device.ieee = t.EUI64.convert("aa:bb:cc:dd:11:22:33:44") return device @pytest.fixture def packet(app, device): return t.ZigbeePacket( src=t.AddrModeAddress( addr_mode=t.AddrMode.NWK, address=app.state.node_info.nwk ), src_ep=0x9A, dst=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=device.nwk), dst_ep=0xBC, tsn=0xDE, profile_id=0x1234, cluster_id=0x0006, data=t.SerializableBytes(b"test data"), source_route=None, extended_timeout=False, tx_options=t.TransmitOptions.NONE, ) async def test_request(app, device, packet): app.build_source_route_to = MagicMock(spec_set=app.build_source_route_to) async def send_request(app, **kwargs): kwargs = { "device": device, "profile": 0x1234, "cluster": 0x0006, "src_ep": 0x9A, "dst_ep": 0xBC, "sequence": 0xDE, "data": b"test data", "expect_reply": True, "use_ieee": False, "extended_timeout": False, **kwargs, } return await app.request(**kwargs) # Test sending with NWK status, msg = await send_request(app) assert status == zigpy.zcl.foundation.Status.SUCCESS assert isinstance(msg, str) app.send_packet.assert_called_once_with(packet) app.send_packet.reset_mock() # Test sending with IEEE await send_request(app, use_ieee=True) app.send_packet.assert_called_once_with( packet.replace( src=t.AddrModeAddress( addr_mode=t.AddrMode.IEEE, address=app.state.node_info.ieee, ), dst=t.AddrModeAddress( addr_mode=t.AddrMode.IEEE, address=device.ieee, ), ) ) app.send_packet.reset_mock() # Test sending with source route app.build_source_route_to.return_value = [0x000A, 0x000B] with patch.dict(app.config, {conf.CONF_SOURCE_ROUTING: True}): await send_request(app) app.build_source_route_to.assert_called_once_with(dest=device) app.send_packet.assert_called_once_with( packet.replace(source_route=[0x000A, 0x000B]) ) app.send_packet.reset_mock() # Test sending without waiting for a reply status, msg = await send_request(app, expect_reply=False) app.send_packet.assert_called_once_with( packet.replace(tx_options=t.TransmitOptions.ACK) ) app.send_packet.reset_mock() # Test explicit ACK control (enabled) status, msg = await send_request(app, ask_for_ack=True) app.send_packet.assert_called_once_with( packet.replace(tx_options=t.TransmitOptions.ACK) ) app.send_packet.reset_mock() # Test explicit ACK control (disabled) status, msg = await send_request(app, ask_for_ack=False) app.send_packet.assert_called_once_with( packet.replace(tx_options=t.TransmitOptions(0)) ) app.send_packet.reset_mock() async def test_request_retrying_success(app, device, packet) -> None: app.send_packet.side_effect = [ DeliveryError("Failure"), DeliveryError("Failure"), None, ] await app.request( device=device, profile=0x1234, cluster=0x0006, src_ep=0x9A, dst_ep=0xBC, sequence=0xDE, data=b"test data", expect_reply=True, use_ieee=False, extended_timeout=False, ) assert app.send_packet.mock_calls == [ call(packet), call( packet.replace( tx_options=packet.tx_options | t.TransmitOptions.FORCE_ROUTE_DISCOVERY ) ), call( packet.replace( tx_options=packet.tx_options | t.TransmitOptions.FORCE_ROUTE_DISCOVERY ) ), ] async def test_request_retrying_failure(app, device, packet) -> None: app.send_packet.side_effect = [ DeliveryError("Failure"), DeliveryError("Failure"), DeliveryError("Failure"), ] with pytest.raises(DeliveryError): await app.request( device=device, profile=0x1234, cluster=0x0006, src_ep=0x9A, dst_ep=0xBC, sequence=0xDE, data=b"test data", expect_reply=True, use_ieee=False, extended_timeout=False, ) assert app.send_packet.mock_calls == [ call(packet), call( packet.replace( tx_options=packet.tx_options | t.TransmitOptions.FORCE_ROUTE_DISCOVERY ) ), call( packet.replace( tx_options=packet.tx_options | t.TransmitOptions.FORCE_ROUTE_DISCOVERY ) ), ] def test_build_source_route_has_relays(app): device = MagicMock() device.relays = [0x1234, 0x5678] assert app.build_source_route_to(device) == [0x5678, 0x1234] def test_build_source_route_no_relays(app): device = MagicMock() device.relays = None assert app.build_source_route_to(device) is None async def test_send_mrequest(app, packet): status, msg = await app.mrequest( group_id=0xABCD, profile=0x1234, cluster=0x0006, src_ep=0x9A, sequence=0xDE, data=b"test data", hops=12, non_member_radius=34, ) assert status == zigpy.zcl.foundation.Status.SUCCESS assert isinstance(msg, str) app.send_packet.assert_called_once_with( packet.replace( dst=t.AddrModeAddress(addr_mode=t.AddrMode.Group, address=0xABCD), dst_ep=None, radius=12, non_member_radius=34, tx_options=t.TransmitOptions.NONE, ) ) async def test_send_broadcast(app, packet): status, msg = await app.broadcast( profile=0x1234, cluster=0x0006, src_ep=0x9A, dst_ep=0xBC, grpid=0x0000, # unused radius=12, sequence=0xDE, data=b"test data", broadcast_address=t.BroadcastAddress.RX_ON_WHEN_IDLE, ) assert status == zigpy.zcl.foundation.Status.SUCCESS assert isinstance(msg, str) app.send_packet.assert_called_once_with( packet.replace( dst=t.AddrModeAddress( addr_mode=t.AddrMode.Broadcast, address=t.BroadcastAddress.RX_ON_WHEN_IDLE, ), radius=12, tx_options=t.TransmitOptions.NONE, ) ) @pytest.fixture def zdo_packet(app, device): return t.ZigbeePacket( src=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=device.nwk), dst=t.AddrModeAddress( addr_mode=t.AddrMode.NWK, address=app.state.node_info.nwk ), src_ep=0x00, # ZDO dst_ep=0x00, tsn=0xDE, profile_id=0x0000, cluster_id=0x0000, data=t.SerializableBytes(b""), source_route=None, extended_timeout=False, tx_options=t.TransmitOptions.ACK, lqi=123, rssi=-80, ) @patch("zigpy.device.Device.initialize", AsyncMock()) async def test_packet_received_new_device_zdo_announce(app, device, zdo_packet): app.handle_join = MagicMock(wraps=app.handle_join) zdo_data = zigpy.zdo.ZDO(None)._serialize( zdo_t.ZDOCmd.Device_annce, *{ "NWKAddr": device.nwk, "IEEEAddr": device.ieee, "Capability": 0x00, }.values(), ) zdo_packet.cluster_id = zdo_t.ZDOCmd.Device_annce zdo_packet.data = t.SerializableBytes( t.uint8_t(zdo_packet.tsn).serialize() + zdo_data ) app.packet_received(zdo_packet) app.handle_join.assert_called_once_with( nwk=device.nwk, ieee=device.ieee, parent_nwk=None ) zigpy_device = app.get_device(ieee=device.ieee) assert zigpy_device.lqi == zdo_packet.lqi assert zigpy_device.rssi == zdo_packet.rssi @patch("zigpy.device.Device.initialize", AsyncMock()) async def test_packet_received_new_device_discovery(app, device, zdo_packet): app.handle_join = MagicMock(wraps=app.handle_join) async def send_packet(packet): if packet.dst_ep != 0x00 or packet.cluster_id != zdo_t.ZDOCmd.IEEE_addr_req: return hdr, args = zigpy.zdo.ZDO(None).deserialize( packet.cluster_id, packet.data.serialize() ) assert args == list( { "NWKAddrOfInterest": device.nwk, "RequestType": zdo_t.AddrRequestType.Single, "StartIndex": 0, }.values() ) zdo_data = zigpy.zdo.ZDO(None)._serialize( zdo_t.ZDOCmd.IEEE_addr_rsp, *{ "Status": zdo_t.Status.SUCCESS, "IEEEAddr": device.ieee, "NWKAddr": device.nwk, "NumAssocDev": 0, "StartIndex": 0, "NWKAddrAssocDevList": [], }.values(), ) # Receive the IEEE address reply zdo_packet.data = t.SerializableBytes( t.uint8_t(zdo_packet.tsn).serialize() + zdo_data ) zdo_packet.cluster_id = zdo_t.ZDOCmd.IEEE_addr_rsp app.packet_received(zdo_packet) app.send_packet = AsyncMock(side_effect=send_packet) # Receive a bogus packet first, to trigger device discovery bogus_packet = zdo_packet.replace(dst_ep=0x01, src_ep=0x01) app.packet_received(bogus_packet) await asyncio.sleep(0.1) app.handle_join.assert_called_once_with( nwk=device.nwk, ieee=device.ieee, parent_nwk=None, handle_rejoin=False ) zigpy_device = app.get_device(ieee=device.ieee) assert zigpy_device.lqi == zdo_packet.lqi assert zigpy_device.rssi == zdo_packet.rssi @patch("zigpy.device.Device.initialize", AsyncMock()) async def test_packet_received_ieee_no_rejoin(app, device, zdo_packet, caplog): device.is_initialized = True app.devices[device.ieee] = device app.handle_join = MagicMock(wraps=app.handle_join) zdo_data = zigpy.zdo.ZDO(None)._serialize( zdo_t.ZDOCmd.IEEE_addr_rsp, *{ "Status": zdo_t.Status.SUCCESS, "IEEEAddr": device.ieee, "NWKAddr": device.nwk, }.values(), ) zdo_packet.cluster_id = zdo_t.ZDOCmd.IEEE_addr_rsp zdo_packet.data = t.SerializableBytes( t.uint8_t(zdo_packet.tsn).serialize() + zdo_data ) app.packet_received(zdo_packet) assert "joined the network" not in caplog.text app.handle_join.assert_called_once_with( nwk=device.nwk, ieee=device.ieee, parent_nwk=None, handle_rejoin=False ) assert len(device.schedule_group_membership_scan.mock_calls) == 0 assert len(device.schedule_initialize.mock_calls) == 0 @patch("zigpy.device.Device.initialize", AsyncMock()) async def test_packet_received_ieee_rejoin(app, device, zdo_packet, caplog): device.is_initialized = True app.devices[device.ieee] = device app.handle_join = MagicMock(wraps=app.handle_join) zdo_data = zigpy.zdo.ZDO(None)._serialize( zdo_t.ZDOCmd.IEEE_addr_rsp, *{ "Status": zdo_t.Status.SUCCESS, "IEEEAddr": device.ieee, "NWKAddr": device.nwk + 1, # NWK has changed }.values(), ) zdo_packet.cluster_id = zdo_t.ZDOCmd.IEEE_addr_rsp zdo_packet.data = t.SerializableBytes( t.uint8_t(zdo_packet.tsn).serialize() + zdo_data ) app.packet_received(zdo_packet) assert "joined the network" not in caplog.text app.handle_join.assert_called_once_with( nwk=device.nwk, ieee=device.ieee, parent_nwk=None, handle_rejoin=False ) assert len(device.schedule_initialize.mock_calls) == 1 async def test_bad_zdo_packet_received(app, device): device.is_initialized = True app.devices[device.ieee] = device bogus_zdo_packet = t.ZigbeePacket( src=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=device.nwk), src_ep=1, dst=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=0x0000), dst_ep=0, # bad destination endpoint tsn=180, profile_id=260, cluster_id=6, data=t.SerializableBytes(b"\x08n\n\x00\x00\x10\x00"), lqi=255, rssi=-30, ) app.packet_received(bogus_zdo_packet) assert len(device.packet_received.mock_calls) == 1 def test_get_device_with_address_nwk(app, device): app.devices[device.ieee] = device assert ( app.get_device_with_address( t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=device.nwk) ) is device ) assert ( app.get_device_with_address( t.AddrModeAddress(addr_mode=t.AddrMode.IEEE, address=device.ieee) ) is device ) with pytest.raises(ValueError): app.get_device_with_address( t.AddrModeAddress(addr_mode=t.AddrMode.Group, address=device.nwk) ) with pytest.raises(KeyError): app.get_device_with_address( t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=device.nwk + 1) ) async def test_request_future_matching(app, make_initialized_device): device = make_initialized_device(app) device._packet_debouncer.filter = MagicMock(return_value=False) ota = device.endpoints[1].add_output_cluster(clusters.general.Ota.cluster_id) req_hdr, req_cmd = ota._create_request( general=False, command_id=ota.commands_by_name["query_next_image"].id, schema=ota.commands_by_name["query_next_image"].schema, disable_default_response=False, direction=foundation.Direction.Client_to_Server, args=(), kwargs={ "field_control": 0, "manufacturer_code": 0x1234, "image_type": 0x5678, "current_file_version": 0x11112222, }, ) packet = t.ZigbeePacket( src=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=device.nwk), src_ep=1, dst=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=0x0000), dst_ep=1, tsn=req_hdr.tsn, profile_id=260, cluster_id=ota.cluster_id, data=t.SerializableBytes(req_hdr.serialize() + req_cmd.serialize()), lqi=255, rssi=-30, ) assert not app._req_listeners[device] with app.wait_for_response( device, [ota.commands_by_name["query_next_image"].schema()] ) as rsp_fut: # Attach two listeners with app.wait_for_response( device, [ota.commands_by_name["query_next_image"].schema()] ) as rsp_fut2: assert app._req_listeners[device] # Listeners are resolved FIFO app.packet_received(packet) assert rsp_fut.done() assert not rsp_fut2.done() app.packet_received(packet) assert rsp_fut.done() assert rsp_fut2.done() # Unhandled packets are ignored app.packet_received(packet) rsp_hdr, rsp_cmd = await rsp_fut assert rsp_hdr == req_hdr assert rsp_cmd == req_cmd assert rsp_cmd.current_file_version == 0x11112222 assert not app._req_listeners[device] async def test_request_callback_matching(app, make_initialized_device): device = make_initialized_device(app) device._packet_debouncer.filter = MagicMock(return_value=False) ota = device.endpoints[1].add_output_cluster(clusters.general.Ota.cluster_id) req_hdr, req_cmd = ota._create_request( general=False, command_id=ota.commands_by_name["query_next_image"].id, schema=ota.commands_by_name["query_next_image"].schema, disable_default_response=False, direction=foundation.Direction.Client_to_Server, args=(), kwargs={ "field_control": 0, "manufacturer_code": 0x1234, "image_type": 0x5678, "current_file_version": 0x11112222, }, ) packet = t.ZigbeePacket( src=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=device.nwk), src_ep=1, dst=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=0x0000), dst_ep=1, tsn=req_hdr.tsn, profile_id=260, cluster_id=ota.cluster_id, data=t.SerializableBytes(req_hdr.serialize() + req_cmd.serialize()), lqi=255, rssi=-30, ) mock_callback = mock.Mock() assert not app._req_listeners[device] with app.callback_for_response( device, [ota.commands_by_name["query_next_image"].schema()], mock_callback ): assert app._req_listeners[device] asyncio.get_running_loop().call_soon(app.packet_received, packet) asyncio.get_running_loop().call_soon(app.packet_received, packet) asyncio.get_running_loop().call_soon(app.packet_received, packet) await asyncio.sleep(0.1) assert len(mock_callback.mock_calls) == 3 assert mock_callback.mock_calls == [mock.call(req_hdr, req_cmd)] * 3 assert not app._req_listeners[device] async def test_energy_scan_default(app): await app.startup() raw_scan_results = [ 170, 191, 181, 165, 179, 169, 196, 163, 174, 162, 190, 186, 191, 178, 204, 187, ] coordinator = app._device coordinator.zdo.Mgmt_NWK_Update_req = AsyncMock( return_value=[ zdo_t.Status.SUCCESS, t.Channels.ALL_CHANNELS, 29, 10, raw_scan_results, ] ) results = await app.energy_scan( channels=t.Channels.ALL_CHANNELS, duration_exp=2, count=1 ) assert len(results) == 16 assert results == dict(zip(range(11, 26 + 1), raw_scan_results)) async def test_energy_scan_not_implemented(app): """Energy scanning still "works" even when the radio doesn't implement it.""" await app.startup() app._device.zdo.Mgmt_NWK_Update_req.side_effect = asyncio.TimeoutError() results = await app.energy_scan( channels=t.Channels.ALL_CHANNELS, duration_exp=2, count=1 ) assert results == {c: 0 for c in range(11, 26 + 1)} async def test_startup_broadcast_failure_due_to_interference(app, caplog): err = DeliveryError( "Failed to deliver packet: ", 225 ) with mock.patch.object(app, "permit", side_effect=err): with caplog.at_level(logging.WARNING): await app.startup() # The application will still start up, however assert "Failed to send startup broadcast" in caplog.text assert "interference" in caplog.text async def test_startup_broadcast_failure_other(app, caplog): with mock.patch.object(app, "permit", side_effect=DeliveryError("Error", 123)): with pytest.raises(DeliveryError, match="^Error$"): await app.startup() @patch("zigpy.application.CHANNEL_CHANGE_SETTINGS_RELOAD_DELAY_S", 0.1) @patch("zigpy.application.CHANNEL_CHANGE_BROADCAST_DELAY_S", 0.01) async def test_move_network_to_new_channel(app): async def nwk_update(*args, **kwargs): async def inner(): await asyncio.sleep( zigpy.application.CHANNEL_CHANGE_SETTINGS_RELOAD_DELAY_S * 5 ) NwkUpdate = args[0] app.state.network_info.channel = list(NwkUpdate.ScanChannels)[0] app.state.network_info.nwk_update_id = NwkUpdate.nwkUpdateId asyncio.create_task(inner()) # noqa: RUF006 await app.startup() assert app.state.network_info.channel != 26 with patch.object( app._device.zdo, "Mgmt_NWK_Update_req", side_effect=nwk_update ) as mock_update: await app.move_network_to_channel(new_channel=26, num_broadcasts=10) assert app.state.network_info.channel == 26 assert len(mock_update.mock_calls) == 1 async def test_move_network_to_new_channel_noop(app): await app.startup() old_channel = app.state.network_info.channel with patch("zigpy.zdo.broadcast") as mock_broadcast: await app.move_network_to_channel(new_channel=old_channel) assert app.state.network_info.channel == old_channel assert len(mock_broadcast.mock_calls) == 0 async def test_startup_multiple_dblistener(app): app._dblistener = AsyncMock() app.connect = AsyncMock(side_effect=RuntimeError()) with pytest.raises(RuntimeError): await app.startup() with pytest.raises(RuntimeError): await app.startup() # The database listener will not be shut down automatically assert len(app._dblistener.shutdown.mock_calls) == 0 async def test_connection_lost(app): exc = RuntimeError() listener = MagicMock() app.add_listener(listener) app.connection_lost(exc) listener.connection_lost.assert_called_with(exc) async def test_watchdog(app): error = RuntimeError() app = make_app({}) app._watchdog_period = 0.1 app._watchdog_feed = AsyncMock(side_effect=[None, None, error]) app.connection_lost = MagicMock() assert app._watchdog_task is None await app.startup() assert app._watchdog_task is not None # We call it once during startup synchronously assert app._watchdog_feed.mock_calls == [call()] assert app.connection_lost.mock_calls == [] await asyncio.sleep(0.5) assert app._watchdog_feed.mock_calls == [call(), call(), call()] assert app.connection_lost.mock_calls == [call(error)] assert app._watchdog_task.done() @pytest.mark.filterwarnings("ignore::DeprecationWarning") async def test_permit_with_key(app): app = make_app({}) app.permit_with_link_key = AsyncMock() with pytest.raises(ValueError): await app.permit_with_key( node=t.EUI64.convert("aa:bb:cc:dd:11:22:33:44"), code=b"invalid code that is far too long and of the wrong parity", time_s=60, ) assert app.permit_with_link_key.mock_calls == [] await app.permit_with_key( node=t.EUI64.convert("aa:bb:cc:dd:11:22:33:44"), code=bytes.fromhex("11223344556677884AF7"), time_s=60, ) assert app.permit_with_link_key.mock_calls == [ call( node=t.EUI64.convert("aa:bb:cc:dd:11:22:33:44"), link_key=t.KeyData.convert("41618FC0C83B0E14A589954B16E31466"), time_s=60, ) ] async def test_probe(app): class BaudSpecificApp(App): _probe_configs = [ {conf.CONF_DEVICE_BAUDRATE: 57600}, {conf.CONF_DEVICE_BAUDRATE: 115200}, ] async def connect(self): if self._config[conf.CONF_DEVICE][conf.CONF_DEVICE_BAUDRATE] != 115200: raise asyncio.TimeoutError # Only one baudrate is valid assert (await BaudSpecificApp.probe({conf.CONF_DEVICE_PATH: "/dev/null"})) == { conf.CONF_DEVICE_PATH: "/dev/null", conf.CONF_DEVICE_BAUDRATE: 115200, conf.CONF_DEVICE_FLOW_CONTROL: None, } class NeverConnectsApp(App): async def connect(self): raise asyncio.TimeoutError # No settings will work assert (await NeverConnectsApp.probe({conf.CONF_DEVICE_PATH: "/dev/null"})) is False async def test_network_scan(app) -> None: beacons = [ t.NetworkBeacon( pan_id=t.NWK(0x1234), extended_pan_id=t.EUI64.convert("11:22:33:44:55:66:77:88"), channel=11, nwk_update_id=1, permit_joining=True, stack_profile=2, lqi=255, rssi=-80, ), t.NetworkBeacon( pan_id=t.NWK(0xABCD), extended_pan_id=t.EUI64.convert("11:22:33:44:55:66:77:88"), channel=15, nwk_update_id=2, permit_joining=False, stack_profile=2, lqi=255, rssi=-40, ), ] with patch.object(app, "_network_scan") as mock_scan: mock_scan.return_value.__aiter__.return_value = beacons results = [ b async for b in app.network_scan( channels=t.Channels.from_channel_list([11, 15]), duration_exp=1, ) ] assert results == beacons assert mock_scan.mock_calls == [ call( channels=t.Channels.from_channel_list([11, 15]), duration_exp=1, ), call().__aiter__(), ] async def test_packet_capture(app) -> None: packets = [ t.CapturedPacket( timestamp=datetime(2021, 1, 1, 0, 0, 0, tzinfo=timezone.utc), rssi=-60, lqi=250, channel=15, data=bytes.fromhex("02007f"), ), t.CapturedPacket( timestamp=datetime(2021, 1, 1, 0, 0, 1, tzinfo=timezone.utc), rssi=-70, lqi=240, channel=15, data=bytes.fromhex( "61886fefbe445600004802653c00001e1228eea3dd0046b8a11c004b120000631ea30c" "f9079829433d9b6165c3b56171df2557407024" ), ), ] with patch.object(app, "_packet_capture") as mock_capture: mock_capture.return_value.__aiter__.return_value = packets results = [p async for p in app.packet_capture(channel=15)] assert results == packets assert packets[0].compute_fcs() == b"\xc8\x3e" assert packets[1].compute_fcs() == b"\x63\x7d" with patch.object(app, "_packet_capture_change_channel"): await app.packet_capture_change_channel(channel=25) assert app._packet_capture_change_channel.mock_calls == [call(channel=25)]