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

Commit 8d1fcb03 authored by Rahul Sabnis's avatar Rahul Sabnis Committed by Android (Google) Code Review
Browse files

Merge changes from topic "generate_oob_timeout-tm-qpr-dev" into tm-qpr-dev 

* changes:
  Adds SL4A_SL4A test for LE scanning
  Add a timeout for generating local oob data
  Add an IRK rotation sl4a_sl4a test
parents 1ec0093e 3cc11eaa

android/app/src/com/android/bluetooth/btservice/AdapterService.java

+18 −1
Original line number Diff line number Diff line
@@ -169,6 +169,7 @@ public class AdapterService extends Service { 
    private static final int MIN_OFFLOADED_FILTERS = 10;

    private static final int MIN_OFFLOADED_SCAN_STORAGE_BYTES = 1024;

    private static final Duration PENDING_SOCKET_HANDOFF_TIMEOUT = Duration.ofMinutes(1);

    private static final Duration GENERATE_LOCAL_OOB_DATA_TIMEOUT = Duration.ofSeconds(2);



    private final Object mEnergyInfoLock = new Object();

    private int mStackReportedState;
@@ -3999,15 +4000,31 @@ public class AdapterService extends Service { 
        if (mOobDataCallbackQueue.peek() != null) {

            try {

                callback.onError(BluetoothStatusCodes.ERROR_ANOTHER_ACTIVE_OOB_REQUEST);

                return;

            } catch (RemoteException e) {

                Log.e(TAG, "Failed to make callback", e);

            }

            return;

        }

        mOobDataCallbackQueue.offer(callback);

        mHandler.postDelayed(() -> removeFromOobDataCallbackQueue(callback),

                GENERATE_LOCAL_OOB_DATA_TIMEOUT.toMillis());

        generateLocalOobDataNative(transport);

    }



    private synchronized void removeFromOobDataCallbackQueue(IBluetoothOobDataCallback callback) {

        if (callback == null) {

            return;

        }



        if (mOobDataCallbackQueue.peek() == callback) {

            try {

                mOobDataCallbackQueue.poll().onError(BluetoothStatusCodes.ERROR_UNKNOWN);

            } catch (RemoteException e) {

                Log.e(TAG, "Failed to make OobDataCallback to remove callback from queue", e);

            }

        }

    }



    /* package */ synchronized void notifyOobDataCallback(int transport, OobData oobData) {

        if (mOobDataCallbackQueue.peek() == null) {

            Log.e(TAG, "Failed to make callback, no callback exists");

system/blueberry/tests/sl4a_sl4a/advertising/le_advertising.py

+3 −3
Original line number Diff line number Diff line
@@ -38,7 +38,7 @@ class LeAdvertisingTest(sl4a_sl4a_base_test.Sl4aSl4aBaseTestClass): 
        super().teardown_test()



    def test_advertise_name(self):

        rpa_address = self.cert_advertiser_.advertise_rpa_public_extended_pdu()

        rpa_address = self.cert_advertiser_.advertise_public_extended_pdu()

        self.dut_scanner_.scan_for_name(self.cert_advertiser_.get_local_advertising_name())

        self.dut_scanner_.stop_scanning()

        self.cert_advertiser_.stop_advertising()
@@ -48,10 +48,10 @@ class LeAdvertisingTest(sl4a_sl4a_base_test.Sl4aSl4aBaseTestClass): 
            self.test_advertise_name()



    def test_advertise_name_twice_no_stop(self):

        rpa_address = self.cert_advertiser_.advertise_rpa_public_extended_pdu()

        rpa_address = self.cert_advertiser_.advertise_public_extended_pdu()

        self.dut_scanner_.scan_for_name(self.cert_advertiser_.get_local_advertising_name())

        self.dut_scanner_.stop_scanning()

        rpa_address = self.cert_advertiser_.advertise_rpa_public_extended_pdu()

        rpa_address = self.cert_advertiser_.advertise_public_extended_pdu()

        self.dut_scanner_.scan_for_name(self.cert_advertiser_.get_local_advertising_name())

        self.dut_scanner_.stop_scanning()

        self.cert_advertiser_.stop_advertising()

system/blueberry/tests/sl4a_sl4a/lib/l2cap.py

0 → 100644
+38 −0
Original line number Diff line number Diff line 
#!/usr/bin/env python3

#

#   Copyright 2022 - The Android Open Source Project

#

#   Licensed under the Apache License, Version 2.0 (the "License");

#   you may not use this file except in compliance with the License.

#   You may obtain a copy of the License at

#

#       http://www.apache.org/licenses/LICENSE-2.0

#

#   Unless required by applicable law or agreed to in writing, software

#   distributed under the License is distributed on an "AS IS" BASIS,

#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

#   See the License for the specific language governing permissions and

#   limitations under the License.



import logging



from blueberry.tests.gd.cert.truth import assertThat





class L2cap:



    __l2cap_connection_timeout = 30  #seconds

    __device = None



    def __init__(self, device):

        self.__device = device



    def create_l2cap_le_coc(self, address, psm, secure):

        logging.info("creating l2cap channel with secure=%r and psm %s", secure, psm)

        self.__device.sl4a.bluetoothSocketConnBeginConnectThreadPsm(address, True, psm, secure)



    # Starts listening on the l2cap server socket, returns the psm

    def listen_using_l2cap_coc(self, secure):

        logging.info("Listening for l2cap channel with secure=%r and psm %s", secure, psm)

        self.__device.sl4a.bluetoothSocketConnBeginAcceptThreadPsm(__l2cap_connection_timeout, True, secure)

        return self.__device.sl4a.bluetoothSocketConnGetPsm()

system/blueberry/tests/sl4a_sl4a/lib/le_advertiser.py

+3 −2
Original line number Diff line number Diff line
@@ -49,16 +49,17 @@ class LeAdvertiser(Closable): 
            return False

        return True



    def advertise_rpa_public_extended_pdu(self, name="SL4A Device"):

    def advertise_public_extended_pdu(self, address_type=common.RANDOM_DEVICE_ADDRESS, name="SL4A Device"):

        if self.is_advertising:

            logging.info("Already advertising!")

            return

        logging.info("Configuring advertisement with address type %d", address_type)

        self.is_advertising = True

        self.device.sl4a.bleSetScanSettingsLegacy(False)

        self.device.sl4a.bleSetAdvertiseSettingsIsConnectable(True)

        self.device.sl4a.bleSetAdvertiseDataIncludeDeviceName(True)

        self.device.sl4a.bleSetAdvertiseSettingsAdvertiseMode(ble_advertise_settings_modes['low_latency'])

        self.device.sl4a.bleSetAdvertiseSettingsOwnAddressType(common.RANDOM_DEVICE_ADDRESS)

        self.device.sl4a.bleSetAdvertiseSettingsOwnAddressType(address_type)

        self.advertise_callback, self.advertise_data, self.advertise_settings = generate_ble_advertise_objects(

            self.device.sl4a)

        self.device.sl4a.bleStartBleAdvertising(self.advertise_callback, self.advertise_data, self.advertise_settings)

system/blueberry/tests/sl4a_sl4a/lib/le_scanner.py

+9 −4
Original line number Diff line number Diff line
@@ -49,7 +49,7 @@ class LeScanner(Closable): 
    def scan_for_address_expect_none(self, address, addr_type):

        if self.is_scanning:

            print("Already scanning!")

            return

            return None

        self.is_scanning = True

        logging.info("Start scanning for identity address {} or type {}".format(address, addr_type))

        self.device.sl4a.bleSetScanSettingsScanMode(ble_scan_settings_modes['low_latency'])
@@ -66,11 +66,12 @@ class LeScanner(Closable): 
        advertising_address = self.__wait_for_scan_result_event(expected_event_name, 1)

        assertThat(advertising_address).isNone()

        logging.info("Filter advertisement with address {}".format(advertising_address))

        return advertising_address



    def scan_for_address(self, address, addr_type):

        if self.is_scanning:

            print("Already scanning!")

            return

            return None

        self.is_scanning = True

        logging.info("Start scanning for identity address {} or type {}".format(address, addr_type))

        self.device.sl4a.bleSetScanSettingsScanMode(ble_scan_settings_modes['low_latency'])
@@ -87,11 +88,12 @@ class LeScanner(Closable): 
        advertising_address = self.__wait_for_scan_result_event(expected_event_name)

        assertThat(advertising_address).isNotNone()

        logging.info("Filter advertisement with address {}".format(advertising_address))

        return advertising_address



    def scan_for_address_with_irk(self, address, addr_type, irk):

        if self.is_scanning:

            print("Already scanning!")

            return

            return None

        self.is_scanning = True

        logging.info("Start scanning for identity address {} or type {} using irk {}".format(address, addr_type, irk))

        self.device.sl4a.bleSetScanSettingsScanMode(ble_scan_settings_modes['low_latency'])
@@ -108,11 +110,12 @@ class LeScanner(Closable): 
        advertising_address = self.__wait_for_scan_result_event(expected_event_name)

        assertThat(advertising_address).isNotNone()

        logging.info("Filter advertisement with address {}".format(advertising_address))

        return advertising_address



    def scan_for_address_with_irk_pending_intent(self, address, addr_type, irk):

        if self.is_scanning:

            print("Already scanning!")

            return

            return None

        self.is_scanning = True

        logging.info("Start scanning for identity address {} or type {} using irk {}".format(address, addr_type, irk))

        self.device.sl4a.bleSetScanSettingsScanMode(ble_scan_settings_modes['low_latency'])
@@ -131,6 +134,7 @@ class LeScanner(Closable): 
        advertising_address = self.__wait_for_scan_result_event(expected_event_name)

        assertThat(advertising_address).isNotNone()

        logging.info("Filter advertisement with address {}".format(advertising_address))

        return advertising_address



    def scan_for_name(self, name):

        if self.is_scanning:
@@ -142,6 +146,7 @@ class LeScanner(Closable): 
        self.device.sl4a.bleSetScanSettingsLegacy(False)

        self.filter_list, self.scan_settings, self.scan_callback = generate_ble_scan_objects(self.device.sl4a)

        expected_event_name = scan_result.format(1)

        self.device.ed.clear_events(expected_event_name)



        # Start scanning on SL4A DUT

        self.device.sl4a.bleSetScanFilterDeviceName(name)