Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit 771aa51b authored by Zach Johnson's avatar Zach Johnson Committed by Automerger Merge Worker
Browse files

Merge "remove invalid test, restucture simple_hal_test" am: a6216543 am: 9e016426 

Original change: https://android-review.googlesource.com/c/platform/system/bt/+/1497896

Change-Id: Ie4d83d43531e9082b1336182a097f9261d3934f5
parents 80bfa7aa 9e016426

system/gd/cert/py_hal.py

+1 −1
Original line number Diff line number Diff line
@@ -42,7 +42,7 @@ class PyHal(Closable): 
        return self.acl_stream



    def send_hci_command(self, command):

        self.device.hal.SendCommand(hal_facade.Command(payload=bytes(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)))

system/gd/hal/cert/simple_hal_test.py

+238 −257
Original line number Diff line number Diff line
@@ -19,6 +19,7 @@ from datetime import timedelta 
from cert.gd_base_test import GdBaseTestClass

from cert.event_stream import EventStream

from cert.truth import assertThat

from cert.py_hal import PyHal

from google.protobuf import empty_pb2

from facade import rootservice_pb2 as facade_rootservice_pb2

from hal import facade_pb2 as hal_facade_pb2
@@ -36,11 +37,16 @@ class SimpleHalTest(GdBaseTestClass): 
    def setup_test(self):

        super().setup_test()



        self.send_dut_hci_command(hci_packets.ResetBuilder())

        self.send_cert_hci_command(hci_packets.ResetBuilder())

        self.dut_hal = PyHal(self.dut)

        self.cert_hal = PyHal(self.cert)



    def send_cert_hci_command(self, command):

        self.cert.hal.SendCommand(hal_facade_pb2.Command(payload=bytes(command.Serialize())), timeout=_GRPC_TIMEOUT)

        self.dut_hal.send_hci_command(hci_packets.ResetBuilder())

        self.cert_hal.send_hci_command(hci_packets.ResetBuilder())



    def teardown_test(self):

        self.dut_hal.close()

        self.cert_hal.close()

        super().teardown_test()



    def send_cert_acl_data(self, handle, pb_flag, b_flag, acl):

        lower = handle & 0xff
@@ -50,10 +56,7 @@ class SimpleHalTest(GdBaseTestClass): 
        lower_length = len(acl) & 0xff

        upper_length = (len(acl) & 0xff00) >> 8

        concatenated = bytes([lower, upper, lower_length, upper_length] + list(acl))

        self.cert.hal.SendAcl(hal_facade_pb2.AclPacket(payload=concatenated))



    def send_dut_hci_command(self, command):

        self.dut.hal.SendCommand(hal_facade_pb2.Command(payload=bytes(command.Serialize())), timeout=_GRPC_TIMEOUT)

        self.cert_hal.send_acl(concatenated)



    def send_dut_acl_data(self, handle, pb_flag, b_flag, acl):

        lower = handle & 0xff
