Blueberry: Migrate GattNotifyTest and GattConnectTest to Mobly (65abee7b) · Commits · e / os / android_packages_modules_Bluetooth

system/blueberry/tests/gd_sl4a/lib/bt_constants.py

+0 −269

Original line number	Original line	Diff line number	Diff line
	@@ -60,7 +60,6 @@ adv_succ = "BleAdvertise{}onSuccess"
	bluetooth_off = "BluetoothStateChangedOff"		bluetooth_off = "BluetoothStateChangedOff"
	bluetooth_on = "BluetoothStateChangedOn"		bluetooth_on = "BluetoothStateChangedOn"
	mtu_changed = "GattConnect{}onMtuChanged"		mtu_changed = "GattConnect{}onMtuChanged"
	gatt_connection_state_change = "GattConnect{}onConnectionStateChange"
	advertising_set_started = "AdvertisingSet{}onAdvertisingSetStarted"		advertising_set_started = "AdvertisingSet{}onAdvertisingSetStarted"
	advertising_set_stopped = "AdvertisingSet{}onAdvertisingSetStopped"		advertising_set_stopped = "AdvertisingSet{}onAdvertisingSetStopped"
	advertising_set_on_own_address_read = "AdvertisingSet{}onOwnAddressRead"		advertising_set_on_own_address_read = "AdvertisingSet{}onOwnAddressRead"
	@@ -261,274 +260,6 @@ ble_advertise_error_code = {

	### Bluetooth Low Energy Constants End ###		### Bluetooth Low Energy Constants End ###

	### Bluetooth GATT Constants Begin ###

	# Gatt Callback error messages
	gatt_cb_err = {
	"char_write_req_err": "Characteristic Write Request event not found. Expected {}",
	"char_write_err": "Characteristic Write event not found. Expected {}",
	"desc_write_req_err": "Descriptor Write Request event not found. Expected {}",
	"desc_write_err": "Descriptor Write event not found. Expected {}",
	"char_read_err": "Characteristic Read event not found. Expected {}",
	"char_read_req_err": "Characteristic Read Request not found. Expected {}",
	"desc_read_err": "Descriptor Read event not found. Expected {}",
	"desc_read_req_err": "Descriptor Read Request event not found. Expected {}",
	"rd_remote_rssi_err": "Read Remote RSSI event not found. Expected {}",
	"gatt_serv_disc_err": "GATT Services Discovered event not found. Expected {}",
	"serv_added_err": "Service Added event not found. Expected {}",
	"mtu_changed_err": "MTU Changed event not found. Expected {}",
	"mtu_serv_changed_err": "MTU Server Changed event not found. Expected {}",
	"gatt_conn_changed_err": "GATT Connection Changed event not found. Expected {}",
	"char_change_err": "GATT Characteristic Changed event not fond. Expected {}",
	"phy_read_err": "Phy Read event not fond. Expected {}",
	"phy_update_err": "Phy Update event not fond. Expected {}",
	"exec_write_err": "GATT Execute Write event not found. Expected {}"
	}

	# GATT callback strings as defined in GattClientFacade.java and
	# GattServerFacade.java implemented callbacks.
	gatt_cb_strings = {
	"char_write_req": "GattServer{}onCharacteristicWriteRequest",
	"exec_write": "GattServer{}onExecuteWrite",
	"char_write": "GattConnect{}onCharacteristicWrite",
	"desc_write_req": "GattServer{}onDescriptorWriteRequest",
	"desc_write": "GattConnect{}onDescriptorWrite",
	"char_read": "GattConnect{}onCharacteristicRead",
	"char_read_req": "GattServer{}onCharacteristicReadRequest",
	"desc_read": "GattConnect{}onDescriptorRead",
	"desc_read_req": "GattServer{}onDescriptorReadRequest",
	"rd_remote_rssi": "GattConnect{}onReadRemoteRssi",
	"gatt_serv_disc": "GattConnect{}onServicesDiscovered",
	"serv_added": "GattServer{}onServiceAdded",
	"mtu_changed": "GattConnect{}onMtuChanged",
	"mtu_serv_changed": "GattServer{}onMtuChanged",
	"gatt_conn_change": "GattConnect{}onConnectionStateChange",
	"char_change": "GattConnect{}onCharacteristicChanged",
	"phy_read": "GattConnect{}onPhyRead",
	"phy_update": "GattConnect{}onPhyUpdate",
	"serv_phy_read": "GattServer{}onPhyRead",
	"serv_phy_update": "GattServer{}onPhyUpdate",
	}

	# GATT event dictionary of expected callbacks and errors.
	gatt_event = {
	"char_write_req": {
	"evt": gatt_cb_strings["char_write_req"],
	"err": gatt_cb_err["char_write_req_err"]
	},
	"exec_write": {
	"evt": gatt_cb_strings["exec_write"],
	"err": gatt_cb_err["exec_write_err"]
	},
	"char_write": {
	"evt": gatt_cb_strings["char_write"],
	"err": gatt_cb_err["char_write_err"]
	},
	"desc_write_req": {
	"evt": gatt_cb_strings["desc_write_req"],
	"err": gatt_cb_err["desc_write_req_err"]
	},
	"desc_write": {
	"evt": gatt_cb_strings["desc_write"],
	"err": gatt_cb_err["desc_write_err"]
	},
	"char_read": {
	"evt": gatt_cb_strings["char_read"],
	"err": gatt_cb_err["char_read_err"]
	},
	"char_read_req": {
	"evt": gatt_cb_strings["char_read_req"],
	"err": gatt_cb_err["char_read_req_err"]
	},
	"desc_read": {
	"evt": gatt_cb_strings["desc_read"],
	"err": gatt_cb_err["desc_read_err"]
	},
	"desc_read_req": {
	"evt": gatt_cb_strings["desc_read_req"],
	"err": gatt_cb_err["desc_read_req_err"]
	},
	"rd_remote_rssi": {
	"evt": gatt_cb_strings["rd_remote_rssi"],
	"err": gatt_cb_err["rd_remote_rssi_err"]
	},
	"gatt_serv_disc": {
	"evt": gatt_cb_strings["gatt_serv_disc"],
	"err": gatt_cb_err["gatt_serv_disc_err"]
	},
	"serv_added": {
	"evt": gatt_cb_strings["serv_added"],
	"err": gatt_cb_err["serv_added_err"]
	},
	"mtu_changed": {
	"evt": gatt_cb_strings["mtu_changed"],
	"err": gatt_cb_err["mtu_changed_err"]
	},
	"gatt_conn_change": {
	"evt": gatt_cb_strings["gatt_conn_change"],
	"err": gatt_cb_err["gatt_conn_changed_err"]
	},
	"char_change": {
	"evt": gatt_cb_strings["char_change"],
	"err": gatt_cb_err["char_change_err"]
	},
	"phy_read": {
	"evt": gatt_cb_strings["phy_read"],
	"err": gatt_cb_err["phy_read_err"]
	},
	"phy_update": {
	"evt": gatt_cb_strings["phy_update"],
	"err": gatt_cb_err["phy_update_err"]
	},
	"serv_phy_read": {
	"evt": gatt_cb_strings["serv_phy_read"],
	"err": gatt_cb_err["phy_read_err"]
	},
	"serv_phy_update": {
	"evt": gatt_cb_strings["serv_phy_update"],
	"err": gatt_cb_err["phy_update_err"]
	}
	}

	# Matches constants of connection states defined in BluetoothGatt.java
	gatt_connection_state = {"disconnected": 0, "connecting": 1, "connected": 2, "disconnecting": 3, "closed": 4}

	# Matches constants of Bluetooth GATT Characteristic values as defined
	# in BluetoothGattCharacteristic.java
	gatt_characteristic = {
	"property_broadcast": 0x01,
	"property_read": 0x02,
	"property_write_no_response": 0x04,
	"property_write": 0x08,
	"property_notify": 0x10,
	"property_indicate": 0x20,
	"property_signed_write": 0x40,
	"property_extended_props": 0x80,
	"permission_read": 0x01,
	"permission_read_encrypted": 0x02,
	"permission_read_encrypted_mitm": 0x04,
	"permission_write": 0x10,
	"permission_write_encrypted": 0x20,
	"permission_write_encrypted_mitm": 0x40,
	"permission_write_signed": 0x80,
	"permission_write_signed_mitm": 0x100,
	"write_type_default": 0x02,
	"write_type_no_response": 0x01,
	"write_type_signed": 0x04,
	}

	# Matches constants of Bluetooth GATT Characteristic values as defined
	# in BluetoothGattDescriptor.java
	gatt_descriptor = {
	"enable_notification_value": [0x01, 0x00],
	"enable_indication_value": [0x02, 0x00],
	"disable_notification_value": [0x00, 0x00],
	"permission_read": 0x01,
	"permission_read_encrypted": 0x02,
	"permission_read_encrypted_mitm": 0x04,
	"permission_write": 0x10,
	"permission_write_encrypted": 0x20,
	"permission_write_encrypted_mitm": 0x40,
	"permission_write_signed": 0x80,
	"permission_write_signed_mitm": 0x100
	}

	# https://www.bluetooth.com/specifications/gatt/descriptors
	gatt_char_desc_uuids = {
	"char_ext_props": '00002900-0000-1000-8000-00805f9b34fb',
	"char_user_desc": '00002901-0000-1000-8000-00805f9b34fb',
	"client_char_cfg": '00002902-0000-1000-8000-00805f9b34fb',
	"server_char_cfg": '00002903-0000-1000-8000-00805f9b34fb',
	"char_fmt_uuid": '00002904-0000-1000-8000-00805f9b34fb',
	"char_agreg_fmt": '00002905-0000-1000-8000-00805f9b34fb',
	"char_valid_range": '00002906-0000-1000-8000-00805f9b34fb',
	"external_report_reference": '00002907-0000-1000-8000-00805f9b34fb',
	"report_reference": '00002908-0000-1000-8000-00805f9b34fb'
	}

	# https://www.bluetooth.com/specifications/gatt/characteristics
	gatt_char_types = {
	"device_name": '00002a00-0000-1000-8000-00805f9b34fb',
	"appearance": '00002a01-0000-1000-8000-00805f9b34fb',
	"peripheral_priv_flag": '00002a02-0000-1000-8000-00805f9b34fb',
	"reconnection_address": '00002a03-0000-1000-8000-00805f9b34fb',
	"peripheral_pref_conn": '00002a04-0000-1000-8000-00805f9b34fb',
	"service_changed": '00002a05-0000-1000-8000-00805f9b34fb',
	"system_id": '00002a23-0000-1000-8000-00805f9b34fb',
	"model_number_string": '00002a24-0000-1000-8000-00805f9b34fb',
	"serial_number_string": '00002a25-0000-1000-8000-00805f9b34fb',
	"firmware_revision_string": '00002a26-0000-1000-8000-00805f9b34fb',
	"hardware_revision_string": '00002a27-0000-1000-8000-00805f9b34fb',
	"software_revision_string": '00002a28-0000-1000-8000-00805f9b34fb',
	"manufacturer_name_string": '00002a29-0000-1000-8000-00805f9b34fb',
	"pnp_id": '00002a50-0000-1000-8000-00805f9b34fb',
	}

	# Matches constants of Bluetooth GATT Characteristic values as defined
	# in BluetoothGattCharacteristic.java
	gatt_characteristic_value_format = {
	"string": 0x1,
	"byte": 0x2,
	"sint8": 0x21,
	"uint8": 0x11,
	"sint16": 0x22,
	"unit16": 0x12,
	"sint32": 0x24,
	"uint32": 0x14
	}

	# Matches constants of Bluetooth Gatt Service types as defined in
	# BluetoothGattService.java
	gatt_service_types = {"primary": 0, "secondary": 1}

	# Matches constants of Bluetooth Gatt Connection Priority values as defined in
	# BluetoothGatt.java
	gatt_connection_priority = {"balanced": 0, "high": 1, "low_power": 2}

	# Min and max MTU values
	gatt_mtu_size = {"min": 23, "max": 217}

	# Gatt Characteristic attribute lengths
	gatt_characteristic_attr_length = {"attr_1": 1, "attr_2": 3, "attr_3": 15}

	# Matches constants of Bluetooth Gatt operations status as defined in
	# BluetoothGatt.java
	gatt_status = {"success": 0, "failure": 0x101}

	# Matches constants of Bluetooth transport values as defined in
	# BluetoothDevice.java
	gatt_transport = {"auto": 0x00, "bredr": 0x01, "le": 0x02}

	# Matches constants of Bluetooth physical channeling values as defined in
	# BluetoothDevice.java
	gatt_phy = {"1m": 1, "2m": 2, "le_coded": 3}

	# Matches constants of Bluetooth physical channeling bitmask values as defined
	# in BluetoothDevice.java
	gatt_phy_mask = {"1m_mask": 1, "2m_mask": 2, "coded_mask": 4}

	# Values as defiend in the Bluetooth GATT specification
	gatt_server_responses = {
	"GATT_SUCCESS": 0x0,
	"GATT_FAILURE": 0x1,
	"GATT_READ_NOT_PERMITTED": 0x2,
	"GATT_WRITE_NOT_PERMITTED": 0x3,
	"GATT_INVALID_PDU": 0x4,
	"GATT_INSUFFICIENT_AUTHENTICATION": 0x5,
	"GATT_REQUEST_NOT_SUPPORTED": 0x6,
	"GATT_INVALID_OFFSET": 0x7,
	"GATT_INSUFFICIENT_AUTHORIZATION": 0x8,
	"GATT_INVALID_ATTRIBUTE_LENGTH": 0xd,
	"GATT_INSUFFICIENT_ENCRYPTION": 0xf,
	"GATT_CONNECTION_CONGESTED": 0x8f,
	"GATT_13_ERR": 0x13,
	"GATT_12_ERR": 0x12,
	"GATT_0C_ERR": 0x0C,
	"GATT_16": 0x16
	}

	### Bluetooth GATT Constants End ###

	### Chameleon Constants Begin ###		### Chameleon Constants Begin ###

	# Chameleon audio bits per sample.		# Chameleon audio bits per sample.

system/blueberry/tests/gd_sl4a/lib/gd_sl4a_base_test.py

+6 −19

Original line number	Original line	Diff line number	Diff line
	@@ -26,11 +26,9 @@ from mobly.base_test import BaseTestClass
	from mobly.controllers.android_device_lib.adb import AdbError		from mobly.controllers.android_device_lib.adb import AdbError
	from mobly.controllers import android_device		from mobly.controllers import android_device
	from mobly.controllers.android_device import MOBLY_CONTROLLER_CONFIG_NAME as ANDROID_DEVICE_COFNIG_NAME		from mobly.controllers.android_device import MOBLY_CONTROLLER_CONFIG_NAME as ANDROID_DEVICE_COFNIG_NAME
	from mobly.controllers.android_device_lib.jsonrpc_client_base import \
	AppRestoreConnectionError
	from mobly.controllers.android_device_lib.services.sl4a_service import Sl4aService
	import mobly.controllers.android_device_lib.sl4a_client as sl4a_client

			from blueberry.utils.mobly_sl4a_utils import setup_sl4a
			from blueberry.utils.mobly_sl4a_utils import teardown_sl4a
	from blueberry.tests.gd.cert.context import get_current_context		from blueberry.tests.gd.cert.context import get_current_context
	from blueberry.tests.gd.cert.gd_device import MOBLY_CONTROLLER_CONFIG_NAME as GD_DEVICE_CONFIG_NAME		from blueberry.tests.gd.cert.gd_device import MOBLY_CONTROLLER_CONFIG_NAME as GD_DEVICE_CONFIG_NAME
	from blueberry.tests.gd_sl4a.lib.ble_lib import enable_bluetooth, disable_bluetooth, BleLib		from blueberry.tests.gd_sl4a.lib.ble_lib import enable_bluetooth, disable_bluetooth, BleLib
	@@ -54,25 +52,14 @@ class GdSl4aBaseTestClass(BaseTestClass):
	self.cert = self.gd_devices[0]		self.cert = self.gd_devices[0]

	# Parse and construct Android device objects		# Parse and construct Android device objects
			self.android_devices = self.register_controller(android_device, required=True)
	server_port = int(self.controller_configs[ANDROID_DEVICE_COFNIG_NAME][0]['server_port'])		server_port = int(self.controller_configs[ANDROID_DEVICE_COFNIG_NAME][0]['server_port'])
	forwarded_port = int(self.controller_configs[ANDROID_DEVICE_COFNIG_NAME][0]['forwarded_port'])		forwarded_port = int(self.controller_configs[ANDROID_DEVICE_COFNIG_NAME][0]['forwarded_port'])
	sl4a_client._DEVICE_SIDE_PORT = server_port
	sl4a_client._APP_START_WAIT_TIME = 0.5
	self.android_devices = self.register_controller(android_device, required=True)
	self.dut = self.android_devices[0]		self.dut = self.android_devices[0]
	self.dut.services.register('sl4a', Sl4aService, start_service=False)		setup_sl4a(self.dut, server_port, forwarded_port)
	try:
	self.dut.sl4a.start()
	except AppRestoreConnectionError:
	pass
	try:
	self.dut.sl4a.clear_host_port()
	except AdbError:
	pass
	sl4a_client._APP_START_WAIT_TIME = 2 * 60
	self.dut.sl4a.restore_app_connection(port=forwarded_port)

	# Enable full btsnoop log		# Enable full btsnoop log
			self.dut.adb.root()
	self.dut.adb.shell("setprop persist.bluetooth.btsnooplogmode full")		self.dut.adb.shell("setprop persist.bluetooth.btsnooplogmode full")
	getprop_result = self.dut.adb.shell("getprop persist.bluetooth.btsnooplogmode") == "full"		getprop_result = self.dut.adb.shell("getprop persist.bluetooth.btsnooplogmode") == "full"
	if not getprop_result:		if not getprop_result:
	@@ -81,7 +68,7 @@ class GdSl4aBaseTestClass(BaseTestClass):
	self.ble = BleLib(dut=self.dut)		self.ble = BleLib(dut=self.dut)

	def teardown_class(self):		def teardown_class(self):
	pass		teardown_sl4a(self.dut)

	def setup_test(self):		def setup_test(self):
	self.cert.rootservice.StartStack(		self.cert.rootservice.StartStack(

system/blueberry/tests/sl4a_sl4a/gatt/gatt_connect_test.py

+975 −110

File changed.

Preview size limit exceeded, changes collapsed.

system/blueberry/tests/sl4a_sl4a/gatt/gatt_connect_with_irk_test.py

0 → 100644

+148 −0

Original line number	Original line	Diff line number	Diff line
			#!/usr/bin/env python3
			#
			# Copyright 2021 - 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 io
			import logging
			import os
			import queue

			from blueberry.facade import common_pb2 as common
			from blueberry.tests.gd.cert.context import get_current_context
			from blueberry.tests.gd.cert.truth import assertThat
			from blueberry.tests.gd_sl4a.lib.ble_lib import generate_ble_advertise_objects
			from blueberry.tests.gd_sl4a.lib.ble_lib import generate_ble_scan_objects
			from blueberry.tests.gd_sl4a.lib.bt_constants import adv_succ
			from blueberry.tests.gd_sl4a.lib.bt_constants import ble_address_types
			from blueberry.tests.gd_sl4a.lib.bt_constants import ble_advertise_settings_modes
			from blueberry.tests.gd_sl4a.lib.bt_constants import ble_scan_settings_modes
			from blueberry.tests.gd_sl4a.lib.bt_constants import scan_result
			from blueberry.tests.sl4a_sl4a.lib import sl4a_sl4a_base_test
			from blueberry.utils.bt_gatt_constants import GattCallbackString
			from blueberry.utils.bt_gatt_constants import GattTransport
			from mobly import test_runner
			from mobly.controllers.android_device_lib.adb import AdbError


			class GattConnectWithIrkTest(sl4a_sl4a_base_test.Sl4aSl4aBaseTestClass):

			def setup_class(self):
			super().setup_class()
			self.default_timeout = 10 # seconds

			def setup_test(self):
			super().setup_test()

			def teardown_test(self):
			super().teardown_test()

			def _wait_for_event(self, expected_event_name, device):
			try:
			event_info = device.ed.pop_event(expected_event_name, self.default_timeout)
			logging.info(event_info)
			except queue.Empty as error:
			logging.error("Failed to find event: %s", expected_event_name)
			return False
			return True

			def _wait_for_scan_result_event(self, expected_event_name, device):
			try:
			event_info = device.ed.pop_event(expected_event_name, self.default_timeout)
			except queue.Empty as error:
			logging.error("Could not find scan result event: %s", expected_event_name)
			return None
			return event_info['data']['Result']['deviceInfo']['address']

			def _get_cert_public_address_and_irk_from_bt_config(self):
			# Pull IRK from SL4A cert side to pass in from SL4A DUT side when scanning
			bt_config_file_path = os.path.join(get_current_context().get_full_output_path(),
			"DUT_%s_bt_config.conf" % self.cert.serial)
			try:
			self.cert.adb.pull(["/data/misc/bluedroid/bt_config.conf", bt_config_file_path])
			except AdbError as error:
			logging.error("Failed to pull SL4A cert BT config")
			return False
			logging.debug("Reading SL4A cert BT config")
			with io.open(bt_config_file_path) as f:
			for line in f.readlines():
			stripped_line = line.strip()
			if (stripped_line.startswith("Address")):
			address_fields = stripped_line.split(' ')
			# API currently requires public address to be capitalized
			address = address_fields[2].upper()
			logging.debug("Found cert address: %s" % address)
			continue
			if (stripped_line.startswith("LE_LOCAL_KEY_IRK")):
			irk_fields = stripped_line.split(' ')
			irk = irk_fields[2]
			logging.debug("Found cert IRK: %s" % irk)
			continue

			return address, irk

			def test_scan_connect_unbonded_device_public_address_with_irk(self):
			# Set up SL4A cert side to advertise
			logging.info("Starting advertising")
			self.cert.sl4a.bleSetAdvertiseSettingsIsConnectable(True)
			self.cert.sl4a.bleSetAdvertiseDataIncludeDeviceName(True)
			self.cert.sl4a.bleSetAdvertiseSettingsAdvertiseMode(ble_advertise_settings_modes['low_latency'])
			self.cert.sl4a.bleSetAdvertiseSettingsOwnAddressType(common.RANDOM_DEVICE_ADDRESS)
			advertise_callback, advertise_data, advertise_settings = generate_ble_advertise_objects(self.cert.sl4a)
			self.cert.sl4a.bleStartBleAdvertising(advertise_callback, advertise_data, advertise_settings)

			# Wait for SL4A cert to start advertising
			assertThat(self._wait_for_event(adv_succ.format(advertise_callback), self.cert)).isTrue()
			logging.info("Advertising started")

			# Pull IRK from SL4A cert side to pass in from SL4A DUT side when scanning
			cert_public_address, irk = self._get_cert_public_address_and_irk_from_bt_config()

			# Set up SL4A DUT side to scan
			addr_type = ble_address_types["public"]
			logging.info("Start scanning for PUBLIC_ADDRESS %s with address type %d and IRK %s" % (cert_public_address,
			addr_type, irk))
			self.dut.sl4a.bleSetScanSettingsScanMode(ble_scan_settings_modes['low_latency'])
			self.dut.sl4a.bleSetScanSettingsLegacy(False)
			filter_list, scan_settings, scan_callback = generate_ble_scan_objects(self.dut.sl4a)
			expected_event_name = scan_result.format(scan_callback)

			# Start scanning on SL4A DUT
			self.dut.sl4a.bleSetScanFilterDeviceAddressTypeAndIrkHexString(cert_public_address, int(addr_type), irk)
			self.dut.sl4a.bleBuildScanFilter(filter_list)
			self.dut.sl4a.bleStartBleScan(filter_list, scan_settings, scan_callback)
			logging.info("Started scanning")

			# Verify that scan result is received on SL4A DUT
			mac_address = self._wait_for_scan_result_event(expected_event_name, self.dut)
			assertThat(mac_address).isNotNone()
			logging.info("Filter advertisement with address {}".format(mac_address))

			# Stop scanning and try to connect GATT
			self.dut.sl4a.bleStopBleScan(scan_callback)
			gatt_callback = self.dut.sl4a.gattCreateGattCallback()
			bluetooth_gatt = self.dut.sl4a.gattClientConnectGatt(gatt_callback, mac_address, False,
			GattTransport.TRANSPORT_LE, False, None)
			assertThat(bluetooth_gatt).isNotNone()

			# Verify that GATT connect event occurs on SL4A DUT
			expected_event_name = GattCallbackString.GATT_CONN_CHANGE.format(gatt_callback)
			assertThat(self._wait_for_event(expected_event_name, self.dut)).isTrue()

			# Test over
			self.cert.sl4a.bleStopBleAdvertising(advertise_callback)


			if __name__ == '__main__':
			test_runner.main()

system/blueberry/tests/sl4a_sl4a/gatt/gatt_notify_test.py

0 → 100644

+98 −0

Original line number	Original line	Diff line number	Diff line
			#!/usr/bin/env python3
			#
			# Copyright (C) 2016 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.
			"""
			This test script exercises GATT notify/indicate procedures.

			Original location:
			tools/test/connectivity/acts_tests/tests/google/ble/gatt/GattNotifyTest.py
			"""

			from blueberry.tests.gd.cert.test_decorators import test_tracker_info

			from blueberry.tests.gd.cert.truth import assertThat
			from blueberry.tests.sl4a_sl4a.lib import gatt_connected_base_test
			from blueberry.utils.bt_gatt_constants import GattCharDesc
			from blueberry.utils.bt_gatt_constants import GattDescriptor
			from blueberry.utils.bt_gatt_constants import GattEvent
			from mobly import test_runner


			class GattNotifyTest(gatt_connected_base_test.GattConnectedBaseTest):

			@test_tracker_info(uuid='e0ba60af-c1f2-4516-a5d5-89e2def0c757')
			def test_notify_char(self):
			"""Test notify characteristic value.

			Test GATT notify characteristic value.

			Steps:
			1. Central: write CCC - register for notifications.
			2. Peripheral: receive CCC modification.
			3. Peripheral: send characteristic notification.
			4. Central: receive notification, verify it's conent matches what was
			sent

			Expected Result:
			Verify that notification registration and delivery works.

			Returns:
			Pass if True
			Fail if False

			TAGS: LE, GATT, Characteristic
			Priority: 0
			"""
			# write CCC descriptor to enable notifications
			self.central.sl4a.gattClientDescriptorSetValue(
			self.bluetooth_gatt, self.discovered_services_index, self.test_service_index, self.NOTIFIABLE_CHAR_UUID,
			GattCharDesc.GATT_CLIENT_CHARAC_CFG_UUID, GattDescriptor.ENABLE_NOTIFICATION_VALUE)

			self.central.sl4a.gattClientWriteDescriptor(self.bluetooth_gatt, self.discovered_services_index,
			self.test_service_index, self.NOTIFIABLE_CHAR_UUID,
			GattCharDesc.GATT_CLIENT_CHARAC_CFG_UUID)

			# enable notifications in app
			self.central.sl4a.gattClientSetCharacteristicNotification(self.bluetooth_gatt, self.discovered_services_index,
			self.test_service_index, self.NOTIFIABLE_CHAR_UUID,
			True)

			event = self._server_wait(GattEvent.DESC_WRITE_REQ)

			request_id = event['data']['requestId']
			bt_device_id = 0
			status = 0
			# confirm notification registration was successful
			self.peripheral.sl4a.gattServerSendResponse(self.gatt_server, bt_device_id, request_id, status, 0, [])
			# wait for client to get response
			event = self._client_wait(GattEvent.DESC_WRITE)

			# set notified value
			notified_value = [1, 5, 9, 7, 5, 3, 6, 4, 8, 2]
			self.peripheral.sl4a.gattServerCharacteristicSetValue(self.notifiable_char_index, notified_value)

			# send notification
			self.peripheral.sl4a.gattServerNotifyCharacteristicChanged(self.gatt_server, bt_device_id,
			self.notifiable_char_index, False)

			# wait for client to receive the notification
			event = self._client_wait(GattEvent.CHAR_CHANGE)
			assertThat(event["data"]["CharacteristicValue"]).isEqualTo(notified_value)

			return True


			if __name__ == '__main__':
			test_runner.main()