OSDN Git Service

Improve signal to noise in DirectHciTest
authorZach Johnson <zachoverflow@google.com>
Fri, 13 Nov 2020 00:53:45 +0000 (16:53 -0800)
committerZach Johnson <zachoverflow@google.com>
Wed, 18 Nov 2020 22:12:43 +0000 (14:12 -0800)
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

index 9c779e8..f48fb68 100644 (file)
@@ -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)