@@ -63,66 +66,55 @@ class SimpleHalTest(GdBaseTestClass): 
        lower_length = len(acl) & 0xff

        upper_length = (len(acl) & 0xff00) >> 8

        concatenated = bytes([lower, upper, lower_length, upper_length] + list(acl))

        self.dut.hal.SendAcl(hal_facade_pb2.AclPacket(payload=concatenated), timeout=_GRPC_TIMEOUT)



    def test_none_event(self):

        with EventStream(self.dut.hal.StreamEvents(empty_pb2.Empty())) as hci_event_stream:

            assertThat(hci_event_stream).emitsNone(timeout=timedelta(seconds=1))

        self.dut_hal.send_acl(concatenated)



    def test_fetch_hci_event(self):

        with EventStream(self.dut.hal.StreamEvents(empty_pb2.Empty())) as hci_event_stream:



            self.send_dut_hci_command(

                hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM,

                                                            '0C:05:04:03:02:01'))

        self.dut_hal.send_hci_command(

            hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM, '0C:05:04:03:02:01'))

        event = hci_packets.LeAddDeviceToConnectListCompleteBuilder(1, hci_packets.ErrorCode.SUCCESS)



            assertThat(hci_event_stream).emits(lambda packet: bytes(event.Serialize()) in packet.payload)

        assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: bytes(event.Serialize()) in packet.payload)



    def test_loopback_hci_command(self):

        with EventStream(self.dut.hal.StreamEvents(empty_pb2.Empty())) as hci_event_stream:



            self.send_dut_hci_command(hci_packets.WriteLoopbackModeBuilder(hci_packets.LoopbackMode.ENABLE_LOCAL))

        self.dut_hal.send_hci_command(hci_packets.WriteLoopbackModeBuilder(hci_packets.LoopbackMode.ENABLE_LOCAL))



        command = hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM,

                                                              '0C:05:04:03:02:01')

            self.send_dut_hci_command(command)

        self.dut_hal.send_hci_command(command)



            assertThat(hci_event_stream).emits(lambda packet: bytes(command.Serialize()) in packet.payload)

        assertThat(

            self.dut_hal.get_hci_event_stream()).emits(lambda packet: bytes(command.Serialize()) in packet.payload)



    def test_inquiry_from_dut(self):

        with EventStream(self.dut.hal.StreamEvents(empty_pb2.Empty())) as hci_event_stream:

        self.cert_hal.send_hci_command(hci_packets.WriteScanEnableBuilder(hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN))



            self.send_cert_hci_command(hci_packets.WriteScanEnableBuilder(hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN))

        lap = hci_packets.Lap()

        lap.lap = 0x33

            self.send_dut_hci_command(hci_packets.InquiryBuilder(lap, 0x30, 0xff))

        self.dut_hal.send_hci_command(hci_packets.InquiryBuilder(lap, 0x30, 0xff))



            assertThat(hci_event_stream).emits(lambda packet: b'\x02\x0f' in packet.payload

        assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: b'\x02\x0f' in packet.payload

                                                              # Expecting an HCI Event (code 0x02, length 0x0f)

                                                             )



    def test_le_ad_scan_cert_advertises(self):

        with EventStream(self.dut.hal.StreamEvents(empty_pb2.Empty())) as hci_event_stream:



        # DUT scans

            self.send_dut_hci_command(hci_packets.LeSetRandomAddressBuilder('0D:05:04:03:02:01'))

        self.dut_hal.send_hci_command(hci_packets.LeSetRandomAddressBuilder('0D:05:04:03:02:01'))

        phy_scan_params = hci_packets.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



            self.send_dut_hci_command(

        self.dut_hal.send_hci_command(

            hci_packets.LeSetExtendedScanParametersBuilder(hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS,

                                                           hci_packets.LeScanningFilterPolicy.ACCEPT_ALL, 1,

                                                           [phy_scan_params]))

            self.send_dut_hci_command(

        self.dut_hal.send_hci_command(

            hci_packets.LeSetExtendedScanEnableBuilder(hci_packets.Enable.ENABLED,

                                                       hci_packets.FilterDuplicates.DISABLED, 0, 0))



        # CERT Advertises

        advertising_handle = 0

            self.send_cert_hci_command(

        self.cert_hal.send_hci_command(

            hci_packets.LeSetExtendedAdvertisingLegacyParametersBuilder(

                advertising_handle,

                hci_packets.LegacyAdvertisingProperties.ADV_IND,
@@ -138,14 +130,14 @@ class SimpleHalTest(GdBaseTestClass): 
                hci_packets.Enable.DISABLED  # Scan request notification

            ))



            self.send_cert_hci_command(

        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

        gap_name.data = list(bytes(b'Im_A_Cert'))



            self.send_cert_hci_command(

        self.cert_hal.send_hci_command(

            hci_packets.LeSetExtendedAdvertisingDataBuilder(

                advertising_handle, hci_packets.Operation.COMPLETE_ADVERTISEMENT,

                hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_name]))
@@ -153,28 +145,23 @@ class SimpleHalTest(GdBaseTestClass): 
        enabled_set.advertising_handle = advertising_handle

        enabled_set.duration = 0

        enabled_set.max_extended_advertising_events = 0

            self.send_cert_hci_command(

        self.cert_hal.send_hci_command(

            hci_packets.LeSetExtendedAdvertisingEnableBuilder(hci_packets.Enable.ENABLED, [enabled_set]))



            assertThat(hci_event_stream).emits(lambda packet: b'Im_A_Cert' in packet.payload)

        assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: b'Im_A_Cert' in packet.payload)



        # Disable Advertising

            self.send_cert_hci_command(

        self.cert_hal.send_hci_command(

            hci_packets.LeSetExtendedAdvertisingEnableBuilder(hci_packets.Enable.DISABLED, [enabled_set]))



        # Disable Scanning

            self.send_dut_hci_command(

        self.dut_hal.send_hci_command(

            hci_packets.LeSetExtendedScanEnableBuilder(hci_packets.Enable.ENABLED,

                                                       hci_packets.FilterDuplicates.DISABLED, 0, 0))



    def test_le_connection_dut_advertises(self):

        with EventStream(self.dut.hal.StreamEvents(empty_pb2.Empty())) as hci_event_stream, \

                EventStream(self.cert.hal.StreamEvents(empty_pb2.Empty())) as cert_hci_event_stream, \

                EventStream(self.dut.hal.StreamAcl(empty_pb2.Empty())) as acl_data_stream, \

                EventStream(self.cert.hal.StreamAcl(empty_pb2.Empty())) as cert_acl_data_stream:



        # Cert Connects

            self.send_cert_hci_command(hci_packets.LeSetRandomAddressBuilder('0C:05:04:03:02:01'))

        self.cert_hal.send_hci_command(hci_packets.LeSetRandomAddressBuilder('0C:05:04:03:02:01'))

        phy_scan_params = hci_packets.LeCreateConnPhyScanParameters()

        phy_scan_params.scan_interval = 0x60

        phy_scan_params.scan_window = 0x30
@@ -184,15 +171,14 @@ class SimpleHalTest(GdBaseTestClass): 
        phy_scan_params.supervision_timeout = 0x1f4

        phy_scan_params.min_ce_length = 0

        phy_scan_params.max_ce_length = 0

            self.send_cert_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]))

        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]))



        # DUT Advertises

        advertising_handle = 0

            self.send_dut_hci_command(

        self.dut_hal.send_hci_command(

            hci_packets.LeSetExtendedAdvertisingLegacyParametersBuilder(

                advertising_handle,

                hci_packets.LegacyAdvertisingProperties.ADV_IND,
@@ -208,14 +194,14 @@ class SimpleHalTest(GdBaseTestClass): 
                hci_packets.Enable.DISABLED  # Scan request notification

            ))



            self.send_dut_hci_command(

        self.dut_hal.send_hci_command(

            hci_packets.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.data = list(bytes(b'Im_The_DUT'))



            self.send_dut_hci_command(

        self.dut_hal.send_hci_command(

            hci_packets.LeSetExtendedAdvertisingDataBuilder(

                advertising_handle, hci_packets.Operation.COMPLETE_ADVERTISEMENT,

                hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_name]))
@@ -224,7 +210,7 @@ class SimpleHalTest(GdBaseTestClass): 
        gap_short_name.data_type = hci_packets.GapDataType.SHORTENED_LOCAL_NAME

        gap_short_name.data = list(bytes(b'Im_The_D'))



            self.send_dut_hci_command(

        self.dut_hal.send_hci_command(

            hci_packets.LeSetExtendedAdvertisingScanResponseBuilder(

                advertising_handle, hci_packets.Operation.COMPLETE_ADVERTISEMENT,

                hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_short_name]))
@@ -233,7 +219,7 @@ class SimpleHalTest(GdBaseTestClass): 
        enabled_set.advertising_handle = advertising_handle

        enabled_set.duration = 0

        enabled_set.max_extended_advertising_events = 0

            self.send_dut_hci_command(

        self.dut_hal.send_hci_command(

            hci_packets.LeSetExtendedAdvertisingEnableBuilder(hci_packets.Enable.ENABLED, [enabled_set]))



        conn_handle = 0xfff
@@ -249,10 +235,10 @@ class SimpleHalTest(GdBaseTestClass): 
                return True

            return False



            assertThat(cert_hci_event_stream).emits(payload_handle)

        assertThat(self.cert_hal.get_hci_event_stream()).emits(payload_handle)

        cert_handle = conn_handle

        conn_handle = 0xfff

            assertThat(hci_event_stream).emits(payload_handle)

        assertThat(self.dut_hal.get_hci_event_stream()).emits(payload_handle)

        dut_handle = conn_handle



        # Send ACL Data
@@ -261,18 +247,14 @@ class SimpleHalTest(GdBaseTestClass): 
        self.send_cert_acl_data(cert_handle, hci_packets.PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE,

                                hci_packets.BroadcastFlag.POINT_TO_POINT, bytes(b'Just SomeMoreAclData'))



            assertThat(cert_acl_data_stream).emits(lambda packet: b'SomeAclData' in packet.payload)

            assertThat(acl_data_stream).emits(lambda packet: b'SomeMoreAclData' in packet.payload)

        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)



    def test_le_connect_list_connection_cert_advertises(self):

        with EventStream(self.dut.hal.StreamEvents(empty_pb2.Empty())) as hci_event_stream, \

            EventStream(self.cert.hal.StreamEvents(empty_pb2.Empty())) as cert_hci_event_stream:



        # DUT Connects

            self.send_dut_hci_command(hci_packets.LeSetRandomAddressBuilder('0D:05:04:03:02:01'))

            self.send_dut_hci_command(

                hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM,

                                                            '0C:05:04:03:02:01'))

        self.dut_hal.send_hci_command(hci_packets.LeSetRandomAddressBuilder('0D:05:04:03:02:01'))

        self.dut_hal.send_hci_command(

            hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM, '0C:05:04:03:02:01'))

        phy_scan_params = hci_packets.LeCreateConnPhyScanParameters()

        phy_scan_params.scan_interval = 0x60

        phy_scan_params.scan_window = 0x30
@@ -282,15 +264,14 @@ class SimpleHalTest(GdBaseTestClass): 
        phy_scan_params.supervision_timeout = 0x1f4

        phy_scan_params.min_ce_length = 0

        phy_scan_params.max_ce_length = 0

            self.send_dut_hci_command(

                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]))

        self.dut_hal.send_hci_command(

            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]))



        # CERT Advertises

        advertising_handle = 1

            self.send_cert_hci_command(

        self.cert_hal.send_hci_command(

            hci_packets.LeSetExtendedAdvertisingLegacyParametersBuilder(

                advertising_handle,

                hci_packets.LegacyAdvertisingProperties.ADV_IND,
@@ -306,14 +287,14 @@ class SimpleHalTest(GdBaseTestClass): 
                hci_packets.Enable.DISABLED  # Scan request notification

            ))



            self.send_cert_hci_command(

        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

        gap_name.data = list(bytes(b'Im_A_Cert'))



            self.send_cert_hci_command(

        self.cert_hal.send_hci_command(

            hci_packets.LeSetExtendedAdvertisingDataBuilder(

                advertising_handle, hci_packets.Operation.COMPLETE_ADVERTISEMENT,

                hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_name]))
@@ -321,11 +302,11 @@ class SimpleHalTest(GdBaseTestClass): 
        enabled_set.advertising_handle = 1

        enabled_set.duration = 0

        enabled_set.max_extended_advertising_events = 0

            self.send_cert_hci_command(

        self.cert_hal.send_hci_command(

            hci_packets.LeSetExtendedAdvertisingEnableBuilder(hci_packets.Enable.ENABLED, [enabled_set]))



        # LeConnectionComplete

            cert_hci_event_stream.assert_event_occurs(

        assertThat(self.cert_hal.get_hci_event_stream()).emits(

            lambda packet: b'\x3e\x13\x01\x00' in packet.payload, timeout=timedelta(seconds=20))

            hci_event_stream.assert_event_occurs(

        assertThat(self.dut_hal.get_hci_event_stream()).emits(

            lambda packet: b'\x3e\x13\x01\x00' in packet.payload, timeout=timedelta(seconds=20))

system/gd/hci/cert/direct_hci_test.py

+2 −2
Original line number Diff line number Diff line
@@ -36,7 +36,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().Serialize())

        self.cert_hal.send_hci_command(hci_packets.ResetBuilder())



    def teardown_test(self):

        self.dut_hci.close()
@@ -44,7 +44,7 @@ class DirectHciTest(GdBaseTestClass): 
        super().teardown_test()



    def send_hal_hci_command(self, command):

        self.cert_hal.send_hci_command(bytes(command.Serialize()))

        self.cert_hal.send_hci_command(command)



    def enqueue_acl_data(self, handle, pb_flag, b_flag, acl):

        acl_msg = hci_facade.AclMsg(