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

Commit 43ff4b50 authored by Treehugger Robot's avatar Treehugger Robot Committed by Automerger Merge Worker
Browse files

Merge "GD-SL4A: Add repeated connection test for background connections" am: 487b80eb 

Original change: https://android-review.googlesource.com/c/platform/packages/modules/Bluetooth/+/2129636



Change-Id: I157f4ded00450eb7eeeeea08993b8072db6c6182
Signed-off-by: default avatarAutomerger Merge Worker <android-build-automerger-merge-worker@system.gserviceaccount.com>
parents 06ab8bd9 487b80eb

system/blueberry/tests/gd/cert/closable.py

+9 −2
Original line number Diff line number Diff line
@@ -15,6 +15,7 @@ 
#   limitations under the License.

import time

from abc import ABC, abstractmethod

import logging





class Closable(ABC):
@@ -23,11 +24,17 @@ class Closable(ABC): 
        return self



    def __exit__(self, type, value, traceback):

        try:

            self.close()

        except Exception:

            logging.warning("Failed to close or already closed")

        return traceback is None



    def __del__(self):

        try:

            self.close()

        except Exception:

            logging.warning("Failed to close or already closed")



    @abstractmethod

    def close(self):

system/blueberry/tests/gd/cert/py_le_acl_manager.py

+5 −2
Original line number Diff line number Diff line
@@ -55,9 +55,12 @@ class PyLeAclManagerAclConnection(IEventStream, Closable): 
        safeClose(self.connection_event_stream)

        safeClose(self.acl_stream)



    def wait_for_disconnection_complete(self):

    def disconnect(self):

        self.le_acl_manager.Disconnect(le_acl_manager_facade.LeHandleMsg(handle=self.handle))



    def wait_for_disconnection_complete(self, timeout=timedelta(seconds=30)):

        disconnection_complete = HciCaptures.DisconnectionCompleteCapture()

        assertThat(self.connection_event_stream).emits(disconnection_complete)

        assertThat(self.connection_event_stream).emits(disconnection_complete, timeout=timeout)

        self.disconnect_reason = disconnection_complete.get().GetReason()



    def send(self, data):

system/blueberry/tests/gd_sl4a/gatt/gatt_connect_low_layer_test.py

0 → 100644
+536 −0

File added.

Preview size limit exceeded, changes collapsed.

system/blueberry/tests/gd_sl4a/gd_sl4a_test_runner.py

+2 −1
Original line number Diff line number Diff line
@@ -16,11 +16,12 @@ 


from blueberry.tests.gd_sl4a.hci.le_advanced_scanning_test import LeAdvancedScanningTest

from blueberry.tests.gd_sl4a.security.oob_pairing_sl4a_test import OobPairingSl4aTest

from blueberry.tests.gd_sl4a.gatt.gatt_connect_low_layer_test import GattConnectLowLayerTest



from mobly import suite_runner

import argparse



ALL_TESTS = [LeAdvancedScanningTest, OobPairingSl4aTest]

ALL_TESTS = [LeAdvancedScanningTest, OobPairingSl4aTest, GattConnectLowLayerTest]





def main():

system/blueberry/utils/bt_gatt_utils.py

+21 −2
Original line number Diff line number Diff line
@@ -48,14 +48,15 @@ def setup_gatt_connection(central: AndroidDevice, 
                          mac_address,

                          autoconnect,

                          transport=GattTransport.TRANSPORT_AUTO,

                          opportunistic=False):

                          opportunistic=False,

                          timeout_seconds=default_timeout):

    gatt_callback = central.sl4a.gattCreateGattCallback()

    log.info("Gatt Connect to mac address {}.".format(mac_address))

    bluetooth_gatt = central.sl4a.gattClientConnectGatt(gatt_callback, mac_address, autoconnect, transport,

                                                        opportunistic, GattPhyMask.PHY_LE_1M_MASK)

    expected_event = GattCallbackString.GATT_CONN_CHANGE.format(gatt_callback)

    try:

        event = central.ed.pop_event(expected_event, default_timeout)

        event = central.ed.pop_event(expected_event, timeout_seconds)

    except Empty:

        close_gatt_client(central, bluetooth_gatt)

        raise GattTestUtilsError("Could not establish a connection to "
@@ -68,6 +69,24 @@ def setup_gatt_connection(central: AndroidDevice, 
    return bluetooth_gatt, gatt_callback





def wait_for_gatt_connection(central: AndroidDevice, gatt_callback, bluetooth_gatt, timeout):

    expected_event = GattCallbackString.GATT_CONN_CHANGE.format(gatt_callback)

    try:

        event = central.ed.pop_event(expected_event, timeout=timeout)

    except Empty:

        close_gatt_client(central, bluetooth_gatt)

        raise GattTestUtilsError("Could not establish a connection to "

                                 "peripheral. Expected event: {}".format(expected_event))

    if event['data']['State'] != GattConnectionState.STATE_CONNECTED:

        close_gatt_client(central, bluetooth_gatt)

        try:

            central.sl4a.gattClientClose(bluetooth_gatt)

        except Exception:

            logging.debug("Failed to close gatt client.")

        raise GattTestUtilsError("Could not establish a connection to "

                                 "peripheral. Event Details: {}".format(pprint.pformat(event)))





def close_gatt_client(central: AndroidDevice, bluetooth_gatt):

    try:

        central.sl4a.gattClientClose(bluetooth_gatt)