Loading system/gd/cert/cert_self_test.py +2 −2 Original line number Diff line number Diff line Loading @@ -19,7 +19,7 @@ import time from mobly import asserts from datetime import datetime, timedelta from cert.gd_base_test_facade_only import GdFacadeOnlyBaseTestClass from acts.base_test import BaseTestClass from cert.event_callback_stream import EventCallbackStream from cert.event_asserts import EventAsserts Loading Loading @@ -86,7 +86,7 @@ class FetchEvents: return None class CertSelfTest(GdFacadeOnlyBaseTestClass): class CertSelfTest(BaseTestClass): def setup_test(self): return True Loading system/gd/cert/gd_base_test_facade_only.py +23 −3 Original line number Diff line number Diff line Loading @@ -15,6 +15,7 @@ # limitations under the License. from acts.base_test import BaseTestClass from facade import rootservice_pb2 as facade_rootservice import importlib import logging Loading @@ -27,7 +28,9 @@ ANDROID_BUILD_TOP = os.environ.get('ANDROID_BUILD_TOP') class GdFacadeOnlyBaseTestClass(BaseTestClass): def setup_class(self): def setup_class(self, dut_module, cert_module): self.dut_module = dut_module self.cert_module = cert_module gd_devices = self.controller_configs.get("GdDevice") Loading Loading @@ -58,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: Loading @@ -71,3 +74,20 @@ class GdFacadeOnlyBaseTestClass(BaseTestClass): logging.error( "rootcanal stopped with code: %d" % rootcanal_return_code) return False def setup_test(self): self.dut.rootservice.StartStack( facade_rootservice.StartStackRequest( module_under_test=facade_rootservice.BluetoothModule.Value( self.dut_module),)) self.cert.rootservice.StartStack( facade_rootservice.StartStackRequest( module_under_test=facade_rootservice.BluetoothModule.Value( self.cert_module),)) self.dut.wait_channel_ready() self.cert.wait_channel_ready() def teardown_test(self): self.dut.rootservice.StopStack(facade_rootservice.StopStackRequest()) self.cert.rootservice.StopStack(facade_rootservice.StopStackRequest()) system/gd/hal/cert/simple_hal_test.py +29 −47 Original line number Diff line number Diff line Loading @@ -28,8 +28,17 @@ import bluetooth_packets_python3 as bt_packets class SimpleHalTest(GdFacadeOnlyBaseTestClass): def setup_class(self): super().setup_class(dut_module='HAL', cert_module='HAL') def setup_test(self): super().setup_test() self.send_dut_hci_command(hci_packets.ResetBuilder()) 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): Loading @@ -41,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): Loading @@ -57,41 +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 setup_test(self): self.device_under_test.rootservice.StartStack( facade_rootservice_pb2.StartStackRequest( module_under_test=facade_rootservice_pb2.BluetoothModule.Value( 'HAL'),)) self.cert_device.rootservice.StartStack( facade_rootservice_pb2.StartStackRequest( module_under_test=facade_rootservice_pb2.BluetoothModule.Value( 'HAL'),)) self.device_under_test.wait_channel_ready() self.cert_device.wait_channel_ready() self.send_dut_hci_command(hci_packets.ResetBuilder()) self.send_cert_hci_command(hci_packets.ResetBuilder()) def teardown_test(self): self.device_under_test.rootservice.StopStack( facade_rootservice_pb2.StopStackRequest()) self.cert_device.rootservice.StopStack( facade_rootservice_pb2.StopStackRequest()) 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) Loading @@ -106,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( Loading @@ -123,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( Loading @@ -140,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) Loading Loading @@ -219,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) Loading Loading @@ -341,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) Loading system/gd/hci/cert/acl_manager_test.py +22 −39 Original line number Diff line number Diff line Loading @@ -22,7 +22,6 @@ from cert.gd_base_test_facade_only import GdFacadeOnlyBaseTestClass from cert.event_callback_stream import EventCallbackStream from cert.event_asserts import EventAsserts from google.protobuf import empty_pb2 as empty_proto from facade import rootservice_pb2 as facade_rootservice from hci.facade import acl_manager_facade_pb2 as acl_manager_facade from neighbor.facade import facade_pb2 as neighbor_facade from hci.facade import controller_facade_pb2 as controller_facade Loading @@ -33,36 +32,20 @@ from bluetooth_packets_python3 import hci_packets class AclManagerTest(GdFacadeOnlyBaseTestClass): def setup_test(self): self.device_under_test.rootservice.StartStack( facade_rootservice.StartStackRequest( module_under_test=facade_rootservice.BluetoothModule.Value( 'HCI_INTERFACES'),)) self.cert_device.rootservice.StartStack( facade_rootservice.StartStackRequest( module_under_test=facade_rootservice.BluetoothModule.Value( 'HCI'),)) self.device_under_test.wait_channel_ready() self.cert_device.wait_channel_ready() def teardown_test(self): self.device_under_test.rootservice.StopStack( facade_rootservice.StopStackRequest()) self.cert_device.rootservice.StopStack( facade_rootservice.StopStackRequest()) def setup_class(self): super().setup_class(dut_module='HCI_INTERFACES', cert_module='HCI') 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( Loading @@ -70,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) Loading Loading @@ -111,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), Loading Loading @@ -169,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( Loading @@ -186,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( Loading @@ -198,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 Loading Loading @@ -232,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( Loading Loading @@ -261,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) Loading Loading @@ -294,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), Loading system/gd/hci/cert/controller_test.py +11 −26 Original line number Diff line number Diff line Loading @@ -25,29 +25,14 @@ from hci.facade import controller_facade_pb2 as controller_facade class ControllerTest(GdFacadeOnlyBaseTestClass): def setup_test(self): self.device_under_test.rootservice.StartStack( facade_rootservice.StartStackRequest( module_under_test=facade_rootservice.BluetoothModule.Value( 'HCI_INTERFACES'),)) self.cert_device.rootservice.StartStack( facade_rootservice.StartStackRequest( module_under_test=facade_rootservice.BluetoothModule.Value( 'HCI_INTERFACES'),)) self.device_under_test.wait_channel_ready() self.cert_device.wait_channel_ready() def teardown_test(self): self.device_under_test.rootservice.StopStack( facade_rootservice.StopStackRequest()) self.cert_device.rootservice.StopStack( facade_rootservice.StopStackRequest()) def setup_class(self): super().setup_class( 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, Loading @@ -58,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, Loading @@ -70,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', Loading Loading
system/gd/cert/cert_self_test.py +2 −2 Original line number Diff line number Diff line Loading @@ -19,7 +19,7 @@ import time from mobly import asserts from datetime import datetime, timedelta from cert.gd_base_test_facade_only import GdFacadeOnlyBaseTestClass from acts.base_test import BaseTestClass from cert.event_callback_stream import EventCallbackStream from cert.event_asserts import EventAsserts Loading Loading @@ -86,7 +86,7 @@ class FetchEvents: return None class CertSelfTest(GdFacadeOnlyBaseTestClass): class CertSelfTest(BaseTestClass): def setup_test(self): return True Loading
system/gd/cert/gd_base_test_facade_only.py +23 −3 Original line number Diff line number Diff line Loading @@ -15,6 +15,7 @@ # limitations under the License. from acts.base_test import BaseTestClass from facade import rootservice_pb2 as facade_rootservice import importlib import logging Loading @@ -27,7 +28,9 @@ ANDROID_BUILD_TOP = os.environ.get('ANDROID_BUILD_TOP') class GdFacadeOnlyBaseTestClass(BaseTestClass): def setup_class(self): def setup_class(self, dut_module, cert_module): self.dut_module = dut_module self.cert_module = cert_module gd_devices = self.controller_configs.get("GdDevice") Loading Loading @@ -58,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: Loading @@ -71,3 +74,20 @@ class GdFacadeOnlyBaseTestClass(BaseTestClass): logging.error( "rootcanal stopped with code: %d" % rootcanal_return_code) return False def setup_test(self): self.dut.rootservice.StartStack( facade_rootservice.StartStackRequest( module_under_test=facade_rootservice.BluetoothModule.Value( self.dut_module),)) self.cert.rootservice.StartStack( facade_rootservice.StartStackRequest( module_under_test=facade_rootservice.BluetoothModule.Value( self.cert_module),)) self.dut.wait_channel_ready() self.cert.wait_channel_ready() def teardown_test(self): self.dut.rootservice.StopStack(facade_rootservice.StopStackRequest()) self.cert.rootservice.StopStack(facade_rootservice.StopStackRequest())
system/gd/hal/cert/simple_hal_test.py +29 −47 Original line number Diff line number Diff line Loading @@ -28,8 +28,17 @@ import bluetooth_packets_python3 as bt_packets class SimpleHalTest(GdFacadeOnlyBaseTestClass): def setup_class(self): super().setup_class(dut_module='HAL', cert_module='HAL') def setup_test(self): super().setup_test() self.send_dut_hci_command(hci_packets.ResetBuilder()) 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): Loading @@ -41,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): Loading @@ -57,41 +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 setup_test(self): self.device_under_test.rootservice.StartStack( facade_rootservice_pb2.StartStackRequest( module_under_test=facade_rootservice_pb2.BluetoothModule.Value( 'HAL'),)) self.cert_device.rootservice.StartStack( facade_rootservice_pb2.StartStackRequest( module_under_test=facade_rootservice_pb2.BluetoothModule.Value( 'HAL'),)) self.device_under_test.wait_channel_ready() self.cert_device.wait_channel_ready() self.send_dut_hci_command(hci_packets.ResetBuilder()) self.send_cert_hci_command(hci_packets.ResetBuilder()) def teardown_test(self): self.device_under_test.rootservice.StopStack( facade_rootservice_pb2.StopStackRequest()) self.cert_device.rootservice.StopStack( facade_rootservice_pb2.StopStackRequest()) 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) Loading @@ -106,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( Loading @@ -123,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( Loading @@ -140,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) Loading Loading @@ -219,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) Loading Loading @@ -341,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) Loading
system/gd/hci/cert/acl_manager_test.py +22 −39 Original line number Diff line number Diff line Loading @@ -22,7 +22,6 @@ from cert.gd_base_test_facade_only import GdFacadeOnlyBaseTestClass from cert.event_callback_stream import EventCallbackStream from cert.event_asserts import EventAsserts from google.protobuf import empty_pb2 as empty_proto from facade import rootservice_pb2 as facade_rootservice from hci.facade import acl_manager_facade_pb2 as acl_manager_facade from neighbor.facade import facade_pb2 as neighbor_facade from hci.facade import controller_facade_pb2 as controller_facade Loading @@ -33,36 +32,20 @@ from bluetooth_packets_python3 import hci_packets class AclManagerTest(GdFacadeOnlyBaseTestClass): def setup_test(self): self.device_under_test.rootservice.StartStack( facade_rootservice.StartStackRequest( module_under_test=facade_rootservice.BluetoothModule.Value( 'HCI_INTERFACES'),)) self.cert_device.rootservice.StartStack( facade_rootservice.StartStackRequest( module_under_test=facade_rootservice.BluetoothModule.Value( 'HCI'),)) self.device_under_test.wait_channel_ready() self.cert_device.wait_channel_ready() def teardown_test(self): self.device_under_test.rootservice.StopStack( facade_rootservice.StopStackRequest()) self.cert_device.rootservice.StopStack( facade_rootservice.StopStackRequest()) def setup_class(self): super().setup_class(dut_module='HCI_INTERFACES', cert_module='HCI') 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( Loading @@ -70,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) Loading Loading @@ -111,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), Loading Loading @@ -169,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( Loading @@ -186,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( Loading @@ -198,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 Loading Loading @@ -232,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( Loading Loading @@ -261,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) Loading Loading @@ -294,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), Loading
system/gd/hci/cert/controller_test.py +11 −26 Original line number Diff line number Diff line Loading @@ -25,29 +25,14 @@ from hci.facade import controller_facade_pb2 as controller_facade class ControllerTest(GdFacadeOnlyBaseTestClass): def setup_test(self): self.device_under_test.rootservice.StartStack( facade_rootservice.StartStackRequest( module_under_test=facade_rootservice.BluetoothModule.Value( 'HCI_INTERFACES'),)) self.cert_device.rootservice.StartStack( facade_rootservice.StartStackRequest( module_under_test=facade_rootservice.BluetoothModule.Value( 'HCI_INTERFACES'),)) self.device_under_test.wait_channel_ready() self.cert_device.wait_channel_ready() def teardown_test(self): self.device_under_test.rootservice.StopStack( facade_rootservice.StopStackRequest()) self.cert_device.rootservice.StopStack( facade_rootservice.StopStackRequest()) def setup_class(self): super().setup_class( 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, Loading @@ -58,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, Loading @@ -70,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', Loading