import asyncio from unittest.mock import AsyncMock, MagicMock, call, patch, sentinel import pytest from zigpy import endpoint, group, zcl import zigpy.device import zigpy.exceptions import zigpy.types as t from zigpy.zcl.foundation import GENERAL_COMMANDS, GeneralCommand, Status as ZCLStatus from zigpy.zdo import types @pytest.fixture def ep(): dev = MagicMock() dev.request = AsyncMock() dev.reply = AsyncMock() return endpoint.Endpoint(dev, 1) async def _test_initialize(ep, profile): async def mockrequest(*args, **kwargs): sd = types.SimpleDescriptor() sd.endpoint = 1 sd.profile = profile sd.device_type = 0xFF sd.input_clusters = [5] sd.output_clusters = [6] return [0, None, sd] ep._device.zdo.Simple_Desc_req = mockrequest await ep.initialize() assert ep.status > endpoint.Status.NEW assert 5 in ep.in_clusters assert 6 in ep.out_clusters async def test_inactive_initialize(ep): async def mockrequest(*args, **kwargs): sd = types.SimpleDescriptor() sd.endpoint = 2 return [131, None, sd] ep._device.zdo.Simple_Desc_req = mockrequest await ep.initialize() assert ep.status == endpoint.Status.ENDPOINT_INACTIVE async def test_initialize_zha(ep): return await _test_initialize(ep, 260) async def test_initialize_zll(ep): return await _test_initialize(ep, 49246) async def test_initialize_other(ep): return await _test_initialize(ep, 0x1234) async def test_initialize_fail(ep): async def mockrequest(*args, **kwargs): return [1, None, None] ep._device.zdo.Simple_Desc_req = mockrequest # The request succeeds but the response is invalid with pytest.raises(zigpy.exceptions.InvalidResponse): await ep.initialize() assert ep.status == endpoint.Status.NEW async def test_reinitialize(ep): await _test_initialize(ep, 260) assert ep.profile_id == 260 ep.profile_id = 10 await _test_initialize(ep, 260) assert ep.profile_id == 10 def test_add_input_cluster(ep): ep.add_input_cluster(0) assert 0 in ep.in_clusters assert ep.in_clusters[0].is_server is True assert ep.in_clusters[0].is_client is False def test_add_custom_input_cluster(ep): mock_cluster = MagicMock() ep.add_input_cluster(0, mock_cluster) assert 0 in ep.in_clusters assert ep.in_clusters[0] is mock_cluster def test_add_output_cluster(ep): ep.add_output_cluster(0) assert 0 in ep.out_clusters assert ep.out_clusters[0].is_server is False assert ep.out_clusters[0].is_client is True def test_add_custom_output_cluster(ep): mock_cluster = MagicMock() ep.add_output_cluster(0, mock_cluster) assert 0 in ep.out_clusters assert ep.out_clusters[0] is mock_cluster def test_multiple_add_input_cluster(ep): ep.add_input_cluster(0) assert ep.in_clusters[0].cluster_id == 0 ep.in_clusters[0].cluster_id = 1 assert ep.in_clusters[0].cluster_id == 1 ep.add_input_cluster(0) assert ep.in_clusters[0].cluster_id == 1 def test_multiple_add_output_cluster(ep): ep.add_output_cluster(0) assert ep.out_clusters[0].cluster_id == 0 ep.out_clusters[0].cluster_id = 1 assert ep.out_clusters[0].cluster_id == 1 ep.add_output_cluster(0) assert ep.out_clusters[0].cluster_id == 1 def test_handle_message(ep): c = ep.add_input_cluster(0) c.handle_message = MagicMock() ep.handle_message(sentinel.profile, 0, sentinel.hdr, sentinel.data) c.handle_message.assert_called_once_with( sentinel.hdr, sentinel.data, dst_addressing=None ) def test_handle_message_output(ep): c = ep.add_output_cluster(0) c.handle_message = MagicMock() ep.handle_message(sentinel.profile, 0, sentinel.hdr, sentinel.data) c.handle_message.assert_called_once_with( sentinel.hdr, sentinel.data, dst_addressing=None ) def test_handle_request_unknown(ep): hdr = MagicMock() hdr.command_id = sentinel.command_id ep.handle_message(sentinel.profile, 99, hdr, sentinel.args) def test_cluster_attr(ep): with pytest.raises(AttributeError): ep.basic # noqa: B018 ep.add_input_cluster(0) assert ep.basic is not None async def test_request(ep): ep.profile_id = 260 await ep.request(7, 8, b"") assert ep._device.request.call_count == 1 assert ep._device.request.await_count == 1 async def test_request_change_profileid(ep): ep.profile_id = 49246 await ep.request(7, 9, b"") ep.profile_id = 49246 await ep.request(0x1000, 10, b"") ep.profile_id = 260 await ep.request(0x1000, 11, b"") assert ep._device.request.call_count == 3 assert ep._device.request.await_count == 3 async def test_reply(ep): ep.profile_id = 260 await ep.reply(7, 8, b"") assert ep._device.reply.call_count == 1 async def test_reply_change_profile_id(ep): ep.profile_id = 49246 await ep.reply(cluster=0x1000, sequence=8, data=b"", command_id=0x3F) assert ep._device.reply.mock_calls == [ call( profile=49246, cluster=0x1000, src_ep=1, dst_ep=1, sequence=8, data=b"", timeout=5, expect_reply=False, use_ieee=False, ask_for_ack=None, priority=t.PacketPriority.NORMAL, ) ] ep._device.reply.reset_mock() await ep.reply(cluster=0x1000, sequence=8, data=b"", command_id=0x40) assert ep._device.reply.mock_calls == [ call( profile=0x0104, cluster=0x1000, src_ep=1, dst_ep=1, sequence=8, data=b"", timeout=5, expect_reply=False, use_ieee=False, ask_for_ack=None, priority=t.PacketPriority.NORMAL, ) ] ep._device.reply.reset_mock() ep.profile_id = 0xBEEF await ep.reply(cluster=0x1000, sequence=8, data=b"", command_id=0x40) assert ep._device.reply.mock_calls == [ call( profile=0xBEEF, cluster=0x1000, src_ep=1, dst_ep=1, sequence=8, data=b"", timeout=5, expect_reply=False, use_ieee=False, ask_for_ack=None, priority=t.PacketPriority.NORMAL, ) ] def _mk_rar(attrid, value, status=0): r = zcl.foundation.ReadAttributeRecord() r.attrid = attrid r.status = status r.value = zcl.foundation.TypeValue() r.value.value = value return r def _get_model_info(ep, attributes={}): clus = ep.add_input_cluster(0) assert 0 in ep.in_clusters assert ep.in_clusters[0] is clus async def mockrequest( foundation, command, schema, args, manufacturer=None, **kwargs ): assert foundation is True assert command == 0 result = [] for attr_id, value in zip(args, attributes[tuple(args)]): if isinstance(value, BaseException): raise value elif value is None: rar = _mk_rar(attr_id, None, status=1) else: raw_attr_value = t.uint8_t(len(value)).serialize() + value rar = _mk_rar(attr_id, t.CharacterString.deserialize(raw_attr_value)[0]) result.append(rar) return [result] clus.request = mockrequest return ep.get_model_info() async def test_get_model_info(ep): mod, man = await _get_model_info( ep, attributes={ (0x0004, 0x0005): (b"Mock Manufacturer", b"Mock Model"), }, ) assert man == "Mock Manufacturer" assert mod == "Mock Model" async def test_init_endpoint_info_none(ep): mod, man = await _get_model_info( ep, attributes={ (0x0004, 0x0005): (None, None), (0x0004,): (None,), (0x0005,): (None,), }, ) assert man is None assert mod is None async def test_get_model_info_missing_basic_cluster(ep): assert zcl.clusters.general.Basic.cluster_id not in ep.in_clusters model, manuf = await ep.get_model_info() assert model is None assert manuf is None async def test_init_endpoint_info_null_padded_manuf(ep): mod, man = await _get_model_info( ep, attributes={ (0x0004, 0x0005): ( b"Mock Manufacturer\x00\x04\\\x00\\\x00\x00\x00\x00\x00\x07", b"Mock Model", ), }, ) assert man == "Mock Manufacturer" assert mod == "Mock Model" async def test_init_endpoint_info_null_padded_model(ep): mod, man = await _get_model_info( ep, attributes={ (0x0004, 0x0005): ( b"Mock Manufacturer", b"Mock Model\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00", ), }, ) assert man == "Mock Manufacturer" assert mod == "Mock Model" async def test_init_endpoint_info_null_padded_manuf_model(ep): mod, man = await _get_model_info( ep, attributes={ (0x0004, 0x0005): ( b"Mock Manufacturer\x00\x04\\\x00\\\x00\x00\x00\x00\x00\x07", b"Mock Model\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00", ), }, ) assert man == "Mock Manufacturer" assert mod == "Mock Model" async def test_get_model_info_delivery_error(ep): with pytest.raises(zigpy.exceptions.ZigbeeException): await _get_model_info( ep, attributes={ (0x0004, 0x0005): ( zigpy.exceptions.ZigbeeException(), zigpy.exceptions.ZigbeeException(), ) }, ) async def test_get_model_info_timeout(ep): with pytest.raises(asyncio.TimeoutError): await _get_model_info( ep, attributes={ (0x0004, 0x0005): (asyncio.TimeoutError(), asyncio.TimeoutError()), (0x0004,): (asyncio.TimeoutError(),), (0x0005,): (asyncio.TimeoutError(),), }, ) async def test_get_model_info_double_read_timeout(ep): mod, man = await _get_model_info( ep, attributes={ # The double read fails (0x0004, 0x0005): (asyncio.TimeoutError(), asyncio.TimeoutError()), # But individually the attributes can be read (0x0004,): (b"Mock Manufacturer",), (0x0005,): (b"Mock Model",), }, ) assert man == "Mock Manufacturer" assert mod == "Mock Model" def _group_add_mock(ep, status=ZCLStatus.SUCCESS, no_groups_cluster=False): async def mock_req(*args, **kwargs): return [status, sentinel.group_id] if not no_groups_cluster: ep.add_input_cluster(4) ep.request = MagicMock(side_effect=mock_req) ep.device.application.groups = MagicMock(spec_set=group.Groups) return ep @pytest.mark.parametrize("status", [ZCLStatus.SUCCESS, ZCLStatus.DUPLICATE_EXISTS]) async def test_add_to_group(ep, status): ep = _group_add_mock(ep, status=status) grp_id, grp_name = 0x1234, "Group 0x1234**" res = await ep.add_to_group(grp_id, grp_name) assert res == status assert ep.request.call_count == 1 groups = ep.device.application.groups assert groups.add_group.call_count == 1 assert groups.remove_group.call_count == 0 assert groups.add_group.call_args[0][0] == grp_id assert groups.add_group.call_args[0][1] == grp_name async def test_add_to_group_no_groups(ep): ep = _group_add_mock(ep, no_groups_cluster=True) grp_id, grp_name = 0x1234, "Group 0x1234**" res = await ep.add_to_group(grp_id, grp_name) assert res != ZCLStatus.SUCCESS assert ep.request.call_count == 0 groups = ep.device.application.groups assert groups.add_group.call_count == 0 assert groups.remove_group.call_count == 0 @pytest.mark.parametrize( "status", (s for s in ZCLStatus if s not in (ZCLStatus.SUCCESS, ZCLStatus.DUPLICATE_EXISTS)), ) async def test_add_to_group_fail(ep, status): ep = _group_add_mock(ep, status=status) grp_id, grp_name = 0x1234, "Group 0x1234**" res = await ep.add_to_group(grp_id, grp_name) assert res != ZCLStatus.SUCCESS assert ep.request.call_count == 1 groups = ep.device.application.groups assert groups.add_group.call_count == 0 assert groups.remove_group.call_count == 0 def _group_remove_mock(ep, success=True, no_groups_cluster=False, not_member=False): async def mock_req(*args, **kwargs): if success: return [ZCLStatus.SUCCESS, sentinel.group_id] return [ZCLStatus.DUPLICATE_EXISTS, sentinel.group_id] if not no_groups_cluster: ep.add_input_cluster(4) ep.request = MagicMock(side_effect=mock_req) ep.device.application.groups = MagicMock(spec_set=group.Groups) grp = MagicMock(spec_set=group.Group) ep.device.application.groups.__contains__.return_value = not not_member ep.device.application.groups.__getitem__.return_value = grp return ep, grp async def test_remove_from_group(ep): grp_id = 0x1234 ep, grp_mock = _group_remove_mock(ep) res = await ep.remove_from_group(grp_id) assert res == ZCLStatus.SUCCESS assert ep.request.call_count == 1 groups = ep.device.application.groups assert groups.add_group.call_count == 0 assert groups.remove_group.call_count == 0 assert groups.__getitem__.call_args[0][0] == grp_id assert grp_mock.add_member.call_count == 0 assert grp_mock.remove_member.call_count == 1 assert grp_mock.remove_member.call_args[0][0] == ep async def test_remove_from_group_no_groups_cluster(ep): grp_id = 0x1234 ep, grp_mock = _group_remove_mock(ep, no_groups_cluster=True) res = await ep.remove_from_group(grp_id) assert res != ZCLStatus.SUCCESS assert ep.request.call_count == 0 groups = ep.device.application.groups assert groups.add_group.call_count == 0 assert groups.remove_group.call_count == 0 assert grp_mock.add_member.call_count == 0 assert grp_mock.remove_member.call_count == 0 async def test_remove_from_group_fail(ep): grp_id = 0x1234 ep, grp_mock = _group_remove_mock(ep, success=False) res = await ep.remove_from_group(grp_id) assert res != ZCLStatus.SUCCESS assert ep.request.call_count == 1 groups = ep.device.application.groups assert groups.add_group.call_count == 0 assert groups.remove_group.call_count == 0 assert grp_mock.add_member.call_count == 0 assert grp_mock.remove_member.call_count == 0 def test_ep_manufacturer(ep): ep.device.manufacturer = sentinel.device_manufacturer assert ep.manufacturer is sentinel.device_manufacturer ep.manufacturer = sentinel.ep_manufacturer assert ep.manufacturer is sentinel.ep_manufacturer def test_ep_model(ep): ep.device.model = sentinel.device_model assert ep.model is sentinel.device_model ep.model = sentinel.ep_model assert ep.model is sentinel.ep_model async def test_group_membership_scan(ep): """Test group membership scan.""" ep.device.application.groups.update_group_membership = MagicMock() await ep.group_membership_scan() assert ep.device.application.groups.update_group_membership.call_count == 0 assert ep.device.request.call_count == 0 ep.add_input_cluster(4) ep.device.request.return_value = [0, [1, 3, 7]] await ep.group_membership_scan() assert ep.device.application.groups.update_group_membership.call_count == 1 assert ep.device.application.groups.update_group_membership.call_args[0][1] == { 1, 3, 7, } assert ep.device.request.call_count == 1 async def test_group_membership_scan_fail(ep): """Test group membership scan failure.""" ep.device.application.groups.update_group_membership = MagicMock() ep.add_input_cluster(4) ep.device.request.side_effect = asyncio.TimeoutError await ep.group_membership_scan() assert ep.device.application.groups.update_group_membership.call_count == 0 assert ep.device.request.call_count == 1 async def test_group_membership_scan_fail_default_response(ep, caplog): """Test group membership scan failure because group commands are unsupported.""" ep.device.application.groups.update_group_membership = MagicMock() ep.add_input_cluster(4) ep.device.request.side_effect = asyncio.TimeoutError with patch.object(ep.groups, "get_membership", new=AsyncMock()) as get_membership: get_membership.return_value = GENERAL_COMMANDS[ GeneralCommand.Default_Response ].schema(command_id=2, status=ZCLStatus.UNSUP_CLUSTER_COMMAND) await ep.group_membership_scan() assert "Device does not support group commands" in caplog.text assert ep.device.application.groups.update_group_membership.call_count == 0 def test_endpoint_manufacturer_id(ep): """Test manufacturer id.""" ep.device.manufacturer_id = sentinel.manufacturer_id assert ep.manufacturer_id is sentinel.manufacturer_id def test_endpoint_repr(ep): ep.status = endpoint.Status.ZDO_INIT # All standard ep.add_input_cluster(0x0001) ep.add_input_cluster(0x0002) ep.add_output_cluster(0x0006) ep.add_output_cluster(0x0008) # Spec-violating but still happens (https://github.com/zigpy/zigpy/issues/758) ep.add_input_cluster(0xEF00) assert "ZDO_INIT" in repr(ep) assert "power:0x0001" in repr(ep) assert "device_temperature:0x0002" in repr(ep) assert "on_off:0x0006" in repr(ep) assert "level:0x0008" in repr(ep) assert "0xEF00" in repr(ep)