| # Copyright 2021-2022 Google LLC | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"); | |
| # you may not use this file except in compliance with the License. | |
| # You may obtain a copy of the License at | |
| # | |
| # https://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| # ----------------------------------------------------------------------------- | |
| # Imports | |
| # ----------------------------------------------------------------------------- | |
| import inspect | |
| import struct | |
| import pytest | |
| from bumble import hci, utils | |
| # ----------------------------------------------------------------------------- | |
| # pylint: disable=invalid-name | |
| def basic_check(x): | |
| packet = bytes(x) | |
| print(packet.hex()) | |
| parsed = hci.HCI_Packet.from_bytes(packet) | |
| x_str = str(x) | |
| parsed_str = str(parsed) | |
| print(x_str) | |
| parsed_bytes = bytes(parsed) | |
| assert x_str == parsed_str | |
| assert packet == parsed_bytes | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_Event(): | |
| event = hci.HCI_Event(event_code=0xF9) | |
| basic_check(event) | |
| event = hci.HCI_Event(event_code=0xF8, parameters=bytes.fromhex('AABBCC')) | |
| basic_check(event) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Connection_Complete_Event(): | |
| address = hci.Address('00:11:22:33:44:55') | |
| event = hci.HCI_LE_Connection_Complete_Event( | |
| status=hci.HCI_SUCCESS, | |
| connection_handle=1, | |
| role=1, | |
| peer_address_type=1, | |
| peer_address=address, | |
| connection_interval=3, | |
| peripheral_latency=4, | |
| supervision_timeout=5, | |
| central_clock_accuracy=6, | |
| ) | |
| basic_check(event) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Advertising_Report_Event(): | |
| address = hci.Address('00:11:22:33:44:55/P') | |
| report = hci.HCI_LE_Advertising_Report_Event.Report( | |
| event_type=hci.HCI_LE_Advertising_Report_Event.EventType.ADV_IND, | |
| address_type=hci.Address.PUBLIC_DEVICE_ADDRESS, | |
| address=address, | |
| data=bytes.fromhex( | |
| '0201061106ba5689a6fabfa2bd01467d6e00fbabad08160a181604659b03' | |
| ), | |
| rssi=100, | |
| ) | |
| event = hci.HCI_LE_Advertising_Report_Event([report]) | |
| basic_check(event) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Extended_Advertising_Report_Event(): | |
| address = hci.Address('00:11:22:33:44:55/P') | |
| report = hci.HCI_LE_Extended_Advertising_Report_Event.Report( | |
| event_type=hci.HCI_LE_Extended_Advertising_Report_Event.EventType.CONNECTABLE_ADVERTISING, | |
| address_type=hci.Address.PUBLIC_DEVICE_ADDRESS, | |
| address=address, | |
| data=bytes.fromhex( | |
| '0201061106ba5689a6fabfa2bd01467d6e00fbabad08160a181604659b03' | |
| ), | |
| rssi=100, | |
| primary_phy=hci.HCI_LE_1M_PHY, | |
| secondary_phy=hci.HCI_LE_CODED_PHY, | |
| advertising_sid=0, | |
| tx_power=10, | |
| periodic_advertising_interval=2, | |
| direct_address=hci.Address('00:11:22:33:44:55/P'), | |
| direct_address_type=hci.Address.PUBLIC_DEVICE_ADDRESS, | |
| ) | |
| event = hci.HCI_LE_Extended_Advertising_Report_Event([report]) | |
| basic_check(event) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Read_Remote_Features_Complete_Event(): | |
| event = hci.HCI_LE_Read_Remote_Features_Complete_Event( | |
| status=hci.HCI_SUCCESS, | |
| connection_handle=0x007, | |
| le_features=bytes.fromhex('0011223344556677'), | |
| ) | |
| basic_check(event) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Connection_Update_Complete_Event(): | |
| event = hci.HCI_LE_Connection_Update_Complete_Event( | |
| status=hci.HCI_SUCCESS, | |
| connection_handle=0x007, | |
| connection_interval=10, | |
| peripheral_latency=3, | |
| supervision_timeout=5, | |
| ) | |
| basic_check(event) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Channel_Selection_Algorithm_Event(): | |
| event = hci.HCI_LE_Channel_Selection_Algorithm_Event( | |
| connection_handle=7, channel_selection_algorithm=1 | |
| ) | |
| basic_check(event) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_Command_Complete_Event(): | |
| # With a serializable object | |
| event1 = hci.HCI_Command_Complete_Event( | |
| num_hci_command_packets=34, | |
| command_opcode=hci.HCI_LE_READ_BUFFER_SIZE_COMMAND, | |
| return_parameters=hci.HCI_LE_Read_Buffer_Size_Command.return_parameters_class( | |
| status=0, | |
| le_acl_data_packet_length=1234, | |
| total_num_le_acl_data_packets=56, | |
| ), | |
| ) | |
| basic_check(event1) | |
| # With a simple status as an integer status | |
| event3 = hci.HCI_Command_Complete_Event( | |
| num_hci_command_packets=1, | |
| command_opcode=hci.HCI_RESET_COMMAND, | |
| return_parameters=hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode(9)), | |
| ) | |
| basic_check(event3) | |
| assert event3.return_parameters.status == 9 | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_Command_Status_Event(): | |
| event = hci.HCI_Command_Status_Event( | |
| status=0, num_hci_command_packets=37, command_opcode=hci.HCI_DISCONNECT_COMMAND | |
| ) | |
| basic_check(event) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_Number_Of_Completed_Packets_Event(): | |
| event = hci.HCI_Number_Of_Completed_Packets_Event( | |
| connection_handles=(1, 2), | |
| num_completed_packets=(3, 4), | |
| ) | |
| basic_check(event) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_Vendor_Event(): | |
| data = bytes.fromhex('01020304') | |
| event = hci.HCI_Vendor_Event(data=data) | |
| event_bytes = bytes(event) | |
| parsed = hci.HCI_Packet.from_bytes(event_bytes) | |
| assert isinstance(parsed, hci.HCI_Vendor_Event) | |
| assert parsed.data == data | |
| class HCI_Custom_Event(hci.HCI_Event): | |
| def __init__(self, blabla): | |
| super().__init__( | |
| event_code=hci.HCI_VENDOR_EVENT, parameters=struct.pack("<I", blabla) | |
| ) | |
| self.name = 'HCI_CUSTOM_EVENT' | |
| self.blabla = blabla | |
| def create_event(payload): | |
| if payload[0] == 1: | |
| return HCI_Custom_Event(blabla=struct.unpack('<I', payload)[0]) | |
| return None | |
| hci.HCI_Event.add_vendor_factory(create_event) | |
| parsed = hci.HCI_Packet.from_bytes(event_bytes) | |
| assert isinstance(parsed, HCI_Custom_Event) | |
| assert parsed.blabla == 0x04030201 | |
| event_bytes2 = event_bytes[:3] + bytes([7]) + event_bytes[4:] | |
| parsed = hci.HCI_Packet.from_bytes(event_bytes2) | |
| assert not isinstance(parsed, HCI_Custom_Event) | |
| assert isinstance(parsed, hci.HCI_Vendor_Event) | |
| hci.HCI_Event.remove_vendor_factory(create_event) | |
| parsed = hci.HCI_Packet.from_bytes(event_bytes) | |
| assert not isinstance(parsed, HCI_Custom_Event) | |
| assert isinstance(parsed, hci.HCI_Vendor_Event) | |
| # ----------------------------------------------------------------------------- | |
| def test_return_parameters() -> None: | |
| params = hci.HCI_Reset_Command.parse_return_parameters(bytes.fromhex('3C')) | |
| assert params.status == hci.HCI_ErrorCode.ADVERTISING_TIMEOUT_ERROR | |
| assert isinstance(params.status, utils.OpenIntEnum) | |
| params = hci.HCI_Read_BD_ADDR_Command.parse_return_parameters( | |
| bytes.fromhex('00001122334455') | |
| ) | |
| assert params.status == hci.HCI_ErrorCode.SUCCESS | |
| assert isinstance(params.status, utils.OpenIntEnum) | |
| assert isinstance(params.bd_addr, hci.Address) | |
| params = hci.HCI_Read_Local_Name_Command.parse_return_parameters( | |
| bytes.fromhex('0068656c6c6f') + bytes(248 - 5) | |
| ) | |
| assert params.status == hci.HCI_ErrorCode.SUCCESS | |
| assert isinstance(params.local_name, bytes) | |
| assert len(params.local_name) == 248 | |
| assert hci.map_null_terminated_utf8_string(params.local_name) == 'hello' | |
| # Some return parameters may be shorter than the full length | |
| # (for Command Complete events with errors) | |
| params = hci.HCI_Read_BD_ADDR_Command.parse_return_parameters( | |
| bytes.fromhex('010011223344') | |
| ) | |
| assert isinstance(params, hci.HCI_StatusReturnParameters) | |
| assert params.status == hci.HCI_ErrorCode.UNKNOWN_HCI_COMMAND_ERROR | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_Command(): | |
| command = hci.HCI_Command(op_code=0x5566) | |
| basic_check(command) | |
| command = hci.HCI_Command(op_code=0x5566, parameters=bytes.fromhex('AABBCC')) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_custom_command(): | |
| @hci.HCI_Command.command | |
| class CustomCommand(hci.HCI_Command): | |
| op_code = 0x7788 | |
| name = 'Custom Command' | |
| command = CustomCommand() | |
| basic_check(command) | |
| parsed = hci.HCI_Packet.from_bytes(bytes(command)) | |
| assert isinstance(parsed, CustomCommand) | |
| assert parsed.op_code == 0x7788 | |
| assert parsed.name == 'Custom Command' | |
| # ----------------------------------------------------------------------------- | |
| def test_custom_event(): | |
| @hci.HCI_Event.event | |
| class CustomEvent(hci.HCI_Event): | |
| event_code = 0x99 | |
| name = 'Custom Event' | |
| event = CustomEvent() | |
| basic_check(event) | |
| parsed = hci.HCI_Packet.from_bytes(bytes(event)) | |
| assert isinstance(parsed, CustomEvent) | |
| assert parsed.event_code == 0x99 | |
| assert parsed.name == 'Custom Event' | |
| # ----------------------------------------------------------------------------- | |
| def test_custom_le_meta_event(): | |
| @hci.HCI_LE_Meta_Event.event | |
| class CustomEvent(hci.HCI_LE_Meta_Event): | |
| subevent_code = 0xFF | |
| name = 'Custom Extended Event' | |
| event = CustomEvent() | |
| basic_check(event) | |
| parsed = hci.HCI_Packet.from_bytes(bytes(event)) | |
| assert isinstance(parsed, CustomEvent) | |
| assert parsed.subevent_code == 0xFF | |
| assert parsed.name == 'Custom Extended Event' | |
| # ----------------------------------------------------------------------------- | |
| @pytest.mark.parametrize( | |
| "clazz,", | |
| [ | |
| clazz[1] | |
| for clazz in inspect.getmembers(hci) | |
| if isinstance(clazz[1], type) | |
| and issubclass(clazz[1], hci.HCI_Command) | |
| and clazz[1] not in (hci.HCI_Command, hci.HCI_SyncCommand, hci.HCI_AsyncCommand) | |
| ], | |
| ) | |
| def test_hci_command_subclasses_op_code(clazz: type[hci.HCI_Command]): | |
| assert clazz.op_code > 0 | |
| assert isinstance(clazz.name, str) | |
| # ----------------------------------------------------------------------------- | |
| @pytest.mark.parametrize( | |
| "clazz,", | |
| [ | |
| clazz[1] | |
| for clazz in inspect.getmembers(hci) | |
| if isinstance(clazz[1], type) | |
| and clazz[1] is not hci.HCI_Event | |
| and issubclass(clazz[1], hci.HCI_Event) | |
| and not issubclass(clazz[1], hci.HCI_Extended_Event) | |
| ], | |
| ) | |
| def test_hci_event_subclasses_event_code(clazz: type[hci.HCI_Event]): | |
| assert clazz.event_code > 0 | |
| assert isinstance(clazz.name, str) | |
| # ----------------------------------------------------------------------------- | |
| @pytest.mark.parametrize( | |
| "clazz,", | |
| [ | |
| clazz[1] | |
| for clazz in inspect.getmembers(hci) | |
| if isinstance(clazz[1], type) | |
| and issubclass(clazz[1], hci.HCI_Extended_Event) | |
| and clazz[1] not in (hci.HCI_Extended_Event, hci.HCI_LE_Meta_Event) | |
| ], | |
| ) | |
| def test_hci_extended_event_subclasses_event_code(clazz: type[hci.HCI_Extended_Event]): | |
| assert clazz.event_code > 0 | |
| assert clazz.subevent_code > 0 | |
| assert isinstance(clazz.name, str) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_PIN_Code_Request_Reply_Command(): | |
| pin_code = b'1234' | |
| pin_code_length = len(pin_code) | |
| # here to make the test pass, we need to | |
| # pad pin_code, as hci.HCI_Object.format_fields | |
| # does not do it for us | |
| padded_pin_code = pin_code + bytes(16 - pin_code_length) | |
| command = hci.HCI_PIN_Code_Request_Reply_Command( | |
| bd_addr=hci.Address( | |
| '00:11:22:33:44:55', address_type=hci.Address.PUBLIC_DEVICE_ADDRESS | |
| ), | |
| pin_code_length=pin_code_length, | |
| pin_code=padded_pin_code, | |
| ) | |
| basic_check(command) | |
| def test_HCI_Reset_Command(): | |
| command = hci.HCI_Reset_Command() | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_Read_Local_Version_Information_Command(): | |
| command = hci.HCI_Read_Local_Version_Information_Command() | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_Read_Local_Supported_Commands_Command(): | |
| command = hci.HCI_Read_Local_Supported_Commands_Command() | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_Read_Local_Supported_Features_Command(): | |
| command = hci.HCI_Read_Local_Supported_Features_Command() | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_Disconnect_Command(): | |
| command = hci.HCI_Disconnect_Command(connection_handle=123, reason=0x11) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_Set_Event_Mask_Command(): | |
| command = hci.HCI_Set_Event_Mask_Command( | |
| event_mask=bytes.fromhex('0011223344556677') | |
| ) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Set_Event_Mask_Command(): | |
| command = hci.HCI_LE_Set_Event_Mask_Command( | |
| le_event_mask=hci.HCI_LE_Set_Event_Mask_Command.mask( | |
| [ | |
| hci.HCI_LE_CONNECTION_COMPLETE_EVENT, | |
| hci.HCI_LE_ENHANCED_CONNECTION_COMPLETE_V2_EVENT, | |
| ] | |
| ) | |
| ) | |
| assert command.le_event_mask == bytes.fromhex('0100000000010000') | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Set_Random_Address_Command(): | |
| command = hci.HCI_LE_Set_Random_Address_Command( | |
| random_address=hci.Address('00:11:22:33:44:55') | |
| ) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Set_Advertising_Parameters_Command(): | |
| command = hci.HCI_LE_Set_Advertising_Parameters_Command( | |
| advertising_interval_min=20, | |
| advertising_interval_max=30, | |
| advertising_type=hci.HCI_LE_Set_Advertising_Parameters_Command.AdvertisingType.ADV_NONCONN_IND, | |
| own_address_type=hci.Address.PUBLIC_DEVICE_ADDRESS, | |
| peer_address_type=hci.Address.RANDOM_DEVICE_ADDRESS, | |
| peer_address=hci.Address('00:11:22:33:44:55'), | |
| advertising_channel_map=0x03, | |
| advertising_filter_policy=1, | |
| ) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Set_Advertising_Data_Command(): | |
| command = hci.HCI_LE_Set_Advertising_Data_Command( | |
| advertising_data=bytes.fromhex('AABBCC') | |
| ) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Set_Scan_Parameters_Command(): | |
| command = hci.HCI_LE_Set_Scan_Parameters_Command( | |
| le_scan_type=1, | |
| le_scan_interval=20, | |
| le_scan_window=10, | |
| own_address_type=1, | |
| scanning_filter_policy=0, | |
| ) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Set_Scan_Enable_Command(): | |
| command = hci.HCI_LE_Set_Scan_Enable_Command(le_scan_enable=1, filter_duplicates=0) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Create_Connection_Command(): | |
| command = hci.HCI_LE_Create_Connection_Command( | |
| le_scan_interval=4, | |
| le_scan_window=5, | |
| initiator_filter_policy=1, | |
| peer_address_type=1, | |
| peer_address=hci.Address('00:11:22:33:44:55'), | |
| own_address_type=2, | |
| connection_interval_min=7, | |
| connection_interval_max=8, | |
| max_latency=9, | |
| supervision_timeout=10, | |
| min_ce_length=11, | |
| max_ce_length=12, | |
| ) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Extended_Create_Connection_Command(): | |
| command = hci.HCI_LE_Extended_Create_Connection_Command( | |
| initiator_filter_policy=0, | |
| own_address_type=0, | |
| peer_address_type=1, | |
| peer_address=hci.Address('00:11:22:33:44:55'), | |
| initiating_phys=3, | |
| scan_intervals=(10, 11), | |
| scan_windows=(12, 13), | |
| connection_interval_mins=(14, 15), | |
| connection_interval_maxs=(16, 17), | |
| max_latencies=(18, 19), | |
| supervision_timeouts=(20, 21), | |
| min_ce_lengths=(100, 101), | |
| max_ce_lengths=(102, 103), | |
| ) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Add_Device_To_Filter_Accept_List_Command(): | |
| command = hci.HCI_LE_Add_Device_To_Filter_Accept_List_Command( | |
| address_type=1, address=hci.Address('00:11:22:33:44:55') | |
| ) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Remove_Device_From_Filter_Accept_List_Command(): | |
| command = hci.HCI_LE_Remove_Device_From_Filter_Accept_List_Command( | |
| address_type=1, address=hci.Address('00:11:22:33:44:55') | |
| ) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Connection_Update_Command(): | |
| command = hci.HCI_LE_Connection_Update_Command( | |
| connection_handle=0x0002, | |
| connection_interval_min=10, | |
| connection_interval_max=20, | |
| max_latency=7, | |
| supervision_timeout=3, | |
| min_ce_length=100, | |
| max_ce_length=200, | |
| ) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Read_Remote_Features_Command(): | |
| command = hci.HCI_LE_Read_Remote_Features_Command(connection_handle=0x0002) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Set_Default_PHY_Command(): | |
| command = hci.HCI_LE_Set_Default_PHY_Command(all_phys=0, tx_phys=1, rx_phys=1) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Set_Extended_Scan_Parameters_Command(): | |
| command = hci.HCI_LE_Set_Extended_Scan_Parameters_Command( | |
| own_address_type=hci.Address.RANDOM_DEVICE_ADDRESS, | |
| # pylint: disable-next=line-too-long | |
| scanning_filter_policy=hci.HCI_LE_Set_Extended_Scan_Parameters_Command.BASIC_FILTERED_POLICY, | |
| scanning_phys=( | |
| 1 << hci.HCI_LE_1M_PHY_BIT | 1 << hci.HCI_LE_CODED_PHY_BIT | 1 << 4 | |
| ), | |
| scan_types=[ | |
| hci.HCI_LE_Set_Extended_Scan_Parameters_Command.ACTIVE_SCANNING, | |
| hci.HCI_LE_Set_Extended_Scan_Parameters_Command.ACTIVE_SCANNING, | |
| hci.HCI_LE_Set_Extended_Scan_Parameters_Command.PASSIVE_SCANNING, | |
| ], | |
| scan_intervals=[1, 2, 3], | |
| scan_windows=[4, 5, 6], | |
| ) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Set_Extended_Advertising_Enable_Command(): | |
| command = hci.HCI_Packet.from_bytes( | |
| bytes.fromhex('0139200e010301050008020600090307000a') | |
| ) | |
| assert command.enable == 1 | |
| assert command.advertising_handles == [1, 2, 3] | |
| assert command.durations == [5, 6, 7] | |
| assert command.max_extended_advertising_events == [8, 9, 10] | |
| command = hci.HCI_LE_Set_Extended_Advertising_Enable_Command( | |
| enable=1, | |
| advertising_handles=[1, 2, 3], | |
| durations=[5, 6, 7], | |
| max_extended_advertising_events=[8, 9, 10], | |
| ) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_LE_Setup_ISO_Data_Path_Command(): | |
| command = hci.HCI_Packet.from_bytes( | |
| bytes.fromhex('016e200d60000001030000000000000000') | |
| ) | |
| assert command.connection_handle == 0x0060 | |
| assert command.data_path_direction == 0x00 | |
| assert command.data_path_id == 0x01 | |
| assert command.codec_id == hci.CodingFormat(hci.CodecID.TRANSPARENT) | |
| assert command.controller_delay == 0 | |
| assert command.codec_configuration == b'' | |
| command = hci.HCI_LE_Setup_ISO_Data_Path_Command( | |
| connection_handle=0x0060, | |
| data_path_direction=0x00, | |
| data_path_id=0x01, | |
| codec_id=hci.CodingFormat(hci.CodecID.TRANSPARENT), | |
| controller_delay=0x00, | |
| codec_configuration=b'', | |
| ) | |
| basic_check(command) | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_Read_Local_Supported_Codecs_Command_Complete(): | |
| returned_parameters = ( | |
| hci.HCI_Read_Local_Supported_Codecs_Command.parse_return_parameters( | |
| bytes( | |
| [ | |
| hci.HCI_SUCCESS, | |
| 3, | |
| hci.CodecID.A_LOG, | |
| hci.CodecID.CVSD, | |
| hci.CodecID.LINEAR_PCM, | |
| 0, | |
| ] | |
| ) | |
| ) | |
| ) | |
| assert returned_parameters.standard_codec_ids == [ | |
| hci.CodecID.A_LOG, | |
| hci.CodecID.CVSD, | |
| hci.CodecID.LINEAR_PCM, | |
| ] | |
| # ----------------------------------------------------------------------------- | |
| def test_HCI_Read_Local_Supported_Codecs_V2_Command_Complete(): | |
| returned_parameters = hci.HCI_Read_Local_Supported_Codecs_V2_Command.parse_return_parameters( | |
| bytes( | |
| [ | |
| hci.HCI_SUCCESS, | |
| 3, | |
| hci.CodecID.A_LOG, | |
| hci.HCI_Read_Local_Supported_Codecs_V2_ReturnParameters.Transport.BR_EDR_ACL, | |
| hci.CodecID.CVSD, | |
| hci.HCI_Read_Local_Supported_Codecs_V2_ReturnParameters.Transport.BR_EDR_SCO, | |
| hci.CodecID.LINEAR_PCM, | |
| hci.HCI_Read_Local_Supported_Codecs_V2_ReturnParameters.Transport.LE_CIS, | |
| 0, | |
| ] | |
| ) | |
| ) | |
| assert returned_parameters.standard_codec_ids == [ | |
| hci.CodecID.A_LOG, | |
| hci.CodecID.CVSD, | |
| hci.CodecID.LINEAR_PCM, | |
| ] | |
| assert returned_parameters.standard_codec_transports == [ | |
| hci.HCI_Read_Local_Supported_Codecs_V2_ReturnParameters.Transport.BR_EDR_ACL, | |
| hci.HCI_Read_Local_Supported_Codecs_V2_ReturnParameters.Transport.BR_EDR_SCO, | |
| hci.HCI_Read_Local_Supported_Codecs_V2_ReturnParameters.Transport.LE_CIS, | |
| ] | |
| # ----------------------------------------------------------------------------- | |
| def test_address(): | |
| a = hci.Address('C4:F2:17:1A:1D:BB') | |
| assert not a.is_public | |
| assert a.is_random | |
| assert a.address_type == hci.Address.RANDOM_DEVICE_ADDRESS | |
| assert not a.is_resolvable | |
| assert not a.is_resolved | |
| assert a.is_static | |
| # ----------------------------------------------------------------------------- | |
| def test_custom(): | |
| data = bytes([0x77, 0x02, 0x01, 0x03]) | |
| packet = hci.HCI_CustomPacket(data) | |
| assert packet.hci_packet_type == 0x77 | |
| assert packet.payload == data | |
| # ----------------------------------------------------------------------------- | |
| def test_iso_data_packet(): | |
| data = bytes.fromhex( | |
| '05616044002ac9f0a193003c00e83b477b00eba8d41dc018bf1a980f0290afe1e7c37652096697' | |
| '52b6a535a8df61e22931ef5a36281bc77ed6a3206d984bcdabee6be831c699cb50e2' | |
| ) | |
| packet = hci.HCI_IsoDataPacket.from_bytes(data) | |
| assert packet.connection_handle == 0x0061 | |
| assert packet.packet_status_flag == 0 | |
| assert packet.pb_flag == 0x02 | |
| assert packet.ts_flag == 0x01 | |
| assert packet.data_total_length == 68 | |
| assert packet.time_stamp == 2716911914 | |
| assert packet.packet_sequence_number == 147 | |
| assert packet.iso_sdu_length == 60 | |
| assert packet.iso_sdu_fragment == bytes.fromhex( | |
| 'e83b477b00eba8d41dc018bf1a980f0290afe1e7c3765209669752b6a535a8df61e22931ef5a3' | |
| '6281bc77ed6a3206d984bcdabee6be831c699cb50e2' | |
| ) | |
| assert bytes(packet) == data | |
| # ----------------------------------------------------------------------------- | |
| def run_test_events(): | |
| test_HCI_Event() | |
| test_HCI_LE_Connection_Complete_Event() | |
| test_HCI_LE_Advertising_Report_Event() | |
| test_HCI_LE_Connection_Update_Complete_Event() | |
| test_HCI_LE_Read_Remote_Features_Complete_Event() | |
| test_HCI_LE_Channel_Selection_Algorithm_Event() | |
| test_HCI_Command_Complete_Event() | |
| test_HCI_Command_Status_Event() | |
| test_HCI_Number_Of_Completed_Packets_Event() | |
| test_HCI_Vendor_Event() | |
| # ----------------------------------------------------------------------------- | |
| def run_test_commands(): | |
| test_HCI_Command() | |
| test_HCI_Reset_Command() | |
| test_HCI_PIN_Code_Request_Reply_Command() | |
| test_HCI_Read_Local_Version_Information_Command() | |
| test_HCI_Read_Local_Supported_Commands_Command() | |
| test_HCI_Read_Local_Supported_Features_Command() | |
| test_HCI_Disconnect_Command() | |
| test_HCI_Set_Event_Mask_Command() | |
| test_HCI_LE_Set_Event_Mask_Command() | |
| test_HCI_LE_Set_Random_Address_Command() | |
| test_HCI_LE_Set_Advertising_Parameters_Command() | |
| test_HCI_LE_Set_Advertising_Data_Command() | |
| test_HCI_LE_Set_Scan_Parameters_Command() | |
| test_HCI_LE_Set_Scan_Enable_Command() | |
| test_HCI_LE_Create_Connection_Command() | |
| test_HCI_LE_Extended_Create_Connection_Command() | |
| test_HCI_LE_Add_Device_To_Filter_Accept_List_Command() | |
| test_HCI_LE_Remove_Device_From_Filter_Accept_List_Command() | |
| test_HCI_LE_Connection_Update_Command() | |
| test_HCI_LE_Read_Remote_Features_Command() | |
| test_HCI_LE_Set_Default_PHY_Command() | |
| test_HCI_LE_Set_Extended_Scan_Parameters_Command() | |
| test_HCI_LE_Set_Extended_Advertising_Enable_Command() | |
| test_HCI_LE_Setup_ISO_Data_Path_Command() | |
| # ----------------------------------------------------------------------------- | |
| if __name__ == '__main__': | |
| run_test_events() | |
| run_test_commands() | |
| test_return_parameters() | |
| test_address() | |
| test_custom() | |
| test_iso_data_packet() |