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

Commit 4b0b637a authored by Chienyuan's avatar Chienyuan
Browse files

gd: Implement cert test for Scanning module 

Tag: #gd-refactor
Bug: 177044452
Test: gd/cert/run
Test: bluetooth_test_gd
BYPASS_LONG_LINES_REASON: Bluetooth likes 120 lines

Change-Id: I258289a2487f050ef43694ffbbdb4e983400972a
parent 2da5c703

system/gd/cert/matchers.py

+9 −1
Original line number Diff line number Diff line
@@ -211,7 +211,7 @@ class HciMatchers(object): 
class AdvertisingMatchers(object):



    @staticmethod

    def CallbackMsg(type, advertiser_id=None, status=None, data=None):

    def AdvertisingCallbackMsg(type, advertiser_id=None, status=None, data=None):

        return lambda event: True if event.message_type == type and (advertiser_id == None or advertiser_id == event.advertiser_id) \

            and (status == None or status == event.status) and (data == None or data == event.data) else False
@@ -221,6 +221,14 @@ class AdvertisingMatchers(object): 
            and (address == None or address == event.address) else False





class ScanningMatchers(object):



    @staticmethod

    def ScanningCallbackMsg(type, status=None, data=None):

        return lambda event: True if event.message_type == type and (status == None or status == event.status) \

            and (data == None or data == event.data) else False





class NeighborMatchers(object):



    @staticmethod

system/gd/hci/cert/le_advertising_manager_test_lib.py

+23 −20
Original line number Diff line number Diff line
@@ -32,8 +32,8 @@ from hci.facade import hci_facade_pb2 as hci_facade 
from hci.facade import \

  le_advertising_manager_facade_pb2 as le_advertising_facade

from hci.facade import le_initiator_address_facade_pb2 as le_initiator_address_facade

from hci.facade.le_advertising_manager_facade_pb2 import AdvertisingCallbackMsgType

from hci.facade.le_advertising_manager_facade_pb2 import AdvertisingStatus

from hci.facade.le_advertising_manager_facade_pb2 import CallbackMsgType





class LeAdvertisingManagerTestBase():
@@ -171,8 +171,9 @@ class LeAdvertisingManagerTestBase(): 
        self.set_address_policy_with_static_address()

        create_response = self.create_advertiser()

        assertThat(self.dut.callback_event_stream).emits(

            AdvertisingMatchers.CallbackMsg(CallbackMsgType.ADVERTISING_SET_STARTED, create_response.advertiser_id,

                                            AdvertisingStatus.SUCCESS, 0x00))

            AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_SET_STARTED,

                                                       create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS,

                                                       0x00))



    def test_enable_advertiser_callback(self):

        self.set_address_policy_with_static_address()
@@ -182,8 +183,9 @@ class LeAdvertisingManagerTestBase(): 
        self.dut.hci_le_advertising_manager.EnableAdvertiser(enable_advertiser_request)



        assertThat(self.dut.callback_event_stream).emits(

            AdvertisingMatchers.CallbackMsg(CallbackMsgType.ADVERTISING_ENABLED, create_response.advertiser_id,

                                            AdvertisingStatus.SUCCESS, 0x01))

            AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_ENABLED,

                                                       create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS,

                                                       0x01))



    def test_disable_advertiser_callback(self):

        self.set_address_policy_with_static_address()
@@ -193,8 +195,9 @@ class LeAdvertisingManagerTestBase(): 
        self.dut.hci_le_advertising_manager.EnableAdvertiser(disable_advertiser_request)



        assertThat(self.dut.callback_event_stream).emits(

            AdvertisingMatchers.CallbackMsg(CallbackMsgType.ADVERTISING_ENABLED, create_response.advertiser_id,

                                            AdvertisingStatus.SUCCESS, 0x00))

            AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_ENABLED,

                                                       create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS,

                                                       0x00))



    def test_set_advertising_data_callback(self):

        self.set_address_policy_with_static_address()
