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

Commit 3b50b95d authored by Myles Watson's avatar Myles Watson
Browse files

Move EventAsserts close to with statements 

Bug: 148096951
Test: ./cert/run_cert_facade_only.sh
Change-Id: Icb9c6d4b555f80ad434824695b76e5a03c36021d
parent f8687784

system/gd/hci/cert/acl_manager_test.py

+11 −8
Original line number Diff line number Diff line
@@ -86,6 +86,10 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass): 
            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:



            cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)

            acl_data_asserts = EventAsserts(acl_data_stream)

            cert_acl_data_asserts = EventAsserts(cert_acl_data_stream)



            # CERT Enables scans and gets its address

            self.enqueue_hci_command(

                hci_packets.WriteScanEnableBuilder(
@@ -108,7 +112,6 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass): 


            self.enqueue_hci_command(hci_packets.ReadBdAddrBuilder(), True)



            cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)

            cert_hci_event_asserts.assert_event_occurs(

                get_address_from_complete)
@@ -119,6 +122,8 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass): 
                                hci_packets.AddressType.PUBLIC_DEVICE_ADDRESS),

                            address=bytes(cert_address,

                                          'utf8')))) as connection_event_stream:



                connection_event_asserts = EventAsserts(connection_event_stream)

                connection_request = None



                def get_connect_request(packet):
@@ -166,7 +171,6 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass): 
                    ))



                # DUT gets a connection complete event and sends and receives

                connection_event_asserts = EventAsserts(connection_event_stream)

                handle = 0xfff

                connection_event_asserts.assert_event_occurs(get_handle)
@@ -177,8 +181,6 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass): 
                            b'\x29\x00\x07\x00This is just SomeMoreAclData from the DUT'

                        )))



                acl_data_asserts = EventAsserts(acl_data_stream)

                cert_acl_data_asserts = EventAsserts(cert_acl_data_stream)

                cert_acl_data_asserts.assert_event_occurs(

                    lambda packet: b'SomeMoreAclData' in packet.data)

                acl_data_asserts.assert_event_occurs(
@@ -191,6 +193,9 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass): 
            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:



            cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)

            acl_data_asserts = EventAsserts(acl_data_stream)



            # CERT Enables scans and gets its address

            self.enqueue_hci_command(

                hci_packets.WriteScanEnableBuilder(
@@ -213,7 +218,6 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass): 


            self.enqueue_hci_command(hci_packets.ReadBdAddrBuilder(), True)



            cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)

            cert_hci_event_asserts.assert_event_occurs(

                get_address_from_complete)
@@ -224,6 +228,8 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass): 
                                hci_packets.AddressType.PUBLIC_DEVICE_ADDRESS),

                            address=bytes(cert_address,

                                          'utf8')))) as connection_event_stream:



                connection_event_asserts = EventAsserts(connection_event_stream)

                connection_request = None



                def get_connect_request(packet):
