Merge "GD-SL4A: Add repeated connection test for background connections" into... (e4cb330c) · Commits · e / os / android_packages_modules_Bluetooth

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)