@@ -209,8 +212,8 @@ class LeAdvertisingManagerTestBase(): 
        self.dut.hci_le_advertising_manager.SetData(set_data_request)



        assertThat(self.dut.callback_event_stream).emits(

            AdvertisingMatchers.CallbackMsg(CallbackMsgType.ADVERTISING_DATA_SET, create_response.advertiser_id,

                                            AdvertisingStatus.SUCCESS))

            AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_DATA_SET,

                                                       create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS))



    def test_set_scan_response_data_callback(self):

        self.set_address_policy_with_static_address()
@@ -225,8 +228,8 @@ class LeAdvertisingManagerTestBase(): 
        self.dut.hci_le_advertising_manager.SetData(set_data_request)



        assertThat(self.dut.callback_event_stream).emits(

            AdvertisingMatchers.CallbackMsg(CallbackMsgType.SCAN_RESPONSE_DATA_SET, create_response.advertiser_id,

                                            AdvertisingStatus.SUCCESS))

            AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.SCAN_RESPONSE_DATA_SET,

                                                       create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS))



    def test_set_parameters_callback(self):

        self.set_address_policy_with_static_address()
@@ -250,8 +253,8 @@ class LeAdvertisingManagerTestBase(): 
        self.dut.hci_le_advertising_manager.SetParameters(set_parameters_request)



        assertThat(self.dut.callback_event_stream).emits(

            AdvertisingMatchers.CallbackMsg(CallbackMsgType.ADVERTISING_PARAMETERS_UPDATED,

                                            create_response.advertiser_id, AdvertisingStatus.SUCCESS))

            AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_PARAMETERS_UPDATED,

                                                       create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS))



    def test_set_periodic_parameters_callback(self):

        self.set_address_policy_with_static_address()
@@ -267,8 +270,8 @@ class LeAdvertisingManagerTestBase(): 
        self.dut.hci_le_advertising_manager.SetPeriodicParameters(set_periodic_parameters_request)



        assertThat(self.dut.callback_event_stream).emits(

            AdvertisingMatchers.CallbackMsg(CallbackMsgType.PERIODIC_ADVERTISING_PARAMETERS_UPDATED,

                                            create_response.advertiser_id))

            AdvertisingMatchers.AdvertisingCallbackMsg(

                AdvertisingCallbackMsgType.PERIODIC_ADVERTISING_PARAMETERS_UPDATED, create_response.advertiser_id))



    def test_set_periodic_data_callback(self):

        self.set_address_policy_with_static_address()
@@ -283,7 +286,7 @@ class LeAdvertisingManagerTestBase(): 
        self.dut.hci_le_advertising_manager.SetPeriodicData(set_periodic_data_request)



        assertThat(self.dut.callback_event_stream).emits(

            AdvertisingMatchers.CallbackMsg(CallbackMsgType.PERIODIC_ADVERTISING_DATA_SET,

            AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.PERIODIC_ADVERTISING_DATA_SET,

                                                       create_response.advertiser_id))



    def test_enable_periodic_advertising_callback(self):
@@ -294,7 +297,7 @@ class LeAdvertisingManagerTestBase(): 
        self.dut.hci_le_advertising_manager.EnablePeriodicAdvertising(enable_periodic_advertising_request)



        assertThat(self.dut.callback_event_stream).emits(

            AdvertisingMatchers.CallbackMsg(CallbackMsgType.PERIODIC_ADVERTISING_ENABLED,

            AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.PERIODIC_ADVERTISING_ENABLED,

                                                       create_response.advertiser_id))



    def test_get_own_address(self):
