Rename device_under_test and cert_device, to simpler more consistent names (a3362b74) · Commits · e / os / android_packages_modules_Bluetooth

system/gd/cert/gd_base_test_facade_only.py

+8 −10

Original line number	Diff line number	Diff line
		@@ -61,8 +61,8 @@ class GdFacadeOnlyBaseTestClass(BaseTestClass):
		self.register_controller(
		importlib.import_module('cert.gd_device'), builtin=True)

		self.device_under_test = self.gd_devices[1]
		self.cert_device = self.gd_devices[0]
		self.dut = self.gd_devices[1]
		self.cert = self.gd_devices[0]

		def teardown_class(self):
		if self.rootcanal_running:
		@@ -76,20 +76,18 @@ class GdFacadeOnlyBaseTestClass(BaseTestClass):
		return False

		def setup_test(self):
		self.device_under_test.rootservice.StartStack(
		self.dut.rootservice.StartStack(
		facade_rootservice.StartStackRequest(
		module_under_test=facade_rootservice.BluetoothModule.Value(
		self.dut_module),))
		self.cert_device.rootservice.StartStack(
		self.cert.rootservice.StartStack(
		facade_rootservice.StartStackRequest(
		module_under_test=facade_rootservice.BluetoothModule.Value(
		self.cert_module),))

		self.device_under_test.wait_channel_ready()
		self.cert_device.wait_channel_ready()
		self.dut.wait_channel_ready()
		self.cert.wait_channel_ready()

		def teardown_test(self):
		self.device_under_test.rootservice.StopStack(
		facade_rootservice.StopStackRequest())
		self.cert_device.rootservice.StopStack(
		facade_rootservice.StopStackRequest())
		self.dut.rootservice.StopStack(facade_rootservice.StopStackRequest())
		self.cert.rootservice.StopStack(facade_rootservice.StopStackRequest())

system/gd/hal/cert/simple_hal_test.py

+20 −25

Original line number	Diff line number	Diff line
		@@ -38,7 +38,7 @@ class SimpleHalTest(GdFacadeOnlyBaseTestClass):
		self.send_cert_hci_command(hci_packets.ResetBuilder())

		def send_cert_hci_command(self, command):
		self.cert_device.hal.SendHciCommand(
		self.cert.hal.SendHciCommand(
		hal_facade_pb2.HciCommandPacket(payload=bytes(command.Serialize())))

		def send_cert_acl_data(self, handle, pb_flag, b_flag, acl):
		@@ -50,11 +50,11 @@ class SimpleHalTest(GdFacadeOnlyBaseTestClass):
		upper_length = (len(acl) & 0xff00) >> 8
		concatenated = bytes([lower, upper, lower_length, upper_length] +
		list(acl))
		self.cert_device.hal.SendHciAcl(
		self.cert.hal.SendHciAcl(
		hal_facade_pb2.HciAclPacket(payload=concatenated))

		def send_dut_hci_command(self, command):
		self.device_under_test.hal.SendHciCommand(
		self.dut.hal.SendHciCommand(
		hal_facade_pb2.HciCommandPacket(payload=bytes(command.Serialize())))

		def send_dut_acl_data(self, handle, pb_flag, b_flag, acl):
		@@ -66,19 +66,17 @@ class SimpleHalTest(GdFacadeOnlyBaseTestClass):
		upper_length = (len(acl) & 0xff00) >> 8
		concatenated = bytes([lower, upper, lower_length, upper_length] +
		list(acl))
		self.device_under_test.hal.SendHciAcl(
		self.dut.hal.SendHciAcl(
		hal_facade_pb2.HciAclPacket(payload=concatenated))

		def test_none_event(self):
		with EventCallbackStream(
		self.device_under_test.hal.FetchHciEvent(
		with EventCallbackStream(self.dut.hal.FetchHciEvent(
		empty_pb2.Empty())) as hci_event_stream:
		hci_event_asserts = EventAsserts(hci_event_stream)
		hci_event_asserts.assert_none(timeout=timedelta(seconds=1))

		def test_fetch_hci_event(self):
		with EventCallbackStream(
		self.device_under_test.hal.FetchHciEvent(
		with EventCallbackStream(self.dut.hal.FetchHciEvent(
		empty_pb2.Empty())) as hci_event_stream:

		hci_event_asserts = EventAsserts(hci_event_stream)
		@@ -93,8 +91,7 @@ class SimpleHalTest(GdFacadeOnlyBaseTestClass):
		lambda packet: bytes(event.Serialize()) in packet.payload)

		def test_loopback_hci_command(self):
		with EventCallbackStream(
		self.device_under_test.hal.FetchHciEvent(
		with EventCallbackStream(self.dut.hal.FetchHciEvent(
		empty_pb2.Empty())) as hci_event_stream:

		self.send_dut_hci_command(
		@@ -110,8 +107,7 @@ class SimpleHalTest(GdFacadeOnlyBaseTestClass):
		lambda packet: bytes(command.Serialize()) in packet.payload)

		def test_inquiry_from_dut(self):
		with EventCallbackStream(
		self.device_under_test.hal.FetchHciEvent(
		with EventCallbackStream(self.dut.hal.FetchHciEvent(
		empty_pb2.Empty())) as hci_event_stream:
		hci_event_asserts = EventAsserts(hci_event_stream)
		self.send_cert_hci_command(
		@@ -127,8 +123,7 @@ class SimpleHalTest(GdFacadeOnlyBaseTestClass):
		)

		def test_le_ad_scan_cert_advertises(self):
		with EventCallbackStream(
		self.device_under_test.hal.FetchHciEvent(
		with EventCallbackStream(self.dut.hal.FetchHciEvent(
		empty_pb2.Empty())) as hci_event_stream:
		hci_event_asserts = EventAsserts(hci_event_stream)

		@@ -206,10 +201,10 @@ class SimpleHalTest(GdFacadeOnlyBaseTestClass):
		hci_packets.FilterDuplicates.DISABLED, 0, 0))

		def test_le_connection_dut_advertises(self):
		with EventCallbackStream(self.device_under_test.hal.FetchHciEvent(empty_pb2.Empty())) as hci_event_stream, \
		EventCallbackStream(self.cert_device.hal.FetchHciEvent(empty_pb2.Empty())) as cert_hci_event_stream, \
		EventCallbackStream(self.device_under_test.hal.FetchHciAcl(empty_pb2.Empty())) as acl_data_stream, \
		EventCallbackStream(self.cert_device.hal.FetchHciAcl(empty_pb2.Empty())) as cert_acl_data_stream:
		with EventCallbackStream(self.dut.hal.FetchHciEvent(empty_pb2.Empty())) as hci_event_stream, \
		EventCallbackStream(self.cert.hal.FetchHciEvent(empty_pb2.Empty())) as cert_hci_event_stream, \
		EventCallbackStream(self.dut.hal.FetchHciAcl(empty_pb2.Empty())) as acl_data_stream, \
		EventCallbackStream(self.cert.hal.FetchHciAcl(empty_pb2.Empty())) as cert_acl_data_stream:

		hci_event_asserts = EventAsserts(hci_event_stream)
		cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)
		@@ -328,8 +323,8 @@ class SimpleHalTest(GdFacadeOnlyBaseTestClass):
		lambda packet: b'SomeMoreAclData' in packet.payload)

		def test_le_white_list_connection_cert_advertises(self):
		with EventCallbackStream(self.device_under_test.hal.FetchHciEvent(empty_pb2.Empty())) as hci_event_stream, \
		EventCallbackStream(self.cert_device.hal.FetchHciEvent(empty_pb2.Empty())) as cert_hci_event_stream:
		with EventCallbackStream(self.dut.hal.FetchHciEvent(empty_pb2.Empty())) as hci_event_stream, \
		EventCallbackStream(self.cert.hal.FetchHciEvent(empty_pb2.Empty())) as cert_hci_event_stream:
		hci_event_asserts = EventAsserts(hci_event_stream)
		cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)

system/gd/hci/cert/acl_manager_test.py

+20 −20

Original line number	Diff line number	Diff line
		@@ -37,15 +37,15 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):

		def register_for_event(self, event_code):
		msg = hci_facade.EventCodeMsg(code=int(event_code))
		self.cert_device.hci.RegisterEventHandler(msg)
		self.cert.hci.RegisterEventHandler(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.cert_device.hci.EnqueueCommandWithComplete(cmd)
		self.cert.hci.EnqueueCommandWithComplete(cmd)
		else:
		self.cert_device.hci.EnqueueCommandWithStatus(cmd)
		self.cert.hci.EnqueueCommandWithStatus(cmd)

		def enqueue_acl_data(self, handle, pb_flag, b_flag, acl):
		acl_msg = hci_facade.AclMsg(
		@@ -53,16 +53,16 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
		packet_boundary_flag=int(pb_flag),
		broadcast_flag=int(b_flag),
		data=acl)
		self.cert_device.hci.SendAclData(acl_msg)
		self.cert.hci.SendAclData(acl_msg)

		def test_dut_connects(self):
		self.register_for_event(hci_packets.EventCode.CONNECTION_REQUEST)
		self.register_for_event(hci_packets.EventCode.CONNECTION_COMPLETE)
		self.register_for_event(
		hci_packets.EventCode.CONNECTION_PACKET_TYPE_CHANGED)
		with EventCallbackStream(self.cert_device.hci.FetchEvents(empty_proto.Empty())) as cert_hci_event_stream, \
		EventCallbackStream(self.cert_device.hci.FetchAclPackets(empty_proto.Empty())) as cert_acl_data_stream, \
		EventCallbackStream(self.device_under_test.hci_acl_manager.FetchAclData(empty_proto.Empty())) as acl_data_stream:
		with EventCallbackStream(self.cert.hci.FetchEvents(empty_proto.Empty())) as cert_hci_event_stream, \
		EventCallbackStream(self.cert.hci.FetchAclPackets(empty_proto.Empty())) as cert_acl_data_stream, \
		EventCallbackStream(self.dut.hci_acl_manager.FetchAclData(empty_proto.Empty())) as acl_data_stream:

		cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)
		acl_data_asserts = EventAsserts(acl_data_stream)
		@@ -94,7 +94,7 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
		get_address_from_complete)

		with EventCallbackStream(
		self.device_under_test.hci_acl_manager.CreateConnection(
		self.dut.hci_acl_manager.CreateConnection(
		acl_manager_facade.ConnectionMsg(
		address_type=int(
		hci_packets.AddressType.PUBLIC_DEVICE_ADDRESS),
		@@ -152,7 +152,7 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
		handle = 0xfff
		connection_event_asserts.assert_event_occurs(get_handle)

		self.device_under_test.hci_acl_manager.SendAclData(
		self.dut.hci_acl_manager.SendAclData(
		acl_manager_facade.AclData(
		handle=handle,
		payload=bytes(
		@@ -169,10 +169,10 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
		self.register_for_event(hci_packets.EventCode.ROLE_CHANGE)
		self.register_for_event(
		hci_packets.EventCode.CONNECTION_PACKET_TYPE_CHANGED)
		with EventCallbackStream(self.cert_device.hci.FetchEvents(empty_proto.Empty())) as cert_hci_event_stream, \
		EventCallbackStream(self.cert_device.hci.FetchAclPackets(empty_proto.Empty())) as cert_acl_data_stream, \
		EventCallbackStream(self.device_under_test.hci_acl_manager.FetchIncomingConnection(empty_proto.Empty())) as incoming_connection_stream, \
		EventCallbackStream(self.device_under_test.hci_acl_manager.FetchAclData(empty_proto.Empty())) as acl_data_stream:
		with EventCallbackStream(self.cert.hci.FetchEvents(empty_proto.Empty())) as cert_hci_event_stream, \
		EventCallbackStream(self.cert.hci.FetchAclPackets(empty_proto.Empty())) as cert_acl_data_stream, \
		EventCallbackStream(self.dut.hci_acl_manager.FetchIncomingConnection(empty_proto.Empty())) as incoming_connection_stream, \
		EventCallbackStream(self.dut.hci_acl_manager.FetchAclData(empty_proto.Empty())) as acl_data_stream:

		cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)
		incoming_connection_asserts = EventAsserts(
		@@ -181,10 +181,10 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
		acl_data_asserts = EventAsserts(acl_data_stream)

		# DUT Enables scans and gets its address
		dut_address = self.device_under_test.hci_controller.GetMacAddress(
		dut_address = self.dut.hci_controller.GetMacAddress(
		empty_proto.Empty()).address

		self.device_under_test.neighbor.EnablePageScan(
		self.dut.neighbor.EnablePageScan(
		neighbor_facade.EnableMsg(enabled=True))

		# Cert connects
		@@ -215,7 +215,7 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
		# DUT gets a connection request
		incoming_connection_asserts.assert_event_occurs(get_handle)

		self.device_under_test.hci_acl_manager.SendAclData(
		self.dut.hci_acl_manager.SendAclData(
		acl_manager_facade.AclData(
		handle=conn_handle,
		payload=bytes(
		@@ -244,9 +244,9 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
		self.register_for_event(hci_packets.EventCode.CONNECTION_COMPLETE)
		self.register_for_event(
		hci_packets.EventCode.CONNECTION_PACKET_TYPE_CHANGED)
		with EventCallbackStream(self.cert_device.hci.FetchEvents(empty_proto.Empty())) as cert_hci_event_stream, \
		EventCallbackStream(self.cert_device.hci.FetchAclPackets(empty_proto.Empty())) as cert_acl_data_stream, \
		EventCallbackStream(self.device_under_test.hci_acl_manager.FetchAclData(empty_proto.Empty())) as acl_data_stream:
		with EventCallbackStream(self.cert.hci.FetchEvents(empty_proto.Empty())) as cert_hci_event_stream, \
		EventCallbackStream(self.cert.hci.FetchAclPackets(empty_proto.Empty())) as cert_acl_data_stream, \
		EventCallbackStream(self.dut.hci_acl_manager.FetchAclData(empty_proto.Empty())) as acl_data_stream:

		cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)
		acl_data_asserts = EventAsserts(acl_data_stream)
		@@ -277,7 +277,7 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
		get_address_from_complete)

		with EventCallbackStream(
		self.device_under_test.hci_acl_manager.CreateConnection(
		self.dut.hci_acl_manager.CreateConnection(
		acl_manager_facade.ConnectionMsg(
		address_type=int(
		hci_packets.AddressType.PUBLIC_DEVICE_ADDRESS),

system/gd/hci/cert/controller_test.py

+8 −8

Original line number	Diff line number	Diff line
		@@ -30,9 +30,9 @@ class ControllerTest(GdFacadeOnlyBaseTestClass):
		dut_module='HCI_INTERFACES', cert_module='HCI_INTERFACES')

		def test_get_addresses(self):
		cert_address_response = self.cert_device.hci_controller.GetMacAddress(
		cert_address_response = self.cert.hci_controller.GetMacAddress(
		empty_proto.Empty())
		dut_address_response = self.device_under_test.hci_controller.GetMacAddress(
		dut_address_response = self.dut.hci_controller.GetMacAddress(
		empty_proto.Empty())
		asserts.assert_true(
		cert_address_response.address != dut_address_response.address,
		@@ -43,11 +43,11 @@ class ControllerTest(GdFacadeOnlyBaseTestClass):
		def test_get_local_extended_features(self):
		request = controller_facade.PageNumberMsg()
		request.page_number = 1
		dut_feature_response1 = self.device_under_test.hci_controller.GetLocalExtendedFeatures(
		dut_feature_response1 = self.dut.hci_controller.GetLocalExtendedFeatures(
		request)
		request0 = controller_facade.PageNumberMsg()
		request0.page_number = 0
		dut_feature_response0 = self.device_under_test.hci_controller.GetLocalExtendedFeatures(
		dut_feature_response0 = self.dut.hci_controller.GetLocalExtendedFeatures(
		request0)
		asserts.assert_true(
		dut_feature_response1.page != dut_feature_response0.page,
		@@ -55,13 +55,13 @@ class ControllerTest(GdFacadeOnlyBaseTestClass):
		dut_feature_response1.page)

		def test_write_local_name(self):
		self.device_under_test.hci_controller.WriteLocalName(
		self.dut.hci_controller.WriteLocalName(
		controller_facade.NameMsg(name=b'ImTheDUT'))
		self.cert_device.hci_controller.WriteLocalName(
		self.cert.hci_controller.WriteLocalName(
		controller_facade.NameMsg(name=b'ImTheCert'))
		cert_name_msg = self.cert_device.hci_controller.GetLocalName(
		cert_name_msg = self.cert.hci_controller.GetLocalName(
		empty_proto.Empty()).name
		dut_name_msg = self.device_under_test.hci_controller.GetLocalName(
		dut_name_msg = self.dut.hci_controller.GetLocalName(
		empty_proto.Empty()).name
		asserts.assert_true(
		dut_name_msg == b'ImTheDUT',

system/gd/hci/cert/direct_hci_test.py

+28 −31

Original line number	Diff line number	Diff line
		@@ -38,28 +38,28 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass):
		def setup_test(self):
		super().setup_test()

		self.cert_device.hal.SendHciCommand(
		self.cert.hal.SendHciCommand(
		hal_facade.HciCommandPacket(
		payload=bytes(hci_packets.ResetBuilder().Serialize())))

		def register_for_event(self, event_code):
		msg = hci_facade.EventCodeMsg(code=int(event_code))
		self.device_under_test.hci.RegisterEventHandler(msg)
		self.dut.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)
		self.dut.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)
		self.dut.hci.EnqueueCommandWithComplete(cmd)
		else:
		self.device_under_test.hci.EnqueueCommandWithStatus(cmd)
		self.dut.hci.EnqueueCommandWithStatus(cmd)

		def send_hal_hci_command(self, command):
		self.cert_device.hal.SendHciCommand(
		self.cert.hal.SendHciCommand(
		hal_facade.HciCommandPacket(payload=bytes(command.Serialize())))

		def enqueue_acl_data(self, handle, pb_flag, b_flag, acl):
		@@ -68,7 +68,7 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass):
		packet_boundary_flag=int(pb_flag),
		broadcast_flag=int(b_flag),
		data=acl)
		self.device_under_test.hci.SendAclData(acl_msg)
		self.dut.hci.SendAclData(acl_msg)

		def send_hal_acl_data(self, handle, pb_flag, b_flag, acl):
		lower = handle & 0xff
		@@ -79,8 +79,7 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass):
		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))
		self.cert.hal.SendHciAcl(hal_facade.HciAclPacket(payload=concatenated))

		def test_local_hci_cmd_and_event(self):
		# Loopback mode responds with ACL and SCO connection complete
		@@ -88,8 +87,7 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass):
		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(
		with EventCallbackStream(self.dut.hci.FetchEvents(
		empty_proto.Empty())) as hci_event_stream:
		hci_event_asserts = EventAsserts(hci_event_stream)

		@@ -106,8 +104,7 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass):

		def test_inquiry_from_dut(self):
		self.register_for_event(hci_packets.EventCode.INQUIRY_RESULT)
		with EventCallbackStream(
		self.device_under_test.hci.FetchEvents(
		with EventCallbackStream(self.dut.hci.FetchEvents(
		empty_proto.Empty())) as hci_event_stream:

		hci_event_asserts = EventAsserts(hci_event_stream)
		@@ -129,7 +126,7 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass):
		hci_packets.SubeventCode.EXTENDED_ADVERTISING_REPORT)
		self.register_for_le_event(hci_packets.SubeventCode.ADVERTISING_REPORT)
		with EventCallbackStream(
		self.device_under_test.hci.FetchLeSubevents(
		self.dut.hci.FetchLeSubevents(
		empty_proto.Empty())) as hci_le_event_stream:

		hci_event_asserts = EventAsserts(hci_le_event_stream)
		@@ -218,11 +215,11 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass):

		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:
		with EventCallbackStream(self.dut.hci.FetchLeSubevents(empty_proto.Empty())) as le_event_stream, \
		EventCallbackStream(self.dut.hci.FetchEvents(empty_proto.Empty())) as event_stream, \
		EventCallbackStream(self.dut.hci.FetchAclPackets(empty_proto.Empty())) as acl_data_stream, \
		EventCallbackStream(self.cert.hal.FetchHciEvent(empty_proto.Empty())) as cert_hci_event_stream, \
		EventCallbackStream(self.cert.hal.FetchHciAcl(empty_proto.Empty())) as cert_acl_data_stream:

		le_event_asserts = EventAsserts(le_event_stream)
		event_asserts = EventAsserts(event_stream)
		@@ -369,8 +366,8 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass):

		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:
		with EventCallbackStream(self.dut.hci.FetchLeSubevents(empty_proto.Empty())) as le_event_stream, \
		EventCallbackStream(self.cert.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)

		@@ -452,10 +449,10 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass):
		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:
		with EventCallbackStream(self.dut.hci.FetchEvents(empty_proto.Empty())) as hci_event_stream, \
		EventCallbackStream(self.dut.hci.FetchAclPackets(empty_proto.Empty())) as acl_data_stream, \
		EventCallbackStream(self.cert.hal.FetchHciEvent(empty_proto.Empty())) as cert_hci_event_stream, \
		EventCallbackStream(self.cert.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)
		@@ -574,10 +571,10 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass):
		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:
		with EventCallbackStream(self.dut.hci.FetchEvents(empty_proto.Empty())) as hci_event_stream, \
		EventCallbackStream(self.dut.hci.FetchAclPackets(empty_proto.Empty())) as acl_data_stream, \
		EventCallbackStream(self.cert.hal.FetchHciEvent(empty_proto.Empty())) as cert_hci_event_stream, \
		EventCallbackStream(self.cert.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)