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):
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()
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()
@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
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)
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()
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)
# 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)