| #!/usr/bin/env python3 |
| # |
| # Copyright 2019 - The Android Open Source Project |
| # |
| # 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 |
| # |
| # http://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. |
| |
| from datetime import timedelta |
| import os |
| import sys |
| import logging |
| |
| from cert.gd_base_test_facade_only import GdFacadeOnlyBaseTestClass |
| from cert.event_callback_stream import EventCallbackStream |
| from cert.event_asserts import EventAsserts |
| from google.protobuf import empty_pb2 as empty_proto |
| from facade import rootservice_pb2 as facade_rootservice |
| from hal import facade_pb2 as hal_facade |
| from hci.facade import facade_pb2 as hci_facade |
| from bluetooth_packets_python3 import hci_packets |
| import bluetooth_packets_python3 as bt_packets |
| |
| |
| class DirectHciTest(GdFacadeOnlyBaseTestClass): |
| |
| def setup_test(self): |
| self.device_under_test.rootservice.StartStack( |
| facade_rootservice.StartStackRequest( |
| module_under_test=facade_rootservice.BluetoothModule.Value( |
| 'HCI'),)) |
| self.cert_device.rootservice.StartStack( |
| facade_rootservice.StartStackRequest( |
| module_under_test=facade_rootservice.BluetoothModule.Value( |
| 'HAL'),)) |
| |
| self.device_under_test.wait_channel_ready() |
| self.cert_device.wait_channel_ready() |
| |
| self.cert_device.hal.SendHciCommand( |
| hal_facade.HciCommandPacket( |
| payload=bytes(hci_packets.ResetBuilder().Serialize()))) |
| |
| def teardown_test(self): |
| self.device_under_test.rootservice.StopStack( |
| facade_rootservice.StopStackRequest()) |
| self.cert_device.rootservice.StopStack( |
| facade_rootservice.StopStackRequest()) |
| |
| def register_for_event(self, event_code): |
| msg = hci_facade.EventCodeMsg(code=int(event_code)) |
| self.device_under_test.hci.RegisterEventHandler(msg) |
| |
| def register_for_le_event(self, event_code): |
| msg = hci_facade.LeSubeventCodeMsg(code=int(event_code)) |
| self.device_under_test.hci.RegisterLeEventHandler(msg) |
| |
| def enqueue_hci_command(self, command, expect_complete): |
| cmd_bytes = bytes(command.Serialize()) |
| cmd = hci_facade.CommandMsg(command=cmd_bytes) |
| if (expect_complete): |
| self.device_under_test.hci.EnqueueCommandWithComplete(cmd) |
| else: |
| self.device_under_test.hci.EnqueueCommandWithStatus(cmd) |
| |
| def send_hal_hci_command(self, command): |
| self.cert_device.hal.SendHciCommand( |
| hal_facade.HciCommandPacket(payload=bytes(command.Serialize()))) |
| |
| def enqueue_acl_data(self, handle, pb_flag, b_flag, acl): |
| acl_msg = hci_facade.AclMsg( |
| handle=int(handle), |
| packet_boundary_flag=int(pb_flag), |
| broadcast_flag=int(b_flag), |
| data=acl) |
| self.device_under_test.hci.SendAclData(acl_msg) |
| |
| def send_hal_acl_data(self, handle, pb_flag, b_flag, acl): |
| lower = handle & 0xff |
| upper = (handle >> 8) & 0xf |
| upper = upper | int(pb_flag) & 0x3 |
| upper = upper | ((int(b_flag) & 0x3) << 2) |
| lower_length = len(acl) & 0xff |
| upper_length = (len(acl) & 0xff00) >> 8 |
| concatenated = bytes([lower, upper, lower_length, upper_length] + |
| list(acl)) |
| self.cert_device.hal.SendHciAcl( |
| hal_facade.HciAclPacket(payload=concatenated)) |
| |
| def test_local_hci_cmd_and_event(self): |
| # Loopback mode responds with ACL and SCO connection complete |
| self.register_for_event(hci_packets.EventCode.CONNECTION_COMPLETE) |
| self.register_for_event(hci_packets.EventCode.LOOPBACK_COMMAND) |
| self.register_for_event( |
| hci_packets.EventCode.CONNECTION_PACKET_TYPE_CHANGED) |
| with EventCallbackStream( |
| self.device_under_test.hci.FetchEvents( |
| empty_proto.Empty())) as hci_event_stream: |
| hci_event_asserts = EventAsserts(hci_event_stream) |
| |
| self.enqueue_hci_command( |
| hci_packets.WriteLoopbackModeBuilder( |
| hci_packets.LoopbackMode.ENABLE_LOCAL), True) |
| |
| cmd2loop = hci_packets.ReadLocalNameBuilder() |
| self.enqueue_hci_command(cmd2loop, True) |
| |
| looped_bytes = bytes(cmd2loop.Serialize()) |
| hci_event_asserts.assert_event_occurs( |
| lambda packet: looped_bytes in packet.event) |
| |
| def test_inquiry_from_dut(self): |
| self.register_for_event(hci_packets.EventCode.INQUIRY_RESULT) |
| with EventCallbackStream( |
| self.device_under_test.hci.FetchEvents( |
| empty_proto.Empty())) as hci_event_stream: |
| |
| hci_event_asserts = EventAsserts(hci_event_stream) |
| |
| self.send_hal_hci_command( |
| hci_packets.WriteScanEnableBuilder( |
| hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN)) |
| lap = hci_packets.Lap() |
| lap.lap = 0x33 |
| self.enqueue_hci_command( |
| hci_packets.InquiryBuilder(lap, 0x30, 0xff), False) |
| hci_event_asserts.assert_event_occurs( |
| lambda packet: b'\x02\x0f' in packet.event |
| # Expecting an HCI Event (code 0x02, length 0x0f) |
| ) |
| |
| def test_le_ad_scan_cert_advertises(self): |
| self.register_for_le_event( |
| hci_packets.SubeventCode.EXTENDED_ADVERTISING_REPORT) |
| self.register_for_le_event(hci_packets.SubeventCode.ADVERTISING_REPORT) |
| with EventCallbackStream( |
| self.device_under_test.hci.FetchLeSubevents( |
| empty_proto.Empty())) as hci_le_event_stream: |
| |
| hci_event_asserts = EventAsserts(hci_le_event_stream) |
| |
| # DUT Scans |
| self.enqueue_hci_command( |
| hci_packets.LeSetRandomAddressBuilder('0D:05:04:03:02:01'), |
| True) |
| phy_scan_params = hci_packets.PhyScanParameters() |
| phy_scan_params.le_scan_interval = 6553 |
| phy_scan_params.le_scan_window = 6553 |
| phy_scan_params.le_scan_type = hci_packets.LeScanType.ACTIVE |
| |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedScanParametersBuilder( |
| hci_packets.AddressType.RANDOM_DEVICE_ADDRESS, |
| hci_packets.LeSetScanningFilterPolicy.ACCEPT_ALL, 1, |
| [phy_scan_params]), True) |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedScanEnableBuilder( |
| hci_packets.Enable.ENABLED, |
| hci_packets.FilterDuplicates.DISABLED, 0, 0), True) |
| |
| # CERT Advertises |
| advertising_handle = 0 |
| self.send_hal_hci_command( |
| hci_packets.LeSetExtendedAdvertisingLegacyParametersBuilder( |
| advertising_handle, |
| hci_packets.LegacyAdvertisingProperties.ADV_IND, |
| 512, |
| 768, |
| 7, |
| hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, |
| hci_packets.PeerAddressType. |
| PUBLIC_DEVICE_OR_IDENTITY_ADDRESS, |
| 'A6:A5:A4:A3:A2:A1', |
| hci_packets.AdvertisingFilterPolicy.ALL_DEVICES, |
| 0xF7, |
| 1, # SID |
| hci_packets.Enable.DISABLED # Scan request notification |
| )) |
| |
| self.send_hal_hci_command( |
| hci_packets.LeSetExtendedAdvertisingRandomAddressBuilder( |
| advertising_handle, '0C:05:04:03:02:01')) |
| gap_name = hci_packets.GapData() |
| gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME |
| gap_name.data = list(bytes(b'Im_A_Cert!')) # TODO: Fix and remove ! |
| |
| self.send_hal_hci_command( |
| hci_packets.LeSetExtendedAdvertisingDataBuilder( |
| advertising_handle, |
| hci_packets.Operation.COMPLETE_ADVERTISEMENT, |
| hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, |
| [gap_name])) |
| |
| gap_short_name = hci_packets.GapData() |
| gap_short_name.data_type = hci_packets.GapDataType.SHORTENED_LOCAL_NAME |
| gap_short_name.data = list(bytes(b'Im_A_C')) |
| |
| self.send_hal_hci_command( |
| hci_packets.LeSetExtendedAdvertisingScanResponseBuilder( |
| advertising_handle, |
| hci_packets.Operation.COMPLETE_ADVERTISEMENT, |
| hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, |
| [gap_short_name])) |
| |
| enabled_set = hci_packets.EnabledSet() |
| enabled_set.advertising_handle = 0 |
| enabled_set.duration = 0 |
| enabled_set.max_extended_advertising_events = 0 |
| self.send_hal_hci_command( |
| hci_packets.LeSetExtendedAdvertisingEnableBuilder( |
| hci_packets.Enable.ENABLED, [enabled_set])) |
| |
| hci_event_asserts.assert_event_occurs( |
| lambda packet: b'Im_A_Cert' in packet.event) |
| |
| self.send_hal_hci_command( |
| hci_packets.LeSetExtendedAdvertisingEnableBuilder( |
| hci_packets.Enable.DISABLED, [enabled_set])) |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedScanEnableBuilder( |
| hci_packets.Enable.DISABLED, |
| hci_packets.FilterDuplicates.DISABLED, 0, 0), True) |
| |
| def test_le_connection_dut_advertises(self): |
| self.register_for_le_event(hci_packets.SubeventCode.CONNECTION_COMPLETE) |
| with EventCallbackStream(self.device_under_test.hci.FetchLeSubevents(empty_proto.Empty())) as le_event_stream, \ |
| EventCallbackStream(self.device_under_test.hci.FetchEvents(empty_proto.Empty())) as event_stream, \ |
| EventCallbackStream(self.device_under_test.hci.FetchAclPackets(empty_proto.Empty())) as acl_data_stream, \ |
| EventCallbackStream(self.cert_device.hal.FetchHciEvent(empty_proto.Empty())) as cert_hci_event_stream, \ |
| EventCallbackStream(self.cert_device.hal.FetchHciAcl(empty_proto.Empty())) as cert_acl_data_stream: |
| |
| le_event_asserts = EventAsserts(le_event_stream) |
| event_asserts = EventAsserts(event_stream) |
| cert_hci_event_asserts = EventAsserts(cert_hci_event_stream) |
| acl_data_asserts = EventAsserts(acl_data_stream) |
| cert_acl_data_asserts = EventAsserts(cert_acl_data_stream) |
| |
| # Cert Connects |
| self.send_hal_hci_command( |
| hci_packets.LeSetRandomAddressBuilder('0C:05:04:03:02:01')) |
| phy_scan_params = hci_packets.LeCreateConnPhyScanParameters() |
| phy_scan_params.scan_interval = 0x60 |
| phy_scan_params.scan_window = 0x30 |
| phy_scan_params.conn_interval_min = 0x18 |
| phy_scan_params.conn_interval_max = 0x28 |
| phy_scan_params.conn_latency = 0 |
| phy_scan_params.supervision_timeout = 0x1f4 |
| phy_scan_params.min_ce_length = 0 |
| phy_scan_params.max_ce_length = 0 |
| self.send_hal_hci_command( |
| hci_packets.LeExtendedCreateConnectionBuilder( |
| hci_packets.InitiatorFilterPolicy.USE_PEER_ADDRESS, |
| hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, |
| hci_packets.AddressType.RANDOM_DEVICE_ADDRESS, |
| '0D:05:04:03:02:01', 1, [phy_scan_params])) |
| |
| # DUT Advertises |
| advertising_handle = 0 |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingLegacyParametersBuilder( |
| advertising_handle, |
| hci_packets.LegacyAdvertisingProperties.ADV_IND, |
| 400, |
| 450, |
| 7, |
| hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, |
| hci_packets.PeerAddressType. |
| PUBLIC_DEVICE_OR_IDENTITY_ADDRESS, |
| '00:00:00:00:00:00', |
| hci_packets.AdvertisingFilterPolicy.ALL_DEVICES, |
| 0xF8, |
| 1, #SID |
| hci_packets.Enable.DISABLED # Scan request notification |
| ), |
| True) |
| |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingRandomAddressBuilder( |
| advertising_handle, '0D:05:04:03:02:01'), True) |
| |
| gap_name = hci_packets.GapData() |
| gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME |
| gap_name.data = list( |
| bytes(b'Im_The_DUT!')) # TODO: Fix and remove ! |
| |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingDataBuilder( |
| advertising_handle, |
| hci_packets.Operation.COMPLETE_ADVERTISEMENT, |
| hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, |
| [gap_name]), True) |
| |
| gap_short_name = hci_packets.GapData() |
| gap_short_name.data_type = hci_packets.GapDataType.SHORTENED_LOCAL_NAME |
| gap_short_name.data = list(bytes(b'Im_The_D')) |
| |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingScanResponseBuilder( |
| advertising_handle, |
| hci_packets.Operation.COMPLETE_ADVERTISEMENT, |
| hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, |
| [gap_short_name]), True) |
| |
| enabled_set = hci_packets.EnabledSet() |
| enabled_set.advertising_handle = advertising_handle |
| enabled_set.duration = 0 |
| enabled_set.max_extended_advertising_events = 0 |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingEnableBuilder( |
| hci_packets.Enable.ENABLED, [enabled_set]), True) |
| |
| # Check for success of Enable |
| event_asserts.assert_event_occurs( |
| lambda packet: b'\x0e\x04\x01\x39\x20\x00' in packet.event) |
| |
| conn_handle = 0xfff |
| |
| def event_handle(packet): |
| packet_bytes = packet.event |
| if b'\x3e\x13\x01\x00' in packet_bytes: |
| nonlocal conn_handle |
| cc_view = hci_packets.LeConnectionCompleteView( |
| hci_packets.LeMetaEventView( |
| hci_packets.EventPacketView( |
| bt_packets.PacketViewLittleEndian( |
| list(packet_bytes))))) |
| conn_handle = cc_view.GetConnectionHandle() |
| return True |
| return False |
| |
| def payload_handle(packet): |
| packet_bytes = packet.payload |
| if b'\x3e\x13\x01\x00' in packet_bytes: |
| nonlocal conn_handle |
| cc_view = hci_packets.LeConnectionCompleteView( |
| hci_packets.LeMetaEventView( |
| hci_packets.EventPacketView( |
| bt_packets.PacketViewLittleEndian( |
| list(packet_bytes))))) |
| conn_handle = cc_view.GetConnectionHandle() |
| return True |
| return False |
| |
| cert_hci_event_asserts.assert_event_occurs(payload_handle) |
| cert_handle = conn_handle |
| conn_handle = 0xfff |
| le_event_asserts.assert_event_occurs(event_handle) |
| dut_handle = conn_handle |
| if dut_handle == 0xfff: |
| logging.warning("Failed to get the DUT handle") |
| return False |
| if cert_handle == 0xfff: |
| logging.warning("Failed to get the CERT handle") |
| return False |
| |
| # Send ACL Data |
| self.enqueue_acl_data( |
| dut_handle, hci_packets.PacketBoundaryFlag. |
| FIRST_NON_AUTOMATICALLY_FLUSHABLE, |
| hci_packets.BroadcastFlag.POINT_TO_POINT, |
| bytes(b'Just SomeAclData')) |
| self.send_hal_acl_data( |
| cert_handle, hci_packets.PacketBoundaryFlag. |
| FIRST_NON_AUTOMATICALLY_FLUSHABLE, |
| hci_packets.BroadcastFlag.POINT_TO_POINT, |
| bytes(b'Just SomeMoreAclData')) |
| |
| cert_acl_data_asserts.assert_event_occurs( |
| lambda packet: logging.debug(packet.payload) or b'SomeAclData' in packet.payload |
| ) |
| acl_data_asserts.assert_event_occurs( |
| lambda packet: logging.debug(packet.data) or b'SomeMoreAclData' in packet.data |
| ) |
| |
| def test_le_white_list_connection_cert_advertises(self): |
| self.register_for_le_event(hci_packets.SubeventCode.CONNECTION_COMPLETE) |
| with EventCallbackStream(self.device_under_test.hci.FetchLeSubevents(empty_proto.Empty())) as le_event_stream, \ |
| EventCallbackStream(self.cert_device.hal.FetchHciEvent(empty_proto.Empty())) as cert_hci_event_stream: |
| le_event_asserts = EventAsserts(le_event_stream) |
| cert_hci_event_asserts = EventAsserts(cert_hci_event_stream) |
| |
| # DUT Connects |
| self.enqueue_hci_command( |
| hci_packets.LeSetRandomAddressBuilder('0D:05:04:03:02:01'), |
| True) |
| self.enqueue_hci_command( |
| hci_packets.LeAddDeviceToWhiteListBuilder( |
| hci_packets.WhiteListAddressType.RANDOM, |
| '0C:05:04:03:02:01'), True) |
| phy_scan_params = hci_packets.LeCreateConnPhyScanParameters() |
| phy_scan_params.scan_interval = 0x60 |
| phy_scan_params.scan_window = 0x30 |
| phy_scan_params.conn_interval_min = 0x18 |
| phy_scan_params.conn_interval_max = 0x28 |
| phy_scan_params.conn_latency = 0 |
| phy_scan_params.supervision_timeout = 0x1f4 |
| phy_scan_params.min_ce_length = 0 |
| phy_scan_params.max_ce_length = 0 |
| self.enqueue_hci_command( |
| hci_packets.LeExtendedCreateConnectionBuilder( |
| hci_packets.InitiatorFilterPolicy.USE_WHITE_LIST, |
| hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, |
| hci_packets.AddressType.RANDOM_DEVICE_ADDRESS, |
| 'BA:D5:A4:A3:A2:A1', 1, [phy_scan_params]), False) |
| |
| # CERT Advertises |
| advertising_handle = 1 |
| self.send_hal_hci_command( |
| hci_packets.LeSetExtendedAdvertisingLegacyParametersBuilder( |
| advertising_handle, |
| hci_packets.LegacyAdvertisingProperties.ADV_IND, |
| 512, |
| 768, |
| 7, |
| hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, |
| hci_packets.PeerAddressType. |
| PUBLIC_DEVICE_OR_IDENTITY_ADDRESS, |
| 'A6:A5:A4:A3:A2:A1', |
| hci_packets.AdvertisingFilterPolicy.ALL_DEVICES, |
| 0x7F, |
| 0, # SID |
| hci_packets.Enable.DISABLED # Scan request notification |
| )) |
| |
| self.send_hal_hci_command( |
| hci_packets.LeSetExtendedAdvertisingRandomAddressBuilder( |
| advertising_handle, '0C:05:04:03:02:01')) |
| |
| gap_name = hci_packets.GapData() |
| gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME |
| gap_name.data = list(bytes(b'Im_A_Cert!')) # TODO: Fix and remove ! |
| |
| self.send_hal_hci_command( |
| hci_packets.LeSetExtendedAdvertisingDataBuilder( |
| advertising_handle, |
| hci_packets.Operation.COMPLETE_ADVERTISEMENT, |
| hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, |
| [gap_name])) |
| enabled_set = hci_packets.EnabledSet() |
| enabled_set.advertising_handle = 1 |
| enabled_set.duration = 0 |
| enabled_set.max_extended_advertising_events = 0 |
| self.send_hal_hci_command( |
| hci_packets.LeSetExtendedAdvertisingEnableBuilder( |
| hci_packets.Enable.ENABLED, [enabled_set])) |
| |
| # LeConnectionComplete |
| cert_hci_event_asserts.assert_event_occurs( |
| lambda packet: b'\x3e\x13\x01\x00' in packet.payload, |
| timeout=timedelta(seconds=20)) |
| le_event_asserts.assert_event_occurs( |
| lambda packet: b'\x3e\x13\x01\x00' in packet.event) |
| |
| def test_connection_dut_connects(self): |
| self.register_for_event(hci_packets.EventCode.CONNECTION_COMPLETE) |
| self.register_for_event( |
| hci_packets.EventCode.CONNECTION_PACKET_TYPE_CHANGED) |
| self.enqueue_hci_command( |
| hci_packets.WritePageTimeoutBuilder(0x4000), True) |
| with EventCallbackStream(self.device_under_test.hci.FetchEvents(empty_proto.Empty())) as hci_event_stream, \ |
| EventCallbackStream(self.device_under_test.hci.FetchAclPackets(empty_proto.Empty())) as acl_data_stream, \ |
| EventCallbackStream(self.cert_device.hal.FetchHciEvent(empty_proto.Empty())) as cert_hci_event_stream, \ |
| EventCallbackStream(self.cert_device.hal.FetchHciAcl(empty_proto.Empty())) as cert_acl_data_stream: |
| |
| cert_hci_event_asserts = EventAsserts(cert_hci_event_stream) |
| hci_event_asserts = EventAsserts(hci_event_stream) |
| acl_data_asserts = EventAsserts(acl_data_stream) |
| cert_acl_data_asserts = EventAsserts(cert_acl_data_stream) |
| |
| address = hci_packets.Address() |
| |
| def get_address_from_complete(packet): |
| packet_bytes = packet.payload |
| if b'\x0e\x0a\x01\x09\x10' in packet_bytes: |
| nonlocal address |
| addr_view = hci_packets.ReadBdAddrCompleteView( |
| hci_packets.CommandCompleteView( |
| hci_packets.EventPacketView( |
| bt_packets.PacketViewLittleEndian( |
| list(packet_bytes))))) |
| address = addr_view.GetBdAddr() |
| return True |
| return False |
| |
| # CERT Enables scans and gets its address |
| self.send_hal_hci_command(hci_packets.ReadBdAddrBuilder()) |
| |
| cert_hci_event_asserts.assert_event_occurs( |
| get_address_from_complete) |
| self.send_hal_hci_command( |
| hci_packets.WriteScanEnableBuilder( |
| hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN)) |
| |
| # DUT Connects |
| self.enqueue_hci_command( |
| hci_packets.CreateConnectionBuilder( |
| address, |
| 0xcc18, # Packet Type |
| hci_packets.PageScanRepetitionMode.R0, |
| 0, |
| hci_packets.ClockOffsetValid.INVALID, |
| hci_packets.CreateConnectionRoleSwitch.ALLOW_ROLE_SWITCH), |
| False) |
| |
| # Cert Accepts |
| connection_request = None |
| |
| def get_connect_request(packet): |
| if b'\x04\x0a' in packet.payload: |
| nonlocal connection_request |
| connection_request = hci_packets.ConnectionRequestView( |
| hci_packets.EventPacketView( |
| bt_packets.PacketViewLittleEndian( |
| list(packet.payload)))) |
| return True |
| return False |
| |
| # Cert Accepts |
| cert_hci_event_asserts.assert_event_occurs( |
| get_connect_request, timeout=timedelta(seconds=20)) |
| self.send_hal_hci_command( |
| hci_packets.AcceptConnectionRequestBuilder( |
| connection_request.GetBdAddr(), |
| hci_packets.AcceptConnectionRequestRole.REMAIN_SLAVE)) |
| |
| conn_handle = 0xfff |
| |
| def get_handle(packet_bytes): |
| if b'\x03\x0b\x00' in packet_bytes: |
| nonlocal conn_handle |
| cc_view = hci_packets.ConnectionCompleteView( |
| hci_packets.EventPacketView( |
| bt_packets.PacketViewLittleEndian( |
| list(packet_bytes)))) |
| conn_handle = cc_view.GetConnectionHandle() |
| return True |
| return False |
| |
| def event_handle(packet): |
| packet_bytes = packet.event |
| return get_handle(packet_bytes) |
| |
| def payload_handle(packet): |
| packet_bytes = packet.payload |
| return get_handle(packet_bytes) |
| |
| cert_hci_event_asserts.assert_event_occurs(payload_handle) |
| cert_handle = conn_handle |
| conn_handle = 0xfff |
| hci_event_asserts.assert_event_occurs(event_handle) |
| dut_handle = conn_handle |
| if dut_handle == 0xfff: |
| logging.warning("Failed to get the DUT handle") |
| return False |
| if cert_handle == 0xfff: |
| logging.warning("Failed to get the CERT handle") |
| return False |
| |
| # Send ACL Data |
| self.enqueue_acl_data( |
| dut_handle, hci_packets.PacketBoundaryFlag. |
| FIRST_NON_AUTOMATICALLY_FLUSHABLE, |
| hci_packets.BroadcastFlag.POINT_TO_POINT, |
| bytes(b'Just SomeAclData')) |
| self.send_hal_acl_data( |
| cert_handle, hci_packets.PacketBoundaryFlag. |
| FIRST_NON_AUTOMATICALLY_FLUSHABLE, |
| hci_packets.BroadcastFlag.POINT_TO_POINT, |
| bytes(b'Just SomeMoreAclData')) |
| |
| cert_acl_data_asserts.assert_event_occurs( |
| lambda packet: b'SomeAclData' in packet.payload) |
| acl_data_asserts.assert_event_occurs( |
| lambda packet: b'SomeMoreAclData' in packet.data) |
| |
| def test_connection_cert_connects(self): |
| self.register_for_event(hci_packets.EventCode.CONNECTION_COMPLETE) |
| self.register_for_event( |
| hci_packets.EventCode.CONNECTION_PACKET_TYPE_CHANGED) |
| self.register_for_event(hci_packets.EventCode.CONNECTION_REQUEST) |
| self.send_hal_hci_command(hci_packets.WritePageTimeoutBuilder(0x4000)) |
| with EventCallbackStream(self.device_under_test.hci.FetchEvents(empty_proto.Empty())) as hci_event_stream, \ |
| EventCallbackStream(self.device_under_test.hci.FetchAclPackets(empty_proto.Empty())) as acl_data_stream, \ |
| EventCallbackStream(self.cert_device.hal.FetchHciEvent(empty_proto.Empty())) as cert_hci_event_stream, \ |
| EventCallbackStream(self.cert_device.hal.FetchHciAcl(empty_proto.Empty())) as cert_acl_data_stream: |
| |
| hci_event_asserts = EventAsserts(hci_event_stream) |
| cert_hci_event_asserts = EventAsserts(cert_hci_event_stream) |
| acl_data_asserts = EventAsserts(acl_data_stream) |
| cert_acl_data_asserts = EventAsserts(cert_acl_data_stream) |
| |
| address = hci_packets.Address() |
| |
| def get_address_from_complete(packet): |
| packet_bytes = packet.event |
| if b'\x0e\x0a\x01\x09\x10' in packet_bytes: |
| nonlocal address |
| addr_view = hci_packets.ReadBdAddrCompleteView( |
| hci_packets.CommandCompleteView( |
| hci_packets.EventPacketView( |
| bt_packets.PacketViewLittleEndian( |
| list(packet_bytes))))) |
| address = addr_view.GetBdAddr() |
| return True |
| return False |
| |
| # DUT Enables scans and gets its address |
| self.enqueue_hci_command( |
| hci_packets.WriteScanEnableBuilder( |
| hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN), True) |
| self.enqueue_hci_command(hci_packets.ReadBdAddrBuilder(), True) |
| |
| hci_event_asserts.assert_event_occurs(get_address_from_complete) |
| |
| # Cert Connects |
| self.send_hal_hci_command( |
| hci_packets.CreateConnectionBuilder( |
| address, |
| 0xcc18, # Packet Type |
| hci_packets.PageScanRepetitionMode.R0, |
| 0, |
| hci_packets.ClockOffsetValid.INVALID, |
| hci_packets.CreateConnectionRoleSwitch.ALLOW_ROLE_SWITCH)) |
| |
| # DUT Accepts |
| connection_request = None |
| |
| def get_connect_request(packet): |
| if b'\x04\x0a' in packet.event: |
| nonlocal connection_request |
| connection_request = hci_packets.ConnectionRequestView( |
| hci_packets.EventPacketView( |
| bt_packets.PacketViewLittleEndian( |
| list(packet.event)))) |
| return True |
| return False |
| |
| hci_event_asserts.assert_event_occurs( |
| get_connect_request, timeout=timedelta(seconds=20)) |
| self.enqueue_hci_command( |
| hci_packets.AcceptConnectionRequestBuilder( |
| connection_request.GetBdAddr(), |
| hci_packets.AcceptConnectionRequestRole.REMAIN_SLAVE), |
| False) |
| |
| conn_handle = 0xfff |
| |
| def get_handle(packet_bytes): |
| if b'\x03\x0b\x00' in packet_bytes: |
| nonlocal conn_handle |
| cc_view = hci_packets.ConnectionCompleteView( |
| hci_packets.EventPacketView( |
| bt_packets.PacketViewLittleEndian( |
| list(packet_bytes)))) |
| conn_handle = cc_view.GetConnectionHandle() |
| return True |
| return False |
| |
| def event_handle(packet): |
| packet_bytes = packet.event |
| return get_handle(packet_bytes) |
| |
| def payload_handle(packet): |
| packet_bytes = packet.payload |
| return get_handle(packet_bytes) |
| |
| cert_hci_event_asserts.assert_event_occurs(payload_handle) |
| cert_handle = conn_handle |
| conn_handle = 0xfff |
| hci_event_asserts.assert_event_occurs(event_handle) |
| dut_handle = conn_handle |
| if dut_handle == 0xfff: |
| logging.warning("Failed to get the DUT handle") |
| return False |
| if cert_handle == 0xfff: |
| logging.warning("Failed to get the CERT handle") |
| return False |
| |
| # Send ACL Data |
| self.enqueue_acl_data( |
| dut_handle, hci_packets.PacketBoundaryFlag. |
| FIRST_NON_AUTOMATICALLY_FLUSHABLE, |
| hci_packets.BroadcastFlag.POINT_TO_POINT, |
| bytes(b'This is just SomeAclData')) |
| self.send_hal_acl_data( |
| cert_handle, hci_packets.PacketBoundaryFlag. |
| FIRST_NON_AUTOMATICALLY_FLUSHABLE, |
| hci_packets.BroadcastFlag.POINT_TO_POINT, |
| bytes(b'This is just SomeMoreAclData')) |
| |
| cert_acl_data_asserts.assert_event_occurs( |
| lambda packet: b'SomeAclData' in packet.payload) |
| acl_data_asserts.assert_event_occurs( |
| lambda packet: b'SomeMoreAclData' in packet.data) |