@@ -262,8 +268,6 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass): 
                cert_hci_event_asserts.assert_event_occurs(get_handle)

                cert_handle = handle



                acl_data_asserts = EventAsserts(acl_data_stream)



                self.enqueue_acl_data(

                    cert_handle, hci_packets.PacketBoundaryFlag.

                    FIRST_AUTOMATICALLY_FLUSHABLE,
@@ -280,7 +284,6 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass): 
                    bytes(b'\x88\x13\x07\x00' + b'Hello' * 1000))



                # DUT gets a connection complete event and sends and receives

                connection_event_asserts = EventAsserts(connection_event_stream)

                connection_event_asserts.assert_event_occurs(get_handle)



                acl_data_asserts.assert_event_occurs(

system/gd/hci/cert/direct_hci_test.py

+23 −21
Original line number Diff line number Diff line
@@ -106,6 +106,7 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
        with EventCallbackStream(

                self.device_under_test.hci.FetchEvents(

                    empty_proto.Empty())) as hci_event_stream:

            hci_event_asserts = EventAsserts(hci_event_stream)



            self.enqueue_hci_command(

                hci_packets.WriteLoopbackModeBuilder(
@@ -115,7 +116,6 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
            self.enqueue_hci_command(cmd2loop, True)



            looped_bytes = bytes(cmd2loop.Serialize())

            hci_event_asserts = EventAsserts(hci_event_stream)

            hci_event_asserts.assert_event_occurs(

                lambda packet: looped_bytes in packet.event)
@@ -124,6 +124,9 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
        with EventCallbackStream(

                self.device_under_test.hci.FetchEvents(

                    empty_proto.Empty())) as hci_event_stream:



            hci_event_asserts = EventAsserts(hci_event_stream)



            self.send_hal_hci_command(

                hci_packets.WriteScanEnableBuilder(

                    hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN))
@@ -131,7 +134,6 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
            lap.lap = 0x33

            self.enqueue_hci_command(

                hci_packets.InquiryBuilder(lap, 0x30, 0xff), False)

            hci_event_asserts = EventAsserts(hci_event_stream)

            hci_event_asserts.assert_event_occurs(

                lambda packet: b'\x02\x0f' in packet.event

                # Expecting an HCI Event (code 0x02, length 0x0f)
@@ -143,6 +145,8 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
                self.device_under_test.hci.FetchLeSubevents(

                    empty_proto.Empty())) as hci_le_event_stream:



            hci_event_asserts = EventAsserts(hci_le_event_stream)



            # DUT Scans

            self.enqueue_hci_command(

                hci_packets.LeSetRandomAddressBuilder('0D:05:04:03:02:01'),
@@ -180,7 +184,6 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
                hci_packets.LeSetAdvertisingEnableBuilder(

                    hci_packets.Enable.ENABLED))



            hci_event_asserts = EventAsserts(hci_le_event_stream)

            hci_event_asserts.assert_event_occurs(

                lambda packet: b'Im_A_Cert' in packet.event)
@@ -199,6 +202,11 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
            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:



            le_event_asserts = EventAsserts(le_event_stream)

            cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)

            acl_data_asserts = EventAsserts(acl_data_stream)

            cert_acl_data_asserts = EventAsserts(cert_acl_data_stream)



            self.send_hal_hci_command(

                hci_packets.LeSetRandomAddressBuilder('0C:05:04:03:02:01'))
@@ -263,9 +271,6 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
                    return True

                return False



            le_event_asserts = EventAsserts(le_event_stream)

            cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)



            cert_hci_event_asserts.assert_event_occurs(payload_handle)

            cert_handle = conn_handle

            conn_handle = 0xfff
@@ -290,8 +295,6 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
                hci_packets.BroadcastFlag.POINT_TO_POINT,

                bytes(b'This is just SomeMoreAclData'))



            acl_data_asserts = EventAsserts(acl_data_stream)

            cert_acl_data_asserts = EventAsserts(cert_acl_data_stream)

            cert_acl_data_asserts.assert_event_occurs(

                lambda packet: logging.debug(packet.payload) or b'SomeAclData' in packet.payload

            )
@@ -357,6 +360,11 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
            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:



            cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)

            hci_event_asserts = EventAsserts(hci_event_stream)

            acl_data_asserts = EventAsserts(acl_data_stream)

            cert_acl_data_asserts = EventAsserts(cert_acl_data_stream)



            address = hci_packets.Address()



            def get_address_from_complete(packet):
@@ -375,7 +383,6 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
            # CERT Enables scans and gets its address

            self.send_hal_hci_command(hci_packets.ReadBdAddrBuilder())



            cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)

            cert_hci_event_asserts.assert_event_occurs(

                get_address_from_complete)

            self.send_hal_hci_command(
@@ -431,8 +438,6 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
                packet_bytes = packet.payload

                return get_handle(packet_bytes)



            hci_event_asserts = EventAsserts(hci_event_stream)



            cert_hci_event_asserts.assert_event_occurs(payload_handle)

            cert_handle = conn_handle

            conn_handle = 0xfff
@@ -457,8 +462,6 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
                hci_packets.BroadcastFlag.POINT_TO_POINT,

                bytes(b'This is just SomeMoreAclData'))



            acl_data_asserts = EventAsserts(acl_data_stream)

            cert_acl_data_asserts = EventAsserts(cert_acl_data_stream)

            cert_acl_data_asserts.assert_event_occurs(

                lambda packet: b'SomeAclData' in packet.payload)

            acl_data_asserts.assert_event_occurs(
@@ -472,6 +475,11 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
            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:



            hci_event_asserts = EventAsserts(hci_event_stream)

            cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)

            acl_data_asserts = EventAsserts(acl_data_stream)

            cert_acl_data_asserts = EventAsserts(cert_acl_data_stream)



            address = hci_packets.Address()



            def get_address_from_complete(packet):
