from hal import facade_pb2 as hal_facade
from bluetooth_packets_python3.hci_packets import WriteScanEnableBuilder
from bluetooth_packets_python3.hci_packets import ScanEnable
+from bluetooth_packets_python3.hci_packets import AclPacketBuilder
+from bluetooth_packets_python3 import RawBuilder
class PyHal(Closable):
def send_hci_command(self, command):
self.device.hal.SendCommand(hal_facade.Command(payload=bytes(command.Serialize())))
- def send_acl(self, acl):
- self.device.hal.SendAcl(hal_facade.AclPacket(payload=bytes(acl)))
+ def send_acl(self, handle, pb_flag, b_flag, data):
+ acl = AclPacketBuilder(handle, pb_flag, b_flag, RawBuilder(data))
+ self.device.hal.SendAcl(hal_facade.AclPacket(payload=bytes(acl.Serialize())))
def enable_inquiry_and_page_scan(self):
self.send_hci_command(WriteScanEnableBuilder(ScanEnable.INQUIRY_AND_PAGE_SCAN))
self.cert_hal.close()
super().teardown_test()
- def send_cert_acl_data(self, handle, pb_flag, b_flag, acl):
- acl_msg = AclPacketBuilder(handle, pb_flag, b_flag, RawBuilder(acl))
- self.cert_hal.send_acl(acl_msg.Serialize())
-
- def send_dut_acl_data(self, handle, pb_flag, b_flag, acl):
- acl_msg = AclPacketBuilder(handle, pb_flag, b_flag, RawBuilder(acl))
- self.dut_hal.send_acl(acl_msg.Serialize())
-
def test_fetch_hci_event(self):
self.dut_hal.send_hci_command(
hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM, '0C:05:04:03:02:01'))
dut_handle = conn_handle
# Send ACL Data
- self.send_dut_acl_data(dut_handle, hci_packets.PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE,
- hci_packets.BroadcastFlag.POINT_TO_POINT, bytes(b'Just SomeAclData'))
- self.send_cert_acl_data(cert_handle, hci_packets.PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE,
- hci_packets.BroadcastFlag.POINT_TO_POINT, bytes(b'Just SomeMoreAclData'))
+ self.dut_hal.send_acl(dut_handle, hci_packets.PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE,
+ hci_packets.BroadcastFlag.POINT_TO_POINT, bytes(b'Just SomeAclData'))
+ self.cert_hal.send_acl(cert_handle, hci_packets.PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE,
+ hci_packets.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_hal.get_acl_stream()).emits(lambda packet: b'SomeMoreAclData' in packet.payload)
handle=int(handle), packet_boundary_flag=int(pb_flag), broadcast_flag=int(b_flag), data=acl)
self.dut.hci.SendAcl(acl_msg)
- def send_hal_acl_data(self, handle, pb_flag, b_flag, acl):
- acl_msg = AclPacketBuilder(handle, pb_flag, b_flag, RawBuilder(acl))
- self.cert_hal.send_acl(acl_msg.Serialize())
-
def test_local_hci_cmd_and_event(self):
# Loopback mode responds with ACL and SCO connection complete
self.dut_hci.register_for_events(EventCode.LOOPBACK_COMMAND)
# Send ACL Data
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,
+ self.cert_hal.send_acl(cert_handle, PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE,
BroadcastFlag.POINT_TO_POINT, bytes(b'Just SomeMoreAclData'))
assertThat(self.cert_hal.get_acl_stream()).emits(
# Send ACL Data
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,
+ self.cert_hal.send_acl(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)
# Send ACL Data
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,
+ self.cert_hal.send_acl(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)