@@ -306,5 +309,5 @@ class LeAdvertisingManagerTestBase(): 
        address_with_type = common.BluetoothAddressWithType(

            address=common.BluetoothAddress(address=bytes(b'd0:05:04:03:02:01')), type=common.RANDOM_DEVICE_ADDRESS)

        assertThat(self.dut.address_event_stream).emits(

            AdvertisingMatchers.AddressMsg(CallbackMsgType.OWN_ADDRESS_READ, create_response.advertiser_id,

            AdvertisingMatchers.AddressMsg(AdvertisingCallbackMsgType.OWN_ADDRESS_READ, create_response.advertiser_id,

                                           address_with_type))

system/gd/hci/cert/le_scanning_manager_test.py

+8 −0
Original line number Diff line number Diff line
@@ -22,3 +22,11 @@ class LeScanningManagerTest(GdBaseTestClass, LeScanningManagerTestBase): 


    def setup_class(self):

        GdBaseTestClass.setup_class(self, dut_module='HCI_INTERFACES', cert_module='HCI_INTERFACES')



    def setup_test(self):

        GdBaseTestClass.setup_test(self)

        LeScanningManagerTestBase.setup_test(self, self.cert)



    def teardown_test(self):

        LeScanningManagerTestBase.teardown_test(self)

        GdBaseTestClass.teardown_test(self)

system/gd/hci/cert/le_scanning_manager_test_lib.py

+151 −29
Original line number Diff line number Diff line
@@ -18,19 +18,40 @@ import os 
import sys

import logging



from bluetooth_packets_python3 import hci_packets

from cert.closable import safeClose

from cert.event_stream import EventStream

from google.protobuf import empty_pb2 as empty_proto

from cert.matchers import ScanningMatchers

from cert.truth import assertThat

from facade import common_pb2 as common

from facade import rootservice_pb2 as facade_rootservice

from google.protobuf import empty_pb2 as empty_proto

from hci.facade import hci_facade_pb2 as hci_facade

from hci.facade import le_scanning_manager_facade_pb2 as le_scanning_facade

from hci.facade import le_advertising_manager_facade_pb2 as le_advertising_facade

from hci.facade import le_initiator_address_facade_pb2 as le_initiator_address_facade

from bluetooth_packets_python3 import hci_packets

from facade import common_pb2 as common

from hci.facade.le_scanning_manager_facade_pb2 import ScanningCallbackMsgType

from hci.facade.le_scanning_manager_facade_pb2 import ScanningStatus





class LeScanningManagerTestBase():



    def setup_test(self, cert):

        self.dut.callback_event_stream = EventStream(

            self.dut.hci_le_scanning_manager.FetchCallbackEvents(empty_proto.Empty()))

        self.dut.advertising_report_stream = EventStream(

            self.dut.hci_le_scanning_manager.FetchAdvertisingReports(empty_proto.Empty()))



    def teardown_test(self):

        if self.dut.callback_event_stream is not None:

            safeClose(self.dut.callback_event_stream)

        else:

            logging.info("DUT: Callback Event Stream is None!")

        if self.dut.advertising_report_stream is not None:

            safeClose(self.dut.advertising_report_stream)

        else:

            logging.info("DUT: Advertising Report Stream is None!")



    def register_for_event(self, event_code):

        msg = hci_facade.EventCodeMsg(code=int(event_code))

        self.cert.hci.RegisterEventHandler(msg)
@@ -47,7 +68,7 @@ class LeScanningManagerTestBase(): 
        else:

            self.cert.hci.EnqueueCommandWithStatus(cmd)



    def test_le_ad_scan_dut_scans(self):

    def set_address_policy_with_static_address(self):

        privacy_policy = le_initiator_address_facade.PrivacyPolicy(

            address_policy=le_initiator_address_facade.AddressPolicy.USE_STATIC_ADDRESS,

            address_with_type=common.BluetoothAddressWithType(
@@ -66,10 +87,9 @@ class LeScanningManagerTestBase(): 
            minimum_rotation_time=0,

            maximum_rotation_time=0)

        self.cert.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(cert_privacy_policy)

        with EventStream(

                # DUT Scans

                self.dut.hci_le_scanning_manager.StartScan(empty_proto.Empty())) as advertising_event_stream:



    def test_le_ad_scan_dut_scans(self):

        self.set_address_policy_with_static_address()

        # CERT Advertises

        gap_name = hci_packets.GapData()

        gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME
@@ -87,7 +107,109 @@ class LeScanningManagerTestBase(): 


        create_response = self.cert.hci_le_advertising_manager.CreateAdvertiser(request)



            advertising_event_stream.assert_event_occurs(lambda packet: b'Im_The_CERT' in packet.event)

        scan_request = le_scanning_facade.ScanRequest(start=True)

        self.dut.hci_le_scanning_manager.Scan(scan_request)



        self.dut.advertising_report_stream.assert_event_occurs(lambda packet: b'Im_The_CERT' in packet.event)



        remove_request = le_advertising_facade.RemoveAdvertiserRequest(advertiser_id=create_response.advertiser_id)

        self.cert.hci_le_advertising_manager.RemoveAdvertiser(remove_request)



    def test_register_scanner_callback(self):

        self.set_address_policy_with_static_address()

        register_request = le_scanning_facade.RegisterScannerRequest(uuid=123)

        self.dut.hci_le_scanning_manager.RegisterScanner(register_request)

        assertThat(self.dut.callback_event_stream).emits(

            ScanningMatchers.ScanningCallbackMsg(ScanningCallbackMsgType.SCANNER_REGISTERED,

                                                 ScanningStatus.SCAN_SUCCESS, 123))



    def test_register_scanner_with_same_uuid(self):

        self.set_address_policy_with_static_address()

        register_request = le_scanning_facade.RegisterScannerRequest(uuid=123)

        self.dut.hci_le_scanning_manager.RegisterScanner(register_request)

        assertThat(self.dut.callback_event_stream).emits(

            ScanningMatchers.ScanningCallbackMsg(ScanningCallbackMsgType.SCANNER_REGISTERED,

                                                 ScanningStatus.SCAN_SUCCESS, 123))

        self.dut.hci_le_scanning_manager.RegisterScanner(register_request)

        assertThat(self.dut.callback_event_stream).emits(

            ScanningMatchers.ScanningCallbackMsg(ScanningCallbackMsgType.SCANNER_REGISTERED,

                                                 ScanningStatus.SCAN_INTERNAL_ERROR, 123))



    def test_set_scan_parameters_callback(self):

        self.set_address_policy_with_static_address()

        set_scan_parameters_request = le_scanning_facade.SetScanParametersRequest(

            scanner_id=0x01, scan_type=le_scanning_facade.LeScanType.ACTIVE, scan_interval=0x10, scan_window=0x04)

        self.dut.hci_le_scanning_manager.SetScanParameters(set_scan_parameters_request)



        assertThat(self.dut.callback_event_stream).emits(

            ScanningMatchers.ScanningCallbackMsg(ScanningCallbackMsgType.SET_SCANNER_PARAMETER_COMPLETE,

                                                 ScanningStatus.SCAN_SUCCESS, 0x01))



    def test_set_scan_parameters_with_invalid_parameter(self):

        self.set_address_policy_with_static_address()

        set_scan_parameters_request = le_scanning_facade.SetScanParametersRequest(

            scanner_id=0x01, scan_type=le_scanning_facade.LeScanType.ACTIVE, scan_interval=0x00, scan_window=0x00)

        self.dut.hci_le_scanning_manager.SetScanParameters(set_scan_parameters_request)



        assertThat(self.dut.callback_event_stream).emits(

            ScanningMatchers.ScanningCallbackMsg(ScanningCallbackMsgType.SET_SCANNER_PARAMETER_COMPLETE,

                                                 ScanningStatus.SCAN_ILLEGAL_PARAMETER))



    def test_active_scan(self):

        self.set_address_policy_with_static_address()

        # CERT Advertises

        gap_name = hci_packets.GapData()

        gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME

        gap_name.data = list(bytes(b'Scan response data'))

        gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize()))

        config = le_advertising_facade.AdvertisingConfig(

            scan_response=[gap_data],

            interval_min=512,

            interval_max=768,

            advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND,

            own_address_type=common.USE_RANDOM_DEVICE_ADDRESS,

            channel_map=7,

            filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES)

        request = le_advertising_facade.CreateAdvertiserRequest(config=config)

        create_response = self.cert.hci_le_advertising_manager.CreateAdvertiser(request)



        set_scan_parameters_request = le_scanning_facade.SetScanParametersRequest(

            scanner_id=0x01, scan_type=le_scanning_facade.LeScanType.ACTIVE, scan_interval=0x10, scan_window=0x04)

        self.dut.hci_le_scanning_manager.SetScanParameters(set_scan_parameters_request)

        scan_request = le_scanning_facade.ScanRequest(start=True)

        self.dut.hci_le_scanning_manager.Scan(scan_request)



        self.dut.advertising_report_stream.assert_event_occurs(lambda packet: b'Scan response data' in packet.event)



        remove_request = le_advertising_facade.RemoveAdvertiserRequest(advertiser_id=create_response.advertiser_id)

        self.cert.hci_le_advertising_manager.RemoveAdvertiser(remove_request)



    def test_passive_scan(self):

        self.set_address_policy_with_static_address()

        # CERT Advertises

        gap_name = hci_packets.GapData()

        gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME

        gap_name.data = list(bytes(b'Scan response data'))

        gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize()))

        config = le_advertising_facade.AdvertisingConfig(

            scan_response=[gap_data],

            interval_min=512,

            interval_max=768,

            advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND,

            own_address_type=common.USE_RANDOM_DEVICE_ADDRESS,

            channel_map=7,

            filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES)

        request = le_advertising_facade.CreateAdvertiserRequest(config=config)

        create_response = self.cert.hci_le_advertising_manager.CreateAdvertiser(request)



        set_scan_parameters_request = le_scanning_facade.SetScanParametersRequest(

            scanner_id=0x01, scan_type=le_scanning_facade.LeScanType.PASSIVE, scan_interval=0x10, scan_window=0x04)

        self.dut.hci_le_scanning_manager.SetScanParameters(set_scan_parameters_request)

        scan_request = le_scanning_facade.ScanRequest(start=True)

        self.dut.hci_le_scanning_manager.Scan(scan_request)



        self.dut.advertising_report_stream.assert_event_occurs_at_most(

            lambda packet: b'Scan response data' in packet.event, 0)



        remove_request = le_advertising_facade.RemoveAdvertiserRequest(advertiser_id=create_response.advertiser_id)

        self.cert.hci_le_advertising_manager.RemoveAdvertiser(remove_request)
 
 No newline at end of file

