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

Commit 2a2172d4 authored by Jack He's avatar Jack He
Browse files

GD-SL4A: Add repeated connection test for background connections 

* Add test_autoconnect_gatt_without_pairing_and_disconnect_quickly
  to check if GATT is tear down fast enought after last client
  unregisters
* Add test_autoconnect_gatt_twice_with_random_address_without_pairing
  to check if we can make subsequent GATT connections
* Both tests are for random address that are unbonded

Bug: 232258524
Test: gd/cert/run --sl4a GattConnectLowLayerTest.test_autoconnect_gatt_twice_with_random_address_without_pairing
Test: gd/cert/run --sl4a GattConnectLowLayerTest.test_autoconnect_gatt_without_pairing_and_disconnect_quickly
Test: gd/cert/run --sl4a GattConnectLowLayerTest.test_disconnect_autoconnect_without_close
Test: gd/cert/run --sl4a GattConnectLowLayerTest.test_autoconnect_without_proactive_disconnect
Test: gd/cert/run --sl4a GattConnectLowLayerTest.test_autoconnect_without_proactive_disconnect_repeatedly
Tag: #bug-fix
Change-Id: Iad813154d87c4bb18e7762b8a4c5ef414979b8e8
parent 4db821d8

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