@@ -493,8 +501,7 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
                    hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN), True)

            self.enqueue_hci_command(hci_packets.ReadBdAddrBuilder(), True)



            dut_hci_event_asserts = EventAsserts(hci_event_stream)

            dut_hci_event_asserts.assert_event_occurs(get_address_from_complete)

            hci_event_asserts.assert_event_occurs(get_address_from_complete)



            # Cert Connects

            self.send_hal_hci_command(
@@ -516,7 +523,7 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
                    return True

                return False



            dut_hci_event_asserts.assert_event_occurs(get_connect_request)

            hci_event_asserts.assert_event_occurs(get_connect_request)

            self.enqueue_hci_command(

                hci_packets.AcceptConnectionRequestBuilder(

                    connection_request.GetBdAddr(),
@@ -544,9 +551,6 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
                packet_bytes = packet.payload

                return get_handle(packet_bytes)



            hci_event_asserts = EventAsserts(hci_event_stream)



            cert_hci_event_asserts = EventAsserts(cert_hci_event_stream)

            cert_hci_event_asserts.assert_event_occurs(payload_handle)

            cert_handle = conn_handle

            conn_handle = 0xfff
@@ -571,8 +575,6 @@ class DirectHciTest(GdFacadeOnlyBaseTestClass): 
                hci_packets.BroadcastFlag.POINT_TO_POINT,

                bytes(b'This is just SomeMoreAclData'))



            acl_data_asserts = EventAsserts(acl_data_stream)

            cert_acl_data_asserts = EventAsserts(cert_acl_data_stream)

            cert_acl_data_asserts.assert_event_occurs(

                lambda packet: b'SomeAclData' in packet.payload)

            acl_data_asserts.assert_event_occurs(

system/gd/hci/cert/le_advertising_manager_test.py

+3 −1
Original line number Diff line number Diff line
@@ -79,6 +79,9 @@ class LeAdvertisingManagerTest(GdFacadeOnlyBaseTestClass): 
        with EventCallbackStream(

                self.cert_device.hci.FetchLeSubevents(

                    empty_proto.Empty())) as hci_le_event_stream:



            hci_event_asserts = EventAsserts(hci_le_event_stream)



            # CERT Scans

            self.enqueue_hci_command(

                hci_packets.LeSetRandomAddressBuilder('0C:05:04:03:02:01'),
@@ -120,7 +123,6 @@ class LeAdvertisingManagerTest(GdFacadeOnlyBaseTestClass): 
            create_response = self.device_under_test.hci_le_advertising_manager.CreateAdvertiser(

                request)



            hci_event_asserts = EventAsserts(hci_le_event_stream)

            hci_event_asserts.assert_event_occurs(

                lambda packet: b'Im_The_DUT' in packet.event)

system/gd/hci/cert/le_scanning_manager_test.py

+2 −1
Original line number Diff line number Diff line
@@ -80,6 +80,8 @@ class LeScanningManagerTest(GdFacadeOnlyBaseTestClass): 
                self.device_under_test.hci_le_scanning_manager.StartScan(

                    empty_proto.Empty())) as advertising_event_stream:



            hci_event_asserts = EventAsserts(advertising_event_stream)



            # CERT Advertises

            gap_name = hci_packets.GapData()

            gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME
@@ -106,7 +108,6 @@ class LeScanningManagerTest(GdFacadeOnlyBaseTestClass): 
            create_response = self.cert_device.hci_le_advertising_manager.CreateAdvertiser(

                request)



            hci_event_asserts = EventAsserts(advertising_event_stream)

            hci_event_asserts.assert_event_occurs(

                lambda packet: b'Im_The_CERT' in packet.event)