system/gd/hci/cert/le_scanning_with_security_test_lib.py

+6 −1
Original line number Diff line number Diff line
@@ -20,6 +20,7 @@ from google.protobuf import empty_pb2 as empty_proto 
from hci.facade import hci_facade_pb2 as hci_facade

from hci.facade import le_advertising_manager_facade_pb2 as le_advertising_facade

from hci.facade import le_initiator_address_facade_pb2 as le_initiator_address_facade

from hci.facade import le_scanning_manager_facade_pb2 as le_scanning_facade

from bluetooth_packets_python3 import hci_packets

from facade import common_pb2 as common
@@ -57,7 +58,8 @@ class LeScanningWithSecurityTestBase(): 
        self.cert.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(cert_privacy_policy)

        with EventStream(

                # DUT Scans

                self.dut.hci_le_scanning_manager.StartScan(empty_proto.Empty())) as advertising_event_stream:

                self.dut.hci_le_scanning_manager.FetchAdvertisingReports(

                    empty_proto.Empty())) as advertising_event_stream:



            # CERT Advertises

            gap_name = hci_packets.GapData()
@@ -76,6 +78,9 @@ class LeScanningWithSecurityTestBase(): 


            create_response = self.cert.hci_le_advertising_manager.CreateAdvertiser(request)



            scan_request = le_scanning_facade.ScanRequest(start=True)

            self.dut.hci_le_scanning_manager.Scan(scan_request)



            assertThat(advertising_event_stream).emits(lambda packet: b'Im_The_CERT' in packet.event)



            remove_request = le_advertising_facade.RemoveAdvertiserRequest(advertiser_id=create_response.advertiser_id)