From 4646b54284fe93c72ef3114591dd362aeeea50b7 Mon Sep 17 00:00:00 2001 From: Zach Johnson Date: Thu, 12 Nov 2020 16:53:45 -0800 Subject: [PATCH] Improve signal to noise in DirectHciTest import concrete objects instead of hci_packets.xyz everywhere Bug: 171749953 Tag: #gd-refactor Test: gd/cert/run --host Change-Id: Ie906397e586ee6c12ef533d9201f7e11f4ca5f35 --- gd/hci/cert/direct_hci_test.py | 290 +++++++++++++++++++++++------------------ 1 file changed, 160 insertions(+), 130 deletions(-) diff --git a/gd/hci/cert/direct_hci_test.py b/gd/hci/cert/direct_hci_test.py index 9c779e87d..f48fb680b 100644 --- a/gd/hci/cert/direct_hci_test.py +++ b/gd/hci/cert/direct_hci_test.py @@ -24,7 +24,56 @@ from cert.py_hal import PyHal from cert.py_hci import PyHci from cert.truth import assertThat from hci.facade import facade_pb2 as hci_facade -from bluetooth_packets_python3 import hci_packets +from bluetooth_packets_python3.hci_packets import EventCode +from bluetooth_packets_python3.hci_packets import LoopbackMode +from bluetooth_packets_python3.hci_packets import WriteLoopbackModeBuilder +from bluetooth_packets_python3.hci_packets import ReadLocalNameBuilder +from bluetooth_packets_python3.hci_packets import WriteScanEnableBuilder +from bluetooth_packets_python3.hci_packets import ScanEnable +from bluetooth_packets_python3.hci_packets import InquiryBuilder +from bluetooth_packets_python3.hci_packets import SubeventCode +from bluetooth_packets_python3.hci_packets import LeSetRandomAddressBuilder +from bluetooth_packets_python3.hci_packets import PhyScanParameters +from bluetooth_packets_python3.hci_packets import LeScanType +from bluetooth_packets_python3.hci_packets import LeSetExtendedScanParametersBuilder +from bluetooth_packets_python3.hci_packets import OwnAddressType +from bluetooth_packets_python3.hci_packets import LeScanningFilterPolicy +from bluetooth_packets_python3.hci_packets import Enable +from bluetooth_packets_python3.hci_packets import FilterDuplicates +from bluetooth_packets_python3.hci_packets import LeSetExtendedAdvertisingLegacyParametersBuilder +from bluetooth_packets_python3.hci_packets import LegacyAdvertisingProperties +from bluetooth_packets_python3.hci_packets import PeerAddressType +from bluetooth_packets_python3.hci_packets import AdvertisingFilterPolicy +from bluetooth_packets_python3.hci_packets import LeSetExtendedAdvertisingRandomAddressBuilder +from bluetooth_packets_python3.hci_packets import GapData +from bluetooth_packets_python3.hci_packets import GapDataType +from bluetooth_packets_python3.hci_packets import LeSetExtendedAdvertisingDataBuilder +from bluetooth_packets_python3.hci_packets import Operation +from bluetooth_packets_python3.hci_packets import FragmentPreference +from bluetooth_packets_python3.hci_packets import LeSetExtendedAdvertisingScanResponseBuilder +from bluetooth_packets_python3.hci_packets import LeSetExtendedAdvertisingEnableBuilder +from bluetooth_packets_python3.hci_packets import LeSetExtendedScanEnableBuilder +from bluetooth_packets_python3.hci_packets import EnabledSet +from bluetooth_packets_python3.hci_packets import LeCreateConnPhyScanParameters +from bluetooth_packets_python3.hci_packets import LeExtendedCreateConnectionBuilder +from bluetooth_packets_python3.hci_packets import InitiatorFilterPolicy +from bluetooth_packets_python3.hci_packets import AddressType +from bluetooth_packets_python3.hci_packets import BroadcastFlag +from bluetooth_packets_python3.hci_packets import ConnectListAddressType +from bluetooth_packets_python3.hci_packets import LeAddDeviceToConnectListBuilder +from bluetooth_packets_python3.hci_packets import LeSetRandomAddressBuilder +from bluetooth_packets_python3.hci_packets import WritePageTimeoutBuilder +from bluetooth_packets_python3.hci_packets import ReadBdAddrBuilder +from bluetooth_packets_python3.hci_packets import CreateConnectionBuilder +from bluetooth_packets_python3.hci_packets import PageScanRepetitionMode +from bluetooth_packets_python3.hci_packets import ClockOffsetValid +from bluetooth_packets_python3.hci_packets import CreateConnectionRoleSwitch +from bluetooth_packets_python3.hci_packets import AcceptConnectionRequestBuilder +from bluetooth_packets_python3.hci_packets import AcceptConnectionRequestRole +from bluetooth_packets_python3.hci_packets import PacketBoundaryFlag +from bluetooth_packets_python3.hci_packets import ResetBuilder +from bluetooth_packets_python3.hci_packets import Lap +from bluetooth_packets_python3.hci_packets import OpCode class DirectHciTest(GdBaseTestClass): @@ -36,7 +85,7 @@ class DirectHciTest(GdBaseTestClass): super().setup_test() self.dut_hci = PyHci(self.dut, acl_streaming=True) self.cert_hal = PyHal(self.cert) - self.cert_hal.send_hci_command(hci_packets.ResetBuilder()) + self.cert_hal.send_hci_command(ResetBuilder()) def teardown_test(self): self.dut_hci.close() @@ -60,98 +109,87 @@ class DirectHciTest(GdBaseTestClass): def test_local_hci_cmd_and_event(self): # Loopback mode responds with ACL and SCO connection complete - self.dut_hci.register_for_events(hci_packets.EventCode.LOOPBACK_COMMAND) + self.dut_hci.register_for_events(EventCode.LOOPBACK_COMMAND) - self.dut_hci.send_command_with_complete( - hci_packets.WriteLoopbackModeBuilder(hci_packets.LoopbackMode.ENABLE_LOCAL)) + self.dut_hci.send_command_with_complete(WriteLoopbackModeBuilder(LoopbackMode.ENABLE_LOCAL)) - cmd2loop = hci_packets.ReadLocalNameBuilder() - self.dut_hci.send_command_with_complete(cmd2loop) + self.dut_hci.send_command_with_complete(ReadLocalNameBuilder()) - looped_bytes = bytes(cmd2loop.Serialize()) + looped_bytes = bytes(ReadLocalNameBuilder().Serialize()) assertThat(self.dut_hci.get_event_stream()).emits(lambda packet: looped_bytes in packet.payload) def test_inquiry_from_dut(self): - self.dut_hci.register_for_events(hci_packets.EventCode.INQUIRY_RESULT) + self.dut_hci.register_for_events(EventCode.INQUIRY_RESULT) - self.cert_hal.send_hci_command(hci_packets.WriteScanEnableBuilder(hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN)) - lap = hci_packets.Lap() + self.cert_hal.send_hci_command(WriteScanEnableBuilder(ScanEnable.INQUIRY_AND_PAGE_SCAN)) + lap = Lap() lap.lap = 0x33 - self.dut_hci.send_command_with_status(hci_packets.InquiryBuilder(lap, 0x30, 0xff)) - assertThat(self.dut_hci.get_event_stream()).emits( - HciMatchers.EventWithCode(hci_packets.EventCode.INQUIRY_RESULT)) + self.dut_hci.send_command_with_status(InquiryBuilder(lap, 0x30, 0xff)) + assertThat(self.dut_hci.get_event_stream()).emits(HciMatchers.EventWithCode(EventCode.INQUIRY_RESULT)) def test_le_ad_scan_cert_advertises(self): - self.dut_hci.register_for_le_events(hci_packets.SubeventCode.EXTENDED_ADVERTISING_REPORT) - self.dut_hci.register_for_le_events(hci_packets.SubeventCode.ADVERTISING_REPORT) + self.dut_hci.register_for_le_events(SubeventCode.EXTENDED_ADVERTISING_REPORT, SubeventCode.ADVERTISING_REPORT) # DUT Scans - self.dut_hci.send_command_with_complete(hci_packets.LeSetRandomAddressBuilder('0D:05:04:03:02:01')) - phy_scan_params = hci_packets.PhyScanParameters() + self.dut_hci.send_command_with_complete(LeSetRandomAddressBuilder('0D:05:04:03:02:01')) + phy_scan_params = 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 + phy_scan_params.le_scan_type = LeScanType.ACTIVE self.dut_hci.send_command_with_complete( - hci_packets.LeSetExtendedScanParametersBuilder(hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, - hci_packets.LeScanningFilterPolicy.ACCEPT_ALL, 1, - [phy_scan_params])) + LeSetExtendedScanParametersBuilder(OwnAddressType.RANDOM_DEVICE_ADDRESS, LeScanningFilterPolicy.ACCEPT_ALL, + 1, [phy_scan_params])) self.dut_hci.send_command_with_complete( - hci_packets.LeSetExtendedScanEnableBuilder(hci_packets.Enable.ENABLED, - hci_packets.FilterDuplicates.DISABLED, 0, 0)) + LeSetExtendedScanEnableBuilder(Enable.ENABLED, FilterDuplicates.DISABLED, 0, 0)) # CERT Advertises advertising_handle = 0 self.cert_hal.send_hci_command( - hci_packets.LeSetExtendedAdvertisingLegacyParametersBuilder( + LeSetExtendedAdvertisingLegacyParametersBuilder( advertising_handle, - hci_packets.LegacyAdvertisingProperties.ADV_IND, + LegacyAdvertisingProperties.ADV_IND, 512, 768, 7, - hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, - hci_packets.PeerAddressType.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS, + OwnAddressType.RANDOM_DEVICE_ADDRESS, + PeerAddressType.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS, 'A6:A5:A4:A3:A2:A1', - hci_packets.AdvertisingFilterPolicy.ALL_DEVICES, + AdvertisingFilterPolicy.ALL_DEVICES, 0xF7, 1, # SID - hci_packets.Enable.DISABLED # Scan request notification + Enable.DISABLED # Scan request notification )) self.cert_hal.send_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 + LeSetExtendedAdvertisingRandomAddressBuilder(advertising_handle, '0C:05:04:03:02:01')) + gap_name = GapData() + gap_name.data_type = GapDataType.COMPLETE_LOCAL_NAME gap_name.data = list(bytes(b'Im_A_Cert')) self.cert_hal.send_hci_command( - hci_packets.LeSetExtendedAdvertisingDataBuilder( - advertising_handle, hci_packets.Operation.COMPLETE_ADVERTISEMENT, - hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_name])) + LeSetExtendedAdvertisingDataBuilder(advertising_handle, Operation.COMPLETE_ADVERTISEMENT, + 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 = GapData() + gap_short_name.data_type = GapDataType.SHORTENED_LOCAL_NAME gap_short_name.data = list(bytes(b'Im_A_C')) self.cert_hal.send_hci_command( - hci_packets.LeSetExtendedAdvertisingScanResponseBuilder( - advertising_handle, hci_packets.Operation.COMPLETE_ADVERTISEMENT, - hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_short_name])) + LeSetExtendedAdvertisingScanResponseBuilder(advertising_handle, Operation.COMPLETE_ADVERTISEMENT, + FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_short_name])) - enabled_set = hci_packets.EnabledSet() + enabled_set = EnabledSet() enabled_set.advertising_handle = 0 enabled_set.duration = 0 enabled_set.max_extended_advertising_events = 0 - self.cert_hal.send_hci_command( - hci_packets.LeSetExtendedAdvertisingEnableBuilder(hci_packets.Enable.ENABLED, [enabled_set])) + self.cert_hal.send_hci_command(LeSetExtendedAdvertisingEnableBuilder(Enable.ENABLED, [enabled_set])) assertThat(self.dut_hci.get_le_event_stream()).emits(lambda packet: b'Im_A_Cert' in packet.payload) - self.cert_hal.send_hci_command( - hci_packets.LeSetExtendedAdvertisingEnableBuilder(hci_packets.Enable.DISABLED, [enabled_set])) + self.cert_hal.send_hci_command(LeSetExtendedAdvertisingEnableBuilder(Enable.DISABLED, [enabled_set])) self.dut_hci.send_command_with_complete( - hci_packets.LeSetExtendedScanEnableBuilder(hci_packets.Enable.DISABLED, - hci_packets.FilterDuplicates.DISABLED, 0, 0)) + LeSetExtendedScanEnableBuilder(Enable.DISABLED, FilterDuplicates.DISABLED, 0, 0)) def _verify_le_connection_complete(self): cert_conn_complete_capture = HalCaptures.LeConnectionCompleteCapture() @@ -166,7 +204,7 @@ class DirectHciTest(GdBaseTestClass): @staticmethod def _create_phy_scan_params(): - phy_scan_params = hci_packets.LeCreateConnPhyScanParameters() + phy_scan_params = LeCreateConnPhyScanParameters() phy_scan_params.scan_interval = 0x60 phy_scan_params.scan_window = 0x30 phy_scan_params.conn_interval_min = 0x18 @@ -178,74 +216,70 @@ class DirectHciTest(GdBaseTestClass): return phy_scan_params def test_le_connection_dut_advertises(self): - self.dut_hci.register_for_le_events(hci_packets.SubeventCode.CONNECTION_COMPLETE) - self.dut_hci.register_for_le_events(hci_packets.SubeventCode.ADVERTISING_SET_TERMINATED) - self.dut_hci.register_for_le_events(hci_packets.SubeventCode.ENHANCED_CONNECTION_COMPLETE) + self.dut_hci.register_for_le_events(SubeventCode.CONNECTION_COMPLETE, SubeventCode.ADVERTISING_SET_TERMINATED, + SubeventCode.ENHANCED_CONNECTION_COMPLETE) # Cert Connects - self.cert_hal.send_hci_command(hci_packets.LeSetRandomAddressBuilder('0C:05:04:03:02:01')) + self.cert_hal.send_hci_command(LeSetRandomAddressBuilder('0C:05:04:03:02:01')) phy_scan_params = DirectHciTest._create_phy_scan_params() self.cert_hal.send_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])) + LeExtendedCreateConnectionBuilder(InitiatorFilterPolicy.USE_PEER_ADDRESS, + OwnAddressType.RANDOM_DEVICE_ADDRESS, AddressType.RANDOM_DEVICE_ADDRESS, + '0D:05:04:03:02:01', 1, [phy_scan_params])) # DUT Advertises advertising_handle = 0 self.dut_hci.send_command_with_complete( - hci_packets.LeSetExtendedAdvertisingLegacyParametersBuilder( + LeSetExtendedAdvertisingLegacyParametersBuilder( advertising_handle, - hci_packets.LegacyAdvertisingProperties.ADV_IND, + LegacyAdvertisingProperties.ADV_IND, 400, 450, 7, - hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, - hci_packets.PeerAddressType.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS, + OwnAddressType.RANDOM_DEVICE_ADDRESS, + PeerAddressType.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS, '00:00:00:00:00:00', - hci_packets.AdvertisingFilterPolicy.ALL_DEVICES, + AdvertisingFilterPolicy.ALL_DEVICES, 0xF8, 1, #SID - hci_packets.Enable.DISABLED # Scan request notification + Enable.DISABLED # Scan request notification )) self.dut_hci.send_command_with_complete( - hci_packets.LeSetExtendedAdvertisingRandomAddressBuilder(advertising_handle, '0D:05:04:03:02:01')) + LeSetExtendedAdvertisingRandomAddressBuilder(advertising_handle, '0D:05:04:03:02:01')) - gap_name = hci_packets.GapData() - gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME + gap_name = GapData() + gap_name.data_type = GapDataType.COMPLETE_LOCAL_NAME gap_name.data = list(bytes(b'Im_The_DUT')) self.dut_hci.send_command_with_complete( - hci_packets.LeSetExtendedAdvertisingDataBuilder( - advertising_handle, hci_packets.Operation.COMPLETE_ADVERTISEMENT, - hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_name])) + LeSetExtendedAdvertisingDataBuilder(advertising_handle, Operation.COMPLETE_ADVERTISEMENT, + 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 = GapData() + gap_short_name.data_type = GapDataType.SHORTENED_LOCAL_NAME gap_short_name.data = list(bytes(b'Im_The_D')) self.dut_hci.send_command_with_complete( - hci_packets.LeSetExtendedAdvertisingScanResponseBuilder( - advertising_handle, hci_packets.Operation.COMPLETE_ADVERTISEMENT, - hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_short_name])) + LeSetExtendedAdvertisingScanResponseBuilder(advertising_handle, Operation.COMPLETE_ADVERTISEMENT, + FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_short_name])) - enabled_set = hci_packets.EnabledSet() + enabled_set = EnabledSet() enabled_set.advertising_handle = advertising_handle enabled_set.duration = 0 enabled_set.max_extended_advertising_events = 0 - self.dut_hci.send_command_with_complete( - hci_packets.LeSetExtendedAdvertisingEnableBuilder(hci_packets.Enable.ENABLED, [enabled_set])) + self.dut_hci.send_command_with_complete(LeSetExtendedAdvertisingEnableBuilder(Enable.ENABLED, [enabled_set])) # Check for success of Enable assertThat(self.dut_hci.get_event_stream()).emits( - HciMatchers.CommandComplete(hci_packets.OpCode.LE_SET_EXTENDED_ADVERTISING_ENABLE)) + HciMatchers.CommandComplete(OpCode.LE_SET_EXTENDED_ADVERTISING_ENABLE)) (dut_handle, cert_handle) = self._verify_le_connection_complete() # 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')) + self.enqueue_acl_data(dut_handle, PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE, + BroadcastFlag.POINT_TO_POINT, bytes(b'Just SomeAclData')) + self.send_hal_acl_data(cert_handle, PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE, + BroadcastFlag.POINT_TO_POINT, bytes(b'Just SomeMoreAclData')) assertThat(self.cert_hal.get_acl_stream()).emits( lambda packet: logging.debug(packet.payload) or b'SomeAclData' in packet.payload) @@ -253,53 +287,50 @@ class DirectHciTest(GdBaseTestClass): lambda packet: logging.debug(packet.data) or b'SomeMoreAclData' in packet.data) def test_le_connect_list_connection_cert_advertises(self): - self.dut_hci.register_for_le_events(hci_packets.SubeventCode.CONNECTION_COMPLETE) - self.dut_hci.register_for_le_events(hci_packets.SubeventCode.ENHANCED_CONNECTION_COMPLETE) + self.dut_hci.register_for_le_events(SubeventCode.CONNECTION_COMPLETE, SubeventCode.ENHANCED_CONNECTION_COMPLETE) # DUT Connects - self.dut_hci.send_command_with_complete(hci_packets.LeSetRandomAddressBuilder('0D:05:04:03:02:01')) + self.dut_hci.send_command_with_complete(LeSetRandomAddressBuilder('0D:05:04:03:02:01')) self.dut_hci.send_command_with_complete( - hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM, '0C:05:04:03:02:01')) + LeAddDeviceToConnectListBuilder(ConnectListAddressType.RANDOM, '0C:05:04:03:02:01')) phy_scan_params = DirectHciTest._create_phy_scan_params() self.dut_hci.send_command_with_status( - hci_packets.LeExtendedCreateConnectionBuilder( - hci_packets.InitiatorFilterPolicy.USE_CONNECT_LIST, hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, - hci_packets.AddressType.RANDOM_DEVICE_ADDRESS, 'BA:D5:A4:A3:A2:A1', 1, [phy_scan_params])) + LeExtendedCreateConnectionBuilder(InitiatorFilterPolicy.USE_CONNECT_LIST, + OwnAddressType.RANDOM_DEVICE_ADDRESS, AddressType.RANDOM_DEVICE_ADDRESS, + 'BA:D5:A4:A3:A2:A1', 1, [phy_scan_params])) # CERT Advertises advertising_handle = 1 self.cert_hal.send_hci_command( - hci_packets.LeSetExtendedAdvertisingLegacyParametersBuilder( + LeSetExtendedAdvertisingLegacyParametersBuilder( advertising_handle, - hci_packets.LegacyAdvertisingProperties.ADV_IND, + LegacyAdvertisingProperties.ADV_IND, 512, 768, 7, - hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, - hci_packets.PeerAddressType.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS, + OwnAddressType.RANDOM_DEVICE_ADDRESS, + PeerAddressType.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS, 'A6:A5:A4:A3:A2:A1', - hci_packets.AdvertisingFilterPolicy.ALL_DEVICES, + AdvertisingFilterPolicy.ALL_DEVICES, 0x7F, 0, # SID - hci_packets.Enable.DISABLED # Scan request notification + Enable.DISABLED # Scan request notification )) self.cert_hal.send_hci_command( - hci_packets.LeSetExtendedAdvertisingRandomAddressBuilder(advertising_handle, '0C:05:04:03:02:01')) + 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 = GapData() + gap_name.data_type = GapDataType.COMPLETE_LOCAL_NAME gap_name.data = list(bytes(b'Im_A_Cert')) self.cert_hal.send_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() + LeSetExtendedAdvertisingDataBuilder(advertising_handle, Operation.COMPLETE_ADVERTISEMENT, + FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_name])) + enabled_set = EnabledSet() enabled_set.advertising_handle = 1 enabled_set.duration = 0 enabled_set.max_extended_advertising_events = 0 - self.cert_hal.send_hci_command( - hci_packets.LeSetExtendedAdvertisingEnableBuilder(hci_packets.Enable.ENABLED, [enabled_set])) + self.cert_hal.send_hci_command(LeSetExtendedAdvertisingEnableBuilder(Enable.ENABLED, [enabled_set])) # LeConnectionComplete self._verify_le_connection_complete() @@ -316,53 +347,52 @@ class DirectHciTest(GdBaseTestClass): return (dut_handle, cert_handle) def test_connection_dut_connects(self): - self.dut_hci.send_command_with_complete(hci_packets.WritePageTimeoutBuilder(0x4000)) + self.dut_hci.send_command_with_complete(WritePageTimeoutBuilder(0x4000)) # CERT Enables scans and gets its address - self.cert_hal.send_hci_command(hci_packets.ReadBdAddrBuilder()) + self.cert_hal.send_hci_command(ReadBdAddrBuilder()) cert_read_bd_addr_capture = HalCaptures.ReadBdAddrCompleteCapture() assertThat(self.cert_hal.get_hci_event_stream()).emits(cert_read_bd_addr_capture) address = cert_read_bd_addr_capture.get().GetBdAddr() - self.cert_hal.send_hci_command(hci_packets.WriteScanEnableBuilder(hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN)) + self.cert_hal.send_hci_command(WriteScanEnableBuilder(ScanEnable.INQUIRY_AND_PAGE_SCAN)) # DUT Connects self.dut_hci.send_command_with_status( - hci_packets.CreateConnectionBuilder( + CreateConnectionBuilder( address, 0xcc18, # Packet Type - hci_packets.PageScanRepetitionMode.R0, + PageScanRepetitionMode.R0, 0, - hci_packets.ClockOffsetValid.INVALID, - hci_packets.CreateConnectionRoleSwitch.ALLOW_ROLE_SWITCH)) + ClockOffsetValid.INVALID, + CreateConnectionRoleSwitch.ALLOW_ROLE_SWITCH)) # Cert Accepts connect_request_capture = HalCaptures.ConnectionRequestCapture() assertThat(self.cert_hal.get_hci_event_stream()).emits(connect_request_capture, timeout=timedelta(seconds=20)) connection_request = connect_request_capture.get() self.cert_hal.send_hci_command( - hci_packets.AcceptConnectionRequestBuilder(connection_request.GetBdAddr(), - hci_packets.AcceptConnectionRequestRole.REMAIN_PERIPHERAL)) + AcceptConnectionRequestBuilder(connection_request.GetBdAddr(), + AcceptConnectionRequestRole.REMAIN_PERIPHERAL)) (dut_handle, cert_handle) = self._verify_connection_complete() # 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')) + self.enqueue_acl_data(dut_handle, PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE, + BroadcastFlag.POINT_TO_POINT, bytes(b'Just SomeAclData')) + self.send_hal_acl_data(cert_handle, PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE, + BroadcastFlag.POINT_TO_POINT, bytes(b'Just SomeMoreAclData')) assertThat(self.cert_hal.get_acl_stream()).emits(lambda packet: b'SomeAclData' in packet.payload) assertThat(self.dut_hci.get_raw_acl_stream()).emits(lambda packet: b'SomeMoreAclData' in packet.data) def test_connection_cert_connects(self): - self.cert_hal.send_hci_command(hci_packets.WritePageTimeoutBuilder(0x4000)) + self.cert_hal.send_hci_command(WritePageTimeoutBuilder(0x4000)) # DUT Enables scans and gets its address - self.dut_hci.send_command_with_complete( - hci_packets.WriteScanEnableBuilder(hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN)) - self.dut_hci.send_command_with_complete(hci_packets.ReadBdAddrBuilder()) + self.dut_hci.send_command_with_complete(WriteScanEnableBuilder(ScanEnable.INQUIRY_AND_PAGE_SCAN)) + self.dut_hci.send_command_with_complete(ReadBdAddrBuilder()) read_bd_addr_capture = HciCaptures.ReadBdAddrCompleteCapture() assertThat(self.dut_hci.get_event_stream()).emits(read_bd_addr_capture) @@ -370,29 +400,29 @@ class DirectHciTest(GdBaseTestClass): # Cert Connects self.cert_hal.send_hci_command( - hci_packets.CreateConnectionBuilder( + CreateConnectionBuilder( address, 0xcc18, # Packet Type - hci_packets.PageScanRepetitionMode.R0, + PageScanRepetitionMode.R0, 0, - hci_packets.ClockOffsetValid.INVALID, - hci_packets.CreateConnectionRoleSwitch.ALLOW_ROLE_SWITCH)) + ClockOffsetValid.INVALID, + CreateConnectionRoleSwitch.ALLOW_ROLE_SWITCH)) # DUT Accepts connection_request_capture = HciCaptures.ConnectionRequestCapture() assertThat(self.dut_hci.get_event_stream()).emits(connection_request_capture, timeout=timedelta(seconds=20)) connection_request = connection_request_capture.get() self.dut_hci.send_command_with_status( - hci_packets.AcceptConnectionRequestBuilder(connection_request.GetBdAddr(), - hci_packets.AcceptConnectionRequestRole.REMAIN_PERIPHERAL)) + AcceptConnectionRequestBuilder(connection_request.GetBdAddr(), + AcceptConnectionRequestRole.REMAIN_PERIPHERAL)) (dut_handle, cert_handle) = self._verify_connection_complete() # 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')) + self.enqueue_acl_data(dut_handle, PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE, + BroadcastFlag.POINT_TO_POINT, bytes(b'This is just SomeAclData')) + self.send_hal_acl_data(cert_handle, PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE, + BroadcastFlag.POINT_TO_POINT, bytes(b'This is just SomeMoreAclData')) assertThat(self.cert_hal.get_acl_stream()).emits(lambda packet: b'SomeAclData' in packet.payload) assertThat(self.dut_hci.get_raw_acl_stream()).emits(lambda packet: b'SomeMoreAclData' in packet.data) -- 2.11.0