Cleanup/Fixes/Logging/Removed Workarounds
Renamed Spp* tests to Rfcomm* tests.
Removed various workaround for fixed bugs.
Improved Gatt connection stress tests.
Improved Gatt connection tests by splitting
out it's own utils and error messages.
Added additional logging to key Gatt connection
tests.
Removed self.tests to necessary areas as now
all tests are called.
Set the default advertising size to 1 for all
Android devices.
Change-Id: I6b2bb25ed0d6293206b29ac643dbac7319cfc5a4
diff --git a/acts/framework/acts/controllers/native_android_device.py b/acts/framework/acts/controllers/native_android_device.py
index eac7bcc..4730f68 100644
--- a/acts/framework/acts/controllers/native_android_device.py
+++ b/acts/framework/acts/controllers/native_android_device.py
@@ -24,7 +24,7 @@
#TODO(tturney): Merge this into android device
-ACTS_CONTROLLER_CONFIG_NAME = "NativeAndroid"
+ACTS_CONTROLLER_CONFIG_NAME = "NativeAndroidDevice"
ACTS_CONTROLLER_REFERENCE_NAME = "native_android_devices"
def create(configs, logger):
@@ -126,4 +126,4 @@
"new session. Abort."))
return droid
self._droid_sessions[droid.uid] = [droid]
- return droid
\ No newline at end of file
+ return droid
diff --git a/acts/framework/acts/test_utils/bt/BleEnum.py b/acts/framework/acts/test_utils/bt/BleEnum.py
index d0db7f9..3982d09 100644
--- a/acts/framework/acts/test_utils/bt/BleEnum.py
+++ b/acts/framework/acts/test_utils/bt/BleEnum.py
@@ -80,73 +80,6 @@
HR_SERVICE = "0000180d-0000-1000-8000-00805f9b34fb"
-class GattConnectionState(Enum):
- STATE_DISCONNECTED = 0
- STATE_CONNECTING = 1
- STATE_CONNECTED = 2
- STATE_DISCONNECTING = 3
-
-
-class BluetoothGattCharacteristic(Enum):
- 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
- FORMAT_UINT8 = 0x11
- FORMAT_UINT16 = 0x12
- FORMAT_UINT32 = 0x14
- FORMAT_SINT8 = 0x21
- FORMAT_SINT16 = 0x22
- FORMAT_SINT32 = 0x24
- FORMAT_SFLOAT = 0x32
- FORMAT_FLOAT = 0x34
-
-
-class BluetoothGattDescriptor(Enum):
- 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
-
-
-class BluetoothGattService(Enum):
- SERVICE_TYPE_PRIMARY = 0
- SERVICE_TYPE_SECONDARY = 1
-
-
-class BluetoothGattConnectionPriority(Enum):
- CONNECTION_PRIORITY_BALANCED = 0
- CONNECTION_PRIORITY_HIGH = 1
- CONNECTION_PRIORITY_LOW_POWER = 2
-
-
-class BluetoothGatt(Enum):
- GATT_SUCCESS = 0
- GATT_FAILURE = 0x101
-
-
class AdvertiseErrorCode(Enum):
DATA_TOO_LARGE = 1
TOO_MANY_ADVERTISERS = 2
@@ -163,8 +96,3 @@
STATE_BLE_TURNING_ON = 14
STATE_BLE_ON = 15
STATE_BLE_TURNING_OFF = 16
-
-
-class BluetoothMtuSize(Enum):
- MIN = 23
- MAX = 217
diff --git a/acts/framework/acts/test_utils/bt/BtEnum.py b/acts/framework/acts/test_utils/bt/BtEnum.py
index ef33684..f6939f3 100644
--- a/acts/framework/acts/test_utils/bt/BtEnum.py
+++ b/acts/framework/acts/test_utils/bt/BtEnum.py
@@ -22,3 +22,13 @@
SCAN_MODE_NONE = 0
SCAN_MODE_CONNECTABLE = 1
SCAN_MODE_CONNECTABLE_DISCOVERABLE = 3
+
+
+class BluetoothAdapterState(Enum):
+ STATE_OFF = 10
+ STATE_TURNING_ON = 11
+ STATE_ON = 12
+ STATE_TURNING_OFF = 13
+ STATE_BLE_TURNING_ON = 14
+ STATE_BLE_ON = 15
+ STATE_BLE_TURNING_OFF = 16
diff --git a/acts/framework/acts/test_utils/bt/GattEnum.py b/acts/framework/acts/test_utils/bt/GattEnum.py
new file mode 100644
index 0000000..a9ee637
--- /dev/null
+++ b/acts/framework/acts/test_utils/bt/GattEnum.py
@@ -0,0 +1,113 @@
+#/usr/bin/env python3.4
+#
+# 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.
+
+from enum import Enum
+
+
+class GattCbErr(Enum):
+ 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 {}"
+ 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 {}"
+ GATT_CONN_CHANGE_ERR = "GATT Connection Changed event not found. Expected {}"
+
+
+class GattCbStrings(Enum):
+ CHAR_WRITE_REQ = "GattServer{}onCharacteristicWriteRequest"
+ CHAR_WRITE = "GattConnect{}onCharacteristicWrite"
+ DESC_WRITE_REQ = "GattServer{}onDescriptorWriteRequest"
+ DESC_WRITE = "GattConnect{}onDescriptorWrite"
+ RD_REMOTE_RSSI = "GattConnect{}onReadRemoteRssi"
+ GATT_SERV_DISC = "GattConnect{}onServicesDiscovered"
+ SERV_ADDED = "GattServer{}onServiceAdded"
+ MTU_CHANGED = "GattConnect{}onMtuChanged"
+ GATT_CONN_CHANGE = "GattConnect{}onConnectionStateChange"
+
+
+class GattConnectionState(Enum):
+ STATE_DISCONNECTED = 0
+ STATE_CONNECTING = 1
+ STATE_CONNECTED = 2
+ STATE_DISCONNECTING = 3
+
+
+class GattCharacteristic(Enum):
+ 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
+ FORMAT_UINT8 = 0x11
+ FORMAT_UINT16 = 0x12
+ FORMAT_UINT32 = 0x14
+ FORMAT_SINT8 = 0x21
+ FORMAT_SINT16 = 0x22
+ FORMAT_SINT32 = 0x24
+ FORMAT_SFLOAT = 0x32
+ FORMAT_FLOAT = 0x34
+
+
+class GattDescriptor(Enum):
+ 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
+
+
+class GattService(Enum):
+ SERVICE_TYPE_PRIMARY = 0
+ SERVICE_TYPE_SECONDARY = 1
+
+
+class GattConnectionPriority(Enum):
+ CONNECTION_PRIORITY_BALANCED = 0
+ CONNECTION_PRIORITY_HIGH = 1
+ CONNECTION_PRIORITY_LOW_POWER = 2
+
+
+class MtuSize(Enum):
+ MIN = 23
+ MAX = 217
+
+
+class BluetoothGatt(Enum):
+ GATT_SUCCESS = 0
+ GATT_FAILURE = 0x101
diff --git a/acts/framework/acts/test_utils/bt/bt_gatt_utils.py b/acts/framework/acts/test_utils/bt/bt_gatt_utils.py
new file mode 100644
index 0000000..d6f7bf1
--- /dev/null
+++ b/acts/framework/acts/test_utils/bt/bt_gatt_utils.py
@@ -0,0 +1,289 @@
+#/usr/bin/env python3.4
+#
+# 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.
+
+from acts.logger import LoggerProxy
+from acts.test_utils.bt.bt_test_utils import get_mac_address_of_generic_advertisement
+from acts.test_utils.bt.GattEnum import GattCbErr
+from acts.test_utils.bt.GattEnum import GattCbStrings
+from acts.test_utils.bt.GattEnum import GattConnectionState
+from acts.test_utils.bt.GattEnum import GattCharacteristic
+from acts.test_utils.bt.GattEnum import GattDescriptor
+from acts.test_utils.bt.GattEnum import GattService
+from acts.test_utils.bt.GattEnum import GattConnectionPriority
+import pprint
+from queue import Empty
+from contextlib import suppress
+
+default_timeout = 10
+log = LoggerProxy()
+
+
+def setup_gatt_connection(cen_ad, mac_address, autoconnect):
+ test_result = True
+ gatt_callback = cen_ad.droid.gattCreateGattCallback()
+ log.info("Gatt Connect to mac address {}.".format(mac_address))
+ bluetooth_gatt = cen_ad.droid.gattClientConnectGatt(
+ gatt_callback, mac_address, autoconnect)
+ expected_event = GattCbStrings.GATT_CONN_CHANGE.value.format(gatt_callback)
+ try:
+ event = cen_ad.ed.pop_event(expected_event, default_timeout)
+ except Empty:
+ log.error(GattCbErr.GATT_CONN_CHANGE_ERR.value.format(expected_event))
+ test_result = False
+ return test_result, bluetooth_gatt, gatt_callback
+ if event['data']['State'] != GattConnectionState.STATE_CONNECTED.value:
+ log.info("Could not establish a connection to peripheral. Event "
+ "Details:".format(pprint.pformat(event)))
+ test_result = False
+ return test_result, bluetooth_gatt, gatt_callback
+
+
+def disconnect_gatt_connection(cen_ad, bluetooth_gatt, gatt_callback):
+ cen_ad.droid.gattClientDisconnect(bluetooth_gatt)
+ expected_event = GattCbStrings.GATT_CONN_CHANGE.value.format(gatt_callback)
+ try:
+ event = cen_ad.ed.pop_event(expected_event, default_timeout)
+ except Empty:
+ log.error(GattCbErr.GATT_CONN_CHANGE_ERR.value.format(expected_event))
+ return False
+ if event['data']['State'] != GattConnectionState.STATE_DISCONNECTED.value:
+ return False
+ return True
+
+
+def orchestrate_gatt_connection(cen_ad, per_ad, le=True, mac_address=None):
+ adv_callback = None
+ if mac_address is None:
+ if le:
+ mac_address, adv_callback = (
+ get_mac_address_of_generic_advertisement(cen_ad, per_ad))
+ else:
+ mac_address = per_ad.droid.bluetoothGetLocalAddress()
+ adv_callback = None
+ autoconnect = False
+ test_result, bluetooth_gatt, gatt_callback = setup_gatt_connection(
+ cen_ad, mac_address, autoconnect)
+ if not test_result:
+ log.error("Could not connect to peripheral.")
+ return False, None, None
+ return bluetooth_gatt, gatt_callback, adv_callback
+
+
+def run_continuous_write_descriptor(cen_droid, cen_ed, per_droid, per_ed,
+ gatt_server, gatt_server_callback,
+ bluetooth_gatt, services_count,
+ discovered_services_index):
+ log.info("Starting continuous write")
+ bt_device_id = 0
+ status = 1
+ offset = 1
+ test_value = "1,2,3,4,5,6,7"
+ test_value_return = "1,2,3"
+ with suppress(Exception):
+ for x in range(100000):
+ for i in range(services_count):
+ characteristic_uuids = (
+ cen_droid.gattClientGetDiscoveredCharacteristicUuids(
+ discovered_services_index, i))
+ log.info(characteristic_uuids)
+ for characteristic in characteristic_uuids:
+ descriptor_uuids = (
+ cen_droid.gattClientGetDiscoveredDescriptorUuids(
+ discovered_services_index, i, characteristic))
+ log.info(descriptor_uuids)
+ for descriptor in descriptor_uuids:
+ log.info("descriptor to be written {}".format(
+ descriptor))
+ cen_droid.gattClientDescriptorSetValue(
+ bluetooth_gatt, discovered_services_index, i,
+ characteristic, descriptor, test_value)
+ cen_droid.gattClientWriteDescriptor(
+ bluetooth_gatt, discovered_services_index, i,
+ characteristic, descriptor)
+ expected_event = GattCbStrings.DESC_WRITE_REQ.value.format(
+ gatt_server_callback)
+ try:
+ event = per_ed.pop_event(expected_event,
+ default_timeout)
+ except Empty:
+ log.error(
+ GattCbErr.DESC_WRITE_REQ_ERR.value.format(
+ expected_event))
+ return False
+ request_id = event['data']['requestId']
+ found_value = event['data']['value']
+ if found_value != test_value:
+ log.error(
+ "Values didn't match. Found: {}, Expected: "
+ "{}".format(found_value, test_value))
+ per_droid.gattServerSendResponse(
+ gatt_server, bt_device_id, request_id, status,
+ offset, test_value_return)
+ expected_event = GattCbStrings.DESC_WRITE.value.format(
+ bluetooth_gatt)
+ try:
+ cen_ed.pop_event(expected_event, default_timeout)
+ except Empty:
+ log.error(GattCbErr.DESC_WRITE_ERR.value.format(
+ expected_event))
+ return False
+
+
+def setup_characteristics_and_descriptors(droid):
+ characteristic_input = [
+ {
+ 'uuid': "aa7edd5a-4d1d-4f0e-883a-d145616a1630",
+ 'property': GattCharacteristic.PROPERTY_WRITE.value
+ | GattCharacteristic.PROPERTY_WRITE_NO_RESPONSE.value,
+ 'permission': GattCharacteristic.PROPERTY_WRITE.value
+ },
+ {
+ 'uuid': "21c0a0bf-ad51-4a2d-8124-b74003e4e8c8",
+ 'property': GattCharacteristic.PROPERTY_NOTIFY.value
+ | GattCharacteristic.PROPERTY_READ.value,
+ 'permission': GattCharacteristic.PERMISSION_READ.value
+ },
+ {
+ 'uuid': "6774191f-6ec3-4aa2-b8a8-cf830e41fda6",
+ 'property': GattCharacteristic.PROPERTY_NOTIFY.value
+ | GattCharacteristic.PROPERTY_READ.value,
+ 'permission': GattCharacteristic.PERMISSION_READ.value
+ },
+ ]
+ descriptor_input = [
+ {
+ 'uuid': "aa7edd5a-4d1d-4f0e-883a-d145616a1630",
+ 'property': GattDescriptor.PERMISSION_READ.value
+ | GattDescriptor.PERMISSION_WRITE.value,
+ }, {
+ 'uuid': "76d5ed92-ca81-4edb-bb6b-9f019665fb32",
+ 'property': GattDescriptor.PERMISSION_READ.value
+ | GattCharacteristic.PERMISSION_WRITE.value,
+ }
+ ]
+ characteristic_list = setup_gatt_characteristics(droid,
+ characteristic_input)
+ descriptor_list = setup_gatt_descriptors(droid, descriptor_input)
+ return characteristic_list, descriptor_list
+
+
+def setup_multiple_services(per_ad):
+ per_droid, per_ed = per_ad.droid, per_ad.ed
+ gatt_server_callback = per_droid.gattServerCreateGattServerCallback()
+ gatt_server = per_droid.gattServerOpenGattServer(gatt_server_callback)
+ characteristic_list, descriptor_list = (
+ setup_characteristics_and_descriptors(per_droid))
+ per_droid.gattServerCharacteristicAddDescriptor(characteristic_list[1],
+ descriptor_list[0])
+ per_droid.gattServerCharacteristicAddDescriptor(characteristic_list[2],
+ descriptor_list[1])
+ gattService = per_droid.gattServerCreateService(
+ "00000000-0000-1000-8000-00805f9b34fb",
+ GattService.SERVICE_TYPE_PRIMARY.value)
+ gattService2 = per_droid.gattServerCreateService(
+ "FFFFFFFF-0000-1000-8000-00805f9b34fb",
+ GattService.SERVICE_TYPE_PRIMARY.value)
+ gattService3 = per_droid.gattServerCreateService(
+ "3846D7A0-69C8-11E4-BA00-0002A5D5C51B",
+ GattService.SERVICE_TYPE_PRIMARY.value)
+ for characteristic in characteristic_list:
+ per_droid.gattServerAddCharacteristicToService(gattService,
+ characteristic)
+ per_droid.gattServerAddService(gatt_server, gattService)
+ expected_event = GattCbStrings.SERV_ADDED.value.format(
+ gatt_server_callback)
+ try:
+ per_ed.pop_event(expected_event, default_timeout)
+ except Empty:
+ log.error(GattCbErr.SERV_ADDED_ERR.value.format(expected_event))
+ return False
+ for characteristic in characteristic_list:
+ per_droid.gattServerAddCharacteristicToService(gattService2,
+ characteristic)
+ per_droid.gattServerAddService(gatt_server, gattService2)
+ try:
+ per_ed.pop_event(expected_event, default_timeout)
+ except Empty:
+ log.error(GattCbErr.SERV_ADDED_ERR.value.format(expected_event))
+ return False
+ for characteristic in characteristic_list:
+ per_droid.gattServerAddCharacteristicToService(gattService3,
+ characteristic)
+ per_droid.gattServerAddService(gatt_server, gattService3)
+ try:
+ per_ed.pop_event(expected_event, default_timeout)
+ except Empty:
+ log.error(GattCbErr.SERV_ADDED_ERR.value.format(expected_event))
+ return False
+ return gatt_server_callback, gatt_server
+
+
+def setup_characteristics_and_descriptors(droid):
+ characteristic_input = [
+ {
+ 'uuid': "aa7edd5a-4d1d-4f0e-883a-d145616a1630",
+ 'property': GattCharacteristic.PROPERTY_WRITE.value
+ | GattCharacteristic.PROPERTY_WRITE_NO_RESPONSE.value,
+ 'permission': GattCharacteristic.PROPERTY_WRITE.value
+ },
+ {
+ 'uuid': "21c0a0bf-ad51-4a2d-8124-b74003e4e8c8",
+ 'property': GattCharacteristic.PROPERTY_NOTIFY.value
+ | GattCharacteristic.PROPERTY_READ.value,
+ 'permission': GattCharacteristic.PERMISSION_READ.value
+ },
+ {
+ 'uuid': "6774191f-6ec3-4aa2-b8a8-cf830e41fda6",
+ 'property': GattCharacteristic.PROPERTY_NOTIFY.value
+ | GattCharacteristic.PROPERTY_READ.value,
+ 'permission': GattCharacteristic.PERMISSION_READ.value
+ },
+ ]
+ descriptor_input = [
+ {
+ 'uuid': "aa7edd5a-4d1d-4f0e-883a-d145616a1630",
+ 'property': GattDescriptor.PERMISSION_READ.value
+ | GattDescriptor.PERMISSION_WRITE.value,
+ }, {
+ 'uuid': "76d5ed92-ca81-4edb-bb6b-9f019665fb32",
+ 'property': GattDescriptor.PERMISSION_READ.value
+ | GattCharacteristic.PERMISSION_WRITE.value,
+ }
+ ]
+ characteristic_list = setup_gatt_characteristics(droid,
+ characteristic_input)
+ descriptor_list = setup_gatt_descriptors(droid, descriptor_input)
+ return characteristic_list, descriptor_list
+
+
+def setup_gatt_characteristics(droid, input):
+ characteristic_list = []
+ for item in input:
+ index = droid.gattServerCreateBluetoothGattCharacteristic(
+ item['uuid'], item['property'], item['permission'])
+ characteristic_list.append(index)
+ return characteristic_list
+
+
+def setup_gatt_descriptors(droid, input):
+ descriptor_list = []
+ for item in input:
+ index = droid.gattServerCreateBluetoothGattDescriptor(
+ item['uuid'],
+ item['property'], )
+ descriptor_list.append(index)
+ log.info("setup descriptor list: {}".format(descriptor_list))
+ return descriptor_list
diff --git a/acts/framework/acts/test_utils/bt/bt_test_utils.py b/acts/framework/acts/test_utils/bt/bt_test_utils.py
index 6c0a80f..30a5e72 100644
--- a/acts/framework/acts/test_utils/bt/bt_test_utils.py
+++ b/acts/framework/acts/test_utils/bt/bt_test_utils.py
@@ -24,8 +24,19 @@
from contextlib2 import suppress
from acts.logger import LoggerProxy
-from acts.test_utils.bt.BleEnum import *
-from acts.test_utils.bt.BtEnum import *
+from acts.test_utils.bt.BleEnum import AdvertiseSettingsAdvertiseMode
+from acts.test_utils.bt.BleEnum import ScanSettingsCallbackType
+from acts.test_utils.bt.BleEnum import ScanSettingsMatchMode
+from acts.test_utils.bt.BleEnum import ScanSettingsMatchNum
+from acts.test_utils.bt.BleEnum import ScanSettingsScanResultType
+from acts.test_utils.bt.BleEnum import ScanSettingsScanMode
+from acts.test_utils.bt.BleEnum import ScanSettingsReportDelaySeconds
+from acts.test_utils.bt.BleEnum import AdvertiseSettingsAdvertiseType
+from acts.test_utils.bt.BleEnum import AdvertiseSettingsAdvertiseTxPower
+from acts.test_utils.bt.BleEnum import ScanSettingsMatchNum
+from acts.test_utils.bt.BleEnum import ScanSettingsScanResultType
+from acts.test_utils.bt.BleEnum import ScanSettingsScanMode
+from acts.test_utils.bt.BtEnum import BluetoothScanModeType
from acts.utils import exe_cmd
default_timeout = 10
@@ -34,21 +45,13 @@
log = LoggerProxy()
# Callback strings
-characteristic_write_request = "GattServer{}onCharacteristicWriteRequest"
-characteristic_write = "GattConnect{}onCharacteristicWrite"
-descriptor_write_request = "GattServer{}onDescriptorWriteRequest"
-descriptor_write = "GattConnect{}onDescriptorWrite"
-read_remote_rssi = "GattConnect{}onReadRemoteRssi"
-gatt_services_discovered = "GattConnect{}onServicesDiscovered"
scan_result = "BleScan{}onScanResults"
scan_failed = "BleScan{}onScanFailed"
-service_added = "GattServer{}onServiceAdded"
batch_scan_result = "BleScan{}onBatchScanResult"
adv_fail = "BleAdvertise{}onFailure"
adv_succ = "BleAdvertise{}onSuccess"
bluetooth_off = "BluetoothStateChangedOff"
bluetooth_on = "BluetoothStateChangedOn"
-mtu_changed = "GattConnect{}onMtuChanged"
# rfcomm test uuids
rfcomm_secure_uuid = "fa87c0d0-afac-11de-8a39-0800200c9a66"
@@ -197,7 +200,7 @@
ed.pop_event(expected_bluetooth_off_event_name,
default_timeout)
except Exception:
- log.info("Failed to toggle Bluetooth off.")
+ log.error("Failed to toggle Bluetooth off.")
return False
# temp sleep for b/17723234
time.sleep(3)
@@ -206,7 +209,7 @@
try:
ed.pop_event(expected_bluetooth_on_event_name, default_timeout)
except Exception:
- log.info("Failed to toggle Bluetooth on.")
+ log.error("Failed to toggle Bluetooth on.")
return False
return True
@@ -216,7 +219,7 @@
for a in android_devices:
d, e = a.droid, a.ed
model = d.getBuildModel()
- max_advertisements = 0
+ max_advertisements = 1
batch_scan_supported = True
if model in advertisements_to_devices.keys():
max_advertisements = advertisements_to_devices[model]
@@ -258,26 +261,6 @@
reset_bluetooth([adv_android_device])
-def setup_gatt_characteristics(droid, input):
- characteristic_list = []
- for item in input:
- index = droid.gattServerCreateBluetoothGattCharacteristic(
- item['uuid'], item['property'], item['permission'])
- characteristic_list.append(index)
- return characteristic_list
-
-
-def setup_gatt_descriptors(droid, input):
- descriptor_list = []
- for item in input:
- index = droid.gattServerCreateBluetoothGattDescriptor(
- item['uuid'],
- item['property'], )
- descriptor_list.append(index)
- log.info("setup descriptor list: {}".format(descriptor_list))
- return descriptor_list
-
-
def get_mac_address_of_generic_advertisement(scan_ad, adv_ad):
adv_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True)
adv_ad.droid.bleSetAdvertiseSettingsAdvertiseMode(
@@ -305,220 +288,6 @@
return mac_address, advertise_callback
-def setup_gatt_connection(cen_ad, mac_address, autoconnect):
- test_result = True
- gatt_callback = cen_ad.droid.gattCreateGattCallback()
- log.info("Gatt Connect to mac address {}.".format(mac_address))
- bluetooth_gatt = cen_ad.droid.gattClientConnectGatt(
- gatt_callback, mac_address, autoconnect)
- event = cen_ad.ed.pop_event(
- "GattConnect{}onConnectionStateChange".format(gatt_callback),
- default_timeout)
- if event['data']['State'] != GattConnectionState.STATE_CONNECTED.value:
- log.info("Could not establish a connection to peripheral. Event "
- "Details:".format(pprint.pformat(event)))
- test_result = False
- # To avoid race condition of quick connect/disconnect
- time.sleep(1)
- return test_result, bluetooth_gatt, gatt_callback
-
-
-def disconnect_gatt_connection(cen_ad, bluetooth_gatt, gatt_callback):
- cen_ad.droid.gattClientDisconnect(bluetooth_gatt)
- event = cen_ad.ed.pop_event(
- "GattConnect{}onConnectionStateChange".format(gatt_callback),
- default_timeout)
- if event['data']['State'] != GattConnectionState.STATE_DISCONNECTED.value:
- return False
- return True
-
-
-def orchestrate_gatt_connection(cen_ad, per_ad, le=True, mac_address=None):
- adv_callback = None
- if mac_address is None:
- if le:
- mac_address, adv_callback = (
- get_mac_address_of_generic_advertisement(cen_ad, per_ad))
- else:
- mac_address = get_bt_mac_address(cen_ad.droid, per_ad.droid, le)
- adv_callback = None
- autoconnect = False
- test_result, bluetooth_gatt, gatt_callback = setup_gatt_connection(
- cen_ad, mac_address, autoconnect)
- if not test_result:
- log.info("Could not connect to peripheral.")
- return False
- return bluetooth_gatt, gatt_callback, adv_callback
-
-
-def run_continuous_write_descriptor(cen_droid, cen_ed, per_droid, per_ed,
- gatt_server, gatt_server_callback,
- bluetooth_gatt, services_count,
- discovered_services_index):
- log.info("starting continuous write")
- bt_device_id = 0
- status = 1
- offset = 1
- test_value = "1,2,3,4,5,6,7"
- test_value_return = "1,2,3"
- with suppress(Exception):
- for x in range(100000):
- for i in range(services_count):
- characteristic_uuids = (
- cen_droid.gattClientGetDiscoveredCharacteristicUuids(
- discovered_services_index, i))
- log.info(characteristic_uuids)
- for characteristic in characteristic_uuids:
- descriptor_uuids = (
- cen_droid.gattClientGetDiscoveredDescriptorUuids(
- discovered_services_index, i, characteristic))
- log.info(descriptor_uuids)
- for descriptor in descriptor_uuids:
- log.info("descriptor to be written {}".format(
- descriptor))
- cen_droid.gattClientDescriptorSetValue(
- bluetooth_gatt, discovered_services_index, i,
- characteristic, descriptor, test_value)
- cen_droid.gattClientWriteDescriptor(
- bluetooth_gatt, discovered_services_index, i,
- characteristic, descriptor)
- event = per_ed.pop_event(
- descriptor_write_request.format(
- gatt_server_callback), default_timeout)
- log.info(
- "onDescriptorWriteRequest event found: {}".format(
- event))
- request_id = event['data']['requestId']
- found_value = event['data']['value']
- if found_value != test_value:
- log.info(
- "Values didn't match. Found: {}, Expected: "
- "{}".format(found_value, test_value))
- per_droid.gattServerSendResponse(
- gatt_server, bt_device_id, request_id, status,
- offset, test_value_return)
- log.info("onDescriptorWrite event found: {}".format(
- cen_ed.pop_event(
- descriptor_write.format(
- bluetooth_gatt), default_timeout)))
-
-
-def setup_characteristics_and_descriptors(droid):
- characteristic_input = [
- {
- 'uuid': "aa7edd5a-4d1d-4f0e-883a-d145616a1630",
- 'property': BluetoothGattCharacteristic.PROPERTY_WRITE.value
- | BluetoothGattCharacteristic.PROPERTY_WRITE_NO_RESPONSE.value,
- 'permission': BluetoothGattCharacteristic.PROPERTY_WRITE.value
- },
- {
- 'uuid': "21c0a0bf-ad51-4a2d-8124-b74003e4e8c8",
- 'property': BluetoothGattCharacteristic.PROPERTY_NOTIFY.value
- | BluetoothGattCharacteristic.PROPERTY_READ.value,
- 'permission': BluetoothGattCharacteristic.PERMISSION_READ.value
- },
- {
- 'uuid': "6774191f-6ec3-4aa2-b8a8-cf830e41fda6",
- 'property': BluetoothGattCharacteristic.PROPERTY_NOTIFY.value
- | BluetoothGattCharacteristic.PROPERTY_READ.value,
- 'permission': BluetoothGattCharacteristic.PERMISSION_READ.value
- },
- ]
- descriptor_input = [
- {
- 'uuid': "aa7edd5a-4d1d-4f0e-883a-d145616a1630",
- 'property': BluetoothGattDescriptor.PERMISSION_READ.value
- | BluetoothGattDescriptor.PERMISSION_WRITE.value,
- }, {
- 'uuid': "76d5ed92-ca81-4edb-bb6b-9f019665fb32",
- 'property': BluetoothGattDescriptor.PERMISSION_READ.value
- | BluetoothGattCharacteristic.PERMISSION_WRITE.value,
- }
- ]
- characteristic_list = setup_gatt_characteristics(droid,
- characteristic_input)
- descriptor_list = setup_gatt_descriptors(droid, descriptor_input)
- return characteristic_list, descriptor_list
-
-
-def setup_multiple_services(per_ad):
- per_droid, per_ed = per_ad.droid, per_ad.ed
- gatt_server_callback = per_droid.gattServerCreateGattServerCallback()
- gatt_server = per_droid.gattServerOpenGattServer(gatt_server_callback)
- characteristic_list, descriptor_list = (
- setup_characteristics_and_descriptors(per_droid))
- per_droid.gattServerCharacteristicAddDescriptor(characteristic_list[1],
- descriptor_list[0])
- per_droid.gattServerCharacteristicAddDescriptor(characteristic_list[2],
- descriptor_list[1])
- gattService = per_droid.gattServerCreateService(
- "00000000-0000-1000-8000-00805f9b34fb",
- BluetoothGattService.SERVICE_TYPE_PRIMARY.value)
- gattService2 = per_droid.gattServerCreateService(
- "FFFFFFFF-0000-1000-8000-00805f9b34fb",
- BluetoothGattService.SERVICE_TYPE_PRIMARY.value)
- gattService3 = per_droid.gattServerCreateService(
- "3846D7A0-69C8-11E4-BA00-0002A5D5C51B",
- BluetoothGattService.SERVICE_TYPE_PRIMARY.value)
- for characteristic in characteristic_list:
- per_droid.gattServerAddCharacteristicToService(gattService,
- characteristic)
- per_droid.gattServerAddService(gatt_server, gattService)
- per_ed.pop_event(
- service_added.format(gatt_server_callback), default_timeout)
- for characteristic in characteristic_list:
- per_droid.gattServerAddCharacteristicToService(gattService2,
- characteristic)
- per_droid.gattServerAddService(gatt_server, gattService2)
- per_ed.pop_event(
- service_added.format(gatt_server_callback), default_timeout)
- for characteristic in characteristic_list:
- per_droid.gattServerAddCharacteristicToService(gattService3,
- characteristic)
- per_droid.gattServerAddService(gatt_server, gattService3)
- per_ed.pop_event(
- service_added.format(gatt_server_callback), default_timeout)
- return gatt_server_callback, gatt_server
-
-
-def setup_characteristics_and_descriptors(droid):
- characteristic_input = [
- {
- 'uuid': "aa7edd5a-4d1d-4f0e-883a-d145616a1630",
- 'property': BluetoothGattCharacteristic.PROPERTY_WRITE.value
- | BluetoothGattCharacteristic.PROPERTY_WRITE_NO_RESPONSE.value,
- 'permission': BluetoothGattCharacteristic.PROPERTY_WRITE.value
- },
- {
- 'uuid': "21c0a0bf-ad51-4a2d-8124-b74003e4e8c8",
- 'property': BluetoothGattCharacteristic.PROPERTY_NOTIFY.value
- | BluetoothGattCharacteristic.PROPERTY_READ.value,
- 'permission': BluetoothGattCharacteristic.PERMISSION_READ.value
- },
- {
- 'uuid': "6774191f-6ec3-4aa2-b8a8-cf830e41fda6",
- 'property': BluetoothGattCharacteristic.PROPERTY_NOTIFY.value
- | BluetoothGattCharacteristic.PROPERTY_READ.value,
- 'permission': BluetoothGattCharacteristic.PERMISSION_READ.value
- },
- ]
- descriptor_input = [
- {
- 'uuid': "aa7edd5a-4d1d-4f0e-883a-d145616a1630",
- 'property': BluetoothGattDescriptor.PERMISSION_READ.value
- | BluetoothGattDescriptor.PERMISSION_WRITE.value,
- }, {
- 'uuid': "76d5ed92-ca81-4edb-bb6b-9f019665fb32",
- 'property': BluetoothGattDescriptor.PERMISSION_READ.value
- | BluetoothGattCharacteristic.PERMISSION_WRITE.value,
- }
- ]
- characteristic_list = setup_gatt_characteristics(droid,
- characteristic_input)
- descriptor_list = setup_gatt_descriptors(droid, descriptor_input)
- return characteristic_list, descriptor_list
-
-
def get_device_local_info(droid):
local_info_dict = {}
local_info_dict['name'] = droid.bluetoothGetLocalName()
@@ -624,8 +393,8 @@
expected_address = sec_droid.bluetoothGetLocalAddress()
bonded = False
for d in bonded_devices:
- if d['address'] == expected_address:
- return True
+ if d['address'] == expected_address:
+ return True
time.sleep(1)
# Timed out trying to bond.
return False
diff --git a/acts/tests/google/ble/api/BleScanApiTest.py b/acts/tests/google/ble/api/BleScanApiTest.py
index a0029ee..bb93cd8 100644
--- a/acts/tests/google/ble/api/BleScanApiTest.py
+++ b/acts/tests/google/ble/api/BleScanApiTest.py
@@ -46,50 +46,10 @@
class BleScanApiTest(BluetoothBaseTest):
- tests = None
def __init__(self, controllers):
BluetoothBaseTest.__init__(self, controllers)
self.droid = self.android_devices[0].droid
- self.tests = (
- "test_start_ble_scan_with_default_settings",
- "test_stop_ble_scan_default_settings",
- "test_scan_settings_callback_type_all_matches",
- "test_scan_settings_set_callback_type_first_match",
- "test_scan_settings_set_callback_type_match_lost",
- "test_scan_settings_set_invalid_callback_type",
- "test_scan_settings_set_scan_mode_low_power",
- "test_scan_settings_set_scan_mode_balanced",
- "test_scan_settings_set_scan_mode_low_latency",
- "test_scan_settings_set_invalid_scan_mode",
- "test_scan_settings_set_report_delay_millis_min",
- "test_scan_settings_set_report_delay_millis_min_plus_one",
- "test_scan_settings_set_report_delay_millis_max",
- "test_scan_settings_set_report_delay_millis_max_minus_one",
- "test_scan_settings_set_invalid_report_delay_millis_min_minus_one",
- "test_scan_settings_set_scan_result_type_full",
- "test_scan_settings_set_scan_result_type_abbreviated",
- "test_scan_settings_set_invalid_scan_result_type",
- "test_scan_filter_set_device_name",
- "test_scan_filter_set_device_name_blank",
- "test_scan_filter_set_device_name_special_chars",
- "test_scan_filter_set_device_address",
- "test_scan_filter_set_invalid_device_address_lower_case",
- "test_scan_filter_set_invalid_device_address_blank",
- "test_scan_filter_set_invalid_device_address_bad_format",
- "test_scan_filter_set_invalid_device_address_bad_address",
- "test_scan_filter_set_manufacturer_id_data",
- "test_scan_filter_set_manufacturer_id_data_mask",
- "test_scan_filter_set_manufacturer_max_id",
- "test_scan_filter_set_manufacturer_data_empty",
- "test_scan_filter_set_manufacturer_data_mask_empty",
- "test_scan_filter_set_invalid_manufacturer_min_id_minus_one",
- "test_scan_filter_set_service_uuid",
- "test_scan_filter_service_uuid_p_service",
- "test_classic_ble_scan_with_service_uuids_p",
- "test_classic_ble_scan_with_service_uuids_hr",
- "test_classic_ble_scan_with_service_uuids_empty_uuid_list",
- "test_classic_ble_scan_with_service_uuids_hr_and_p", )
def _format_defaults(self, input):
"""
diff --git a/acts/tests/google/ble/api/GattApiTest.py b/acts/tests/google/ble/api/GattApiTest.py
index 78f4ee8..4edee1e 100644
--- a/acts/tests/google/ble/api/GattApiTest.py
+++ b/acts/tests/google/ble/api/GattApiTest.py
@@ -24,14 +24,10 @@
class GattApiTest(BluetoothBaseTest):
- tests = None
def __init__(self, controllers):
BluetoothBaseTest.__init__(self, controllers)
self.ad = self.android_devices[0]
- self.tests = ("test_open_gatt_server",
- "test_open_gatt_server_on_same_callback",
- "test_open_gatt_server_on_invalid_callback", )
def setup_class(self):
return setup_multiple_devices_for_bt_test(self.android_devices)
diff --git a/acts/tests/google/ble/beacon_tests/BeaconSwarmTest.py b/acts/tests/google/ble/beacon_tests/BeaconSwarmTest.py
index 8a35d0e..fa51d82 100644
--- a/acts/tests/google/ble/beacon_tests/BeaconSwarmTest.py
+++ b/acts/tests/google/ble/beacon_tests/BeaconSwarmTest.py
@@ -37,7 +37,6 @@
class BeaconSwarmTest(BluetoothBaseTest):
- tests = None
default_timeout = 10
beacon_swarm_count = 0
advertising_device_name_list = []
@@ -46,10 +45,6 @@
def __init__(self, controllers):
BluetoothBaseTest.__init__(self, controllers)
self.scn_ad = self.android_devices[0]
- self.tests = ("test_swarm_1000_on_scan_result",
- "test_swarm_10000_on_batch_scan_result",
- "test_swarm_rotate_addresses",
- "test_swarm_scan_result_filter_each_device_name", )
def setup_test(self):
self.log.debug(log_energy_info(self.android_devices, "Start"))
diff --git a/acts/tests/google/ble/concurrency/ConcurrentBleAdvertisingTest.py b/acts/tests/google/ble/concurrency/ConcurrentBleAdvertisingTest.py
index 26ac084..57ef8a7 100644
--- a/acts/tests/google/ble/concurrency/ConcurrentBleAdvertisingTest.py
+++ b/acts/tests/google/ble/concurrency/ConcurrentBleAdvertisingTest.py
@@ -37,7 +37,6 @@
class ConcurrentBleAdvertisingTest(BluetoothBaseTest):
- tests = None
default_timeout = 10
max_advertisements = 4
@@ -50,20 +49,6 @@
if self.max_advertisements == 0:
self.tests = ()
return
- self.tests = (
- "test_max_advertisements_defaults",
- "test_max_advertisements_include_device_name_and_filter_device_name",
- "test_max_advertisements_exclude_device_name_and_filter_device_name",
- "test_max_advertisements_with_manufacturer_data",
- "test_max_advertisements_with_manufacturer_data_mask",
- "test_max_advertisements_with_service_data",
- "test_max_advertisements_with_manufacturer_data_mask_and_include_device_name",
- "test_max_advertisements_with_service_uuids",
- "test_max_advertisements_with_service_uuid_and_service_mask",
- "test_max_advertisements_plus_one",
- "test_start_two_advertisements_on_same_callback",
- "test_toggle_advertiser_bt_state",
- "test_restart_advertise_callback_after_bt_toggle", )
def on_fail(self, test_name, begin_time):
self.log.debug(
diff --git a/acts/tests/google/ble/concurrency/ConcurrentBleScanningTest.py b/acts/tests/google/ble/concurrency/ConcurrentBleScanningTest.py
index 9b64e10..dc95269 100644
--- a/acts/tests/google/ble/concurrency/ConcurrentBleScanningTest.py
+++ b/acts/tests/google/ble/concurrency/ConcurrentBleScanningTest.py
@@ -47,11 +47,6 @@
if self.droid_list[1]['max_advertisements'] == 0:
self.tests = ("test_max_concurrent_ble_scans_plus_one", )
return
- self.tests = (
- "test_max_concurrent_ble_scans",
- "test_max_concurrent_ble_scans_then_discover_advertisement",
- "test_max_concurrent_ble_scans_plus_one",
- "test_max_concurrent_ble_scans_verify_scans_stop_independently", )
def on_fail(self, test_name, begin_time):
self.log.debug("Test {} failed. Gathering bugreport and btsnoop logs."
diff --git a/acts/tests/google/ble/examples/BleExamplesTest.py b/acts/tests/google/ble/examples/BleExamplesTest.py
index 37912de..7691a00 100644
--- a/acts/tests/google/ble/examples/BleExamplesTest.py
+++ b/acts/tests/google/ble/examples/BleExamplesTest.py
@@ -29,7 +29,6 @@
class BleExamplesTest(BluetoothBaseTest):
- tests = None
default_timeout = 10
active_scan_callback_list = []
active_adv_callback_list = []
@@ -46,7 +45,6 @@
if self.droid_list[1]['max_advertisements'] == 0:
self.tests = ()
return
- self.tests = ("test_bt_toggle", )
def teardown_test(self):
cleanup_scanners_and_advertisers(
diff --git a/acts/tests/google/ble/filtering/FilteringTest.py b/acts/tests/google/ble/filtering/FilteringTest.py
index 6257ae9..61324cd 100644
--- a/acts/tests/google/ble/filtering/FilteringTest.py
+++ b/acts/tests/google/ble/filtering/FilteringTest.py
@@ -34,7 +34,6 @@
class FilteringTest(BluetoothBaseTest):
- tests = None
default_timeout = 30
valid_filter_suite = [
@@ -171,13 +170,6 @@
self.scn_ad.droid.getBuildModel()))
self.log.info("Advertiser device model: {}".format(
self.adv_ad.droid.getBuildModel()))
- self.tests = ("test_valid_filters",
- "test_valid_filters_opportunistic_scan",
- "test_default_advertisement",
- "test_settings_in_effect_suite",
- "test_filters_suite",
- "test_filters_suite_opportunistic_scan",
- "test_non_connectable_advertise_data", )
def blescan_verify_onfailure_event_handler(self, event):
self.log.debug("Verifying {} event".format(adv_fail))
diff --git a/acts/tests/google/ble/filtering/UniqueFilteringTest.py b/acts/tests/google/ble/filtering/UniqueFilteringTest.py
index 5e3a702..6c575aa 100644
--- a/acts/tests/google/ble/filtering/UniqueFilteringTest.py
+++ b/acts/tests/google/ble/filtering/UniqueFilteringTest.py
@@ -34,7 +34,6 @@
class UniqueFilteringTest(BluetoothBaseTest):
- tests = None
default_timeout = 10
def __init__(self, controllers):
diff --git a/acts/tests/google/ble/gatt/GattConnectTest.py b/acts/tests/google/ble/gatt/GattConnectTest.py
index 3bee22c..3cc3ce9 100644
--- a/acts/tests/google/ble/gatt/GattConnectTest.py
+++ b/acts/tests/google/ble/gatt/GattConnectTest.py
@@ -23,30 +23,23 @@
from contextlib import suppress
from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
-from acts.test_utils.bt.BleEnum import BluetoothGattCharacteristic
-from acts.test_utils.bt.BleEnum import BluetoothGattDescriptor
-from acts.test_utils.bt.BleEnum import BluetoothGattService
-from acts.test_utils.bt.BleEnum import BluetoothMtuSize
-from acts.test_utils.bt.bt_test_utils import characteristic_write_request
-from acts.test_utils.bt.bt_test_utils import characteristic_write
-from acts.test_utils.bt.bt_test_utils import descriptor_write
-from acts.test_utils.bt.bt_test_utils import descriptor_write_request
-from acts.test_utils.bt.bt_test_utils import disconnect_gatt_connection
-from acts.test_utils.bt.bt_test_utils import gatt_services_discovered
+from acts.test_utils.bt.GattEnum import GattCharacteristic
+from acts.test_utils.bt.GattEnum import GattDescriptor
+from acts.test_utils.bt.GattEnum import GattService
+from acts.test_utils.bt.GattEnum import MtuSize
+from acts.test_utils.bt.GattEnum import GattCbErr
+from acts.test_utils.bt.GattEnum import GattCbStrings
+from acts.test_utils.bt.bt_gatt_utils import disconnect_gatt_connection
+from acts.test_utils.bt.bt_gatt_utils import orchestrate_gatt_connection
+from acts.test_utils.bt.bt_gatt_utils import setup_gatt_characteristics
+from acts.test_utils.bt.bt_gatt_utils import setup_gatt_connection
+from acts.test_utils.bt.bt_gatt_utils import setup_gatt_descriptors
from acts.test_utils.bt.bt_test_utils import get_advanced_droid_list
from acts.test_utils.bt.bt_test_utils import get_mac_address_of_generic_advertisement
-from acts.test_utils.bt.bt_test_utils import mtu_changed
-from acts.test_utils.bt.bt_test_utils import orchestrate_gatt_connection
-from acts.test_utils.bt.bt_test_utils import read_remote_rssi
-from acts.test_utils.bt.bt_test_utils import service_added
-from acts.test_utils.bt.bt_test_utils import setup_gatt_characteristics
-from acts.test_utils.bt.bt_test_utils import setup_gatt_connection
-from acts.test_utils.bt.bt_test_utils import setup_gatt_descriptors
from acts.test_utils.bt.bt_test_utils import log_energy_info
class GattConnectTest(BluetoothBaseTest):
- tests = None
adv_instances = []
default_timeout = 10
default_discovery_timeout = 3
@@ -60,18 +53,6 @@
if self.droid_list[1]['max_advertisements'] == 0:
self.tests = ()
return
- self.tests = (
- "test_gatt_connect",
- "test_gatt_request_min_mtu",
- "test_gatt_request_max_mtu",
- "test_gatt_request_out_of_bounds_mtu",
- "test_gatt_connect_trigger_on_read_rssi",
- "test_gatt_connect_trigger_on_services_discovered",
- "test_gatt_connect_trigger_on_services_discovered_iterate_attributes",
- "test_gatt_connect_with_service_uuid_variations",
- "test_gatt_connect_in_quick_succession",
- "test_write_descriptor_stress",
- "test_gatt_connect_mitm_attack", )
def teardown_test(self):
for adv in self.adv_instances:
@@ -83,32 +64,32 @@
characteristic_input = [
{
'uuid': "aa7edd5a-4d1d-4f0e-883a-d145616a1630",
- 'property': BluetoothGattCharacteristic.PROPERTY_WRITE.value
- | BluetoothGattCharacteristic.PROPERTY_WRITE_NO_RESPONSE.value,
- 'permission': BluetoothGattCharacteristic.PROPERTY_WRITE.value
+ 'property': GattCharacteristic.PROPERTY_WRITE.value
+ | GattCharacteristic.PROPERTY_WRITE_NO_RESPONSE.value,
+ 'permission': GattCharacteristic.PROPERTY_WRITE.value
},
{
'uuid': "21c0a0bf-ad51-4a2d-8124-b74003e4e8c8",
- 'property': BluetoothGattCharacteristic.PROPERTY_NOTIFY.value
- | BluetoothGattCharacteristic.PROPERTY_READ.value,
- 'permission': BluetoothGattCharacteristic.PERMISSION_READ.value
+ 'property': GattCharacteristic.PROPERTY_NOTIFY.value
+ | GattCharacteristic.PROPERTY_READ.value,
+ 'permission': GattCharacteristic.PERMISSION_READ.value
},
{
'uuid': "6774191f-6ec3-4aa2-b8a8-cf830e41fda6",
- 'property': BluetoothGattCharacteristic.PROPERTY_NOTIFY.value
- | BluetoothGattCharacteristic.PROPERTY_READ.value,
- 'permission': BluetoothGattCharacteristic.PERMISSION_READ.value
+ 'property': GattCharacteristic.PROPERTY_NOTIFY.value
+ | GattCharacteristic.PROPERTY_READ.value,
+ 'permission': GattCharacteristic.PERMISSION_READ.value
},
]
descriptor_input = [
{
'uuid': "aa7edd5a-4d1d-4f0e-883a-d145616a1630",
- 'property': BluetoothGattDescriptor.PERMISSION_READ.value
- | BluetoothGattDescriptor.PERMISSION_WRITE.value,
+ 'property': GattDescriptor.PERMISSION_READ.value
+ | GattDescriptor.PERMISSION_WRITE.value,
}, {
'uuid': "76d5ed92-ca81-4edb-bb6b-9f019665fb32",
- 'property': BluetoothGattDescriptor.PERMISSION_READ.value
- | BluetoothGattCharacteristic.PERMISSION_WRITE.value,
+ 'property': GattDescriptor.PERMISSION_READ.value
+ | GattCharacteristic.PERMISSION_WRITE.value,
}
]
characteristic_list = setup_gatt_characteristics(droid,
@@ -120,6 +101,7 @@
self.log.info("Disconnecting from peripheral device.")
test_result = disconnect_gatt_connection(self.cen_ad, bluetooth_gatt,
gatt_callback)
+ self.cen_ad.droid.gattClientClose(gatt_callback)
if not test_result:
self.log.info("Failed to disconnect from peripheral device.")
return False
@@ -146,10 +128,17 @@
descriptor))
def _find_service_added_event(self, gatt_server_callback, uuid):
- event = self.per_ad.ed.pop_event(
- service_added.format(gatt_server_callback), self.default_timeout)
+ expected_event = GattCbStrings.SERV_ADDED.value.format(
+ gatt_server_callback)
+ try:
+ event = self.per_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ except Empty:
+ self.log.error(GattCbErr.SERV_ADDED_ERR.value.format(
+ expected_event))
+ return False
if event['data']['serviceUuid'].lower() != uuid.lower():
- self.log.info("Uuid mismatch. Found: {}, Expected {}.".format(
+ self.log.error("Uuid mismatch. Found: {}, Expected {}.".format(
event['data']['serviceUuid'], uuid))
return False
return True
@@ -167,13 +156,13 @@
characteristic_list[2], descriptor_list[1])
gatt_service = self.per_ad.droid.gattServerCreateService(
"00000000-0000-1000-8000-00805f9b34fb",
- BluetoothGattService.SERVICE_TYPE_PRIMARY.value)
+ GattService.SERVICE_TYPE_PRIMARY.value)
gatt_service2 = self.per_ad.droid.gattServerCreateService(
"FFFFFFFF-0000-1000-8000-00805f9b34fb",
- BluetoothGattService.SERVICE_TYPE_PRIMARY.value)
+ GattService.SERVICE_TYPE_PRIMARY.value)
gatt_service3 = self.per_ad.droid.gattServerCreateService(
"3846D7A0-69C8-11E4-BA00-0002A5D5C51B",
- BluetoothGattService.SERVICE_TYPE_PRIMARY.value)
+ GattService.SERVICE_TYPE_PRIMARY.value)
for characteristic in characteristic_list:
self.per_ad.droid.gattServerAddCharacteristicToService(
gatt_service, characteristic)
@@ -265,10 +254,19 @@
orchestrate_gatt_connection(self.cen_ad, self.per_ad))
self.adv_instances.append(adv_callback)
self.cen_ad.droid.gattClientRequestMtu(bluetooth_gatt,
- BluetoothMtuSize.MIN.value)
- mtu_event = self.cen_ad.ed.pop_event(mtu_changed.format(
- bluetooth_gatt))
- if mtu_event['data']['MTU'] != BluetoothMtuSize.MIN.value:
+ MtuSize.MIN.value)
+ expected_event = GattCbStrings.MTU_CHANGED.value.format(bluetooth_gatt)
+ try:
+ mtu_event = self.cen_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ mtu_size_found = mtu_event['data']['MTU']
+ if mtu_size_found != MtuSize.MIN.value:
+ self.log.error("MTU size found: {}, expected: {}".format(
+ mtu_size_found, MtuSize.MIN.value))
+ return False
+ except Empty:
+ self.log.error(GattCbErr.MTU_CHANGED_ERR.value.format(
+ expected_event))
return False
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -306,10 +304,19 @@
orchestrate_gatt_connection(self.cen_ad, self.per_ad))
self.adv_instances.append(adv_callback)
self.cen_ad.droid.gattClientRequestMtu(bluetooth_gatt,
- BluetoothMtuSize.MAX.value)
- mtu_event = self.cen_ad.ed.pop_event(mtu_changed.format(
- bluetooth_gatt))
- if mtu_event['data']['MTU'] != BluetoothMtuSize.MAX.value:
+ MtuSize.MAX.value)
+ expected_event = GattCbStrings.MTU_CHANGED.value.format(bluetooth_gatt)
+ try:
+ mtu_event = self.cen_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ mtu_size_found = mtu_event['data']['MTU']
+ if mtu_size_found != MtuSize.MAX.value:
+ self.log.error("MTU size found: {}, expected: {}".format(
+ mtu_size_found, MtuSize.MAX.value))
+ return False
+ except Empty:
+ self.log.error(GattCbErr.MTU_CHANGED_ERR.value.format(
+ expected_event))
return False
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -348,16 +355,16 @@
orchestrate_gatt_connection(self.cen_ad, self.per_ad))
self.adv_instances.append(adv_callback)
self.cen_ad.droid.gattClientRequestMtu(bluetooth_gatt,
- BluetoothMtuSize.MIN.value - 1)
+ MtuSize.MIN.value - 1)
+ expected_event = GattCbStrings.MTU_CHANGED.value.format(bluetooth_gatt)
try:
- self.cen_ad.ed.pop_event(
- mtu_changed.format(bluetooth_gatt), self.default_timeout)
+ self.cen_ad.ed.pop_event(expected_event, self.default_timeout)
self.log.error("Found {} event when it wasn't expected".format(
- mtu_changed.format(bluetooth_gatt)))
+ expected_event))
return False
- except Exception:
- self.log.debug("Successfully didn't find {} event"
- .format(mtu_changed.format(bluetooth_gatt)))
+ except Empty:
+ self.log.debug("Successfully didn't find {} event".format(
+ expected_event))
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -391,9 +398,14 @@
bluetooth_gatt, gatt_callback, adv_callback = (
orchestrate_gatt_connection(self.cen_ad, self.per_ad))
self.adv_instances.append(adv_callback)
+ expected_event = GattCbStrings.RD_REMOTE_RSSI.value.format(
+ gatt_callback)
if self.cen_ad.droid.gattClientReadRSSI(bluetooth_gatt):
- self.cen_ad.ed.pop_event(
- read_remote_rssi.format(gatt_callback), self.default_timeout)
+ try:
+ self.cen_ad.ed.pop_event(expected_event, self.default_timeout)
+ except Empty:
+ self.log.error(GattCbErr.RD_REMOTE_RSSI_ERR.value.format(
+ expected_event))
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -428,9 +440,15 @@
orchestrate_gatt_connection(self.cen_ad, self.per_ad))
self.adv_instances.append(adv_callback)
if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
- event = self.cen_ad.ed.pop_event(
- gatt_services_discovered.format(gatt_callback),
- self.default_timeout)
+ expected_event = GattCbStrings.GATT_SERV_DISC.value.format(
+ gatt_callback)
+ try:
+ event = self.cen_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ except Empty:
+ self.log.error(GattCbErr.GATT_SERV_DISC_ERR.value.format(
+ expected_event))
+ return False
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -471,10 +489,16 @@
orchestrate_gatt_connection(self.cen_ad, self.per_ad))
self.adv_instances.append(adv_callback)
if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
- event = self.cen_ad.ed.pop_event(
- gatt_services_discovered.format(gatt_callback),
- self.default_timeout)
- discovered_services_index = event['data']['ServicesIndex']
+ expected_event = GattCbStrings.GATT_SERV_DISC.value.format(
+ gatt_callback)
+ try:
+ event = self.cen_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ discovered_services_index = event['data']['ServicesIndex']
+ except Empty:
+ self.log.error(GattCbErr.GATT_SERV_DISC_ERR.value.format(
+ expected_event))
+ return False
self._iterate_attributes(discovered_services_index)
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -514,9 +538,15 @@
orchestrate_gatt_connection(self.cen_ad, self.per_ad))
self.adv_instances.append(adv_callback)
if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
- event = self.cen_ad.ed.pop_event(
- gatt_services_discovered.format(gatt_callback),
- self.default_timeout)
+ expected_event = GattCbStrings.GATT_SERV_DISC.value.format(
+ gatt_callback)
+ try:
+ event = self.cen_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ except Empty:
+ self.log.error(GattCbErr.GATT_SERV_DISC_ERR.value.format(
+ expected_event))
+ return False
discovered_services_index = event['data']['ServicesIndex']
self._iterate_attributes(discovered_services_index)
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
@@ -552,7 +582,8 @@
mac_address, adv_callback = get_mac_address_of_generic_advertisement(
self.cen_ad, self.per_ad)
autoconnect = False
- for i in range(20):
+ for i in range(1000):
+ self.log.info("Starting connection iteration {}".format(i + 1))
test_result, bluetooth_gatt, gatt_callback = setup_gatt_connection(
self.cen_ad, mac_address, autoconnect)
if not test_result:
@@ -560,8 +591,6 @@
return False
test_result = self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
- # Temporary fix
- time.sleep(3)
if not test_result:
self.log.info("Failed to disconnect from peripheral device.")
return False
@@ -603,9 +632,15 @@
orchestrate_gatt_connection(self.cen_ad, self.per_ad))
self.adv_instances.append(adv_callback)
if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
- event = self.cen_ad.ed.pop_event(
- gatt_services_discovered.format(gatt_callback),
- self.default_timeout)
+ expected_event = GattCbStrings.GATT_SERV_DISC.value.format(
+ gatt_callback)
+ try:
+ event = self.cen_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ except Empty:
+ self.log.error(GattCbErr.GATT_SERV_DISC_ERR.value.format(
+ expected_event))
+ return False
discovered_services_index = event['data']['ServicesIndex']
else:
self.log.info("Failed to discover services.")
@@ -633,33 +668,45 @@
discovered_services_index, i, characteristic))
for x in range(100):
for descriptor in descriptor_uuids:
+ self.log.info(
+ "Starting write iteration {} on (Characteristic::Descriptor) {}::{}".format(
+ x + 1, characteristic, descriptor))
self.cen_ad.droid.gattClientDescriptorSetValue(
bluetooth_gatt, discovered_services_index, i,
characteristic, descriptor, test_value)
self.cen_ad.droid.gattClientWriteDescriptor(
bluetooth_gatt, discovered_services_index, i,
characteristic, descriptor)
- event = self.per_ad.ed.pop_event(
- descriptor_write_request.format(
- gatt_server_callback), self.default_timeout)
- self.log.info("{} event found: {}".format(
- descriptor_write_request.format(
- gatt_callback), event))
+ expected_event = GattCbStrings.DESC_WRITE_REQ.value.format(
+ gatt_server_callback)
+ try:
+ event = self.per_ad.ed.pop_event(
+ expected_event, self.default_timeout)
+ except Empty:
+ self.log.error(
+ GattCbErr.DESC_WRITE_REQ_ERR.value.format(
+ expected_event))
+ return False
request_id = event['data']['requestId']
found_value = event['data']['value']
if found_value != test_value:
- self.log.info("Values didn't match. Found: {}, "
- "Expected: {}".format(found_value,
- test_value))
+ self.log.error("Values didn't match. Found: {}, "
+ "Expected: {}".format(found_value,
+ test_value))
return False
self.per_ad.droid.gattServerSendResponse(
gatt_server, bt_device_id, request_id, status,
offset, test_value_return)
- self.log.info("onDescriptorWrite event found: {}"
- .format(self.cen_ad.ed.pop_event(
- descriptor_write.format(
- gatt_callback),
- self.default_timeout)))
+ expected_event = GattCbStrings.DESC_WRITE.value.format(
+ gatt_callback)
+ try:
+ self.cen_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ except Empty:
+ self.log.error(
+ GattCbErr.DESC_WRITE_ERR.value.format(
+ expected_event))
+ return False
return True
@BluetoothBaseTest.bt_test_wrap
@@ -705,19 +752,18 @@
characteristic = (
self.per_ad.droid.gattServerCreateBluetoothGattCharacteristic(
- characteristic_uuid,
- BluetoothGattCharacteristic.PROPERTY_WRITE.value,
- BluetoothGattCharacteristic.PERMISSION_WRITE.value))
+ characteristic_uuid, GattCharacteristic.PROPERTY_WRITE.value,
+ GattCharacteristic.PERMISSION_WRITE.value))
descriptor = self.per_ad.droid.gattServerCreateBluetoothGattDescriptor(
descriptor_uuid,
- BluetoothGattDescriptor.PERMISSION_READ.value
- | BluetoothGattDescriptor.PERMISSION_WRITE.value, )
+ GattDescriptor.PERMISSION_READ.value
+ | GattDescriptor.PERMISSION_WRITE.value, )
self.per_ad.droid.gattServerCharacteristicAddDescriptor(characteristic,
descriptor)
gatt_service = self.per_ad.droid.gattServerCreateService(
- service_uuid, BluetoothGattService.SERVICE_TYPE_PRIMARY.value)
+ service_uuid, GattService.SERVICE_TYPE_PRIMARY.value)
self.per_ad.droid.gattServerAddCharacteristicToService(gatt_service,
characteristic)
@@ -732,9 +778,14 @@
orchestrate_gatt_connection(self.cen_ad, self.per_ad))
if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
- event = self.cen_ad.ed.pop_event(
- gatt_services_discovered.format(gatt_callback),
- self.default_timeout)
+ expected_event = GattCbStrings.GATT_SERV_DISC.value.format(
+ gatt_callback)
+ try:
+ event = self.cen_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ except Empty:
+ self.log.error(GattCbErr.GATT_SERV_DISC_ERR.value.format(
+ expected_event))
discovered_services_index = event['data']['ServicesIndex']
services_count = self.cen_ad.droid.gattClientGetDiscoveredServicesCount(
discovered_services_index)
@@ -760,9 +811,15 @@
bluetooth_gatt, discovered_services_index, disc_service_index,
characteristic_uuid)
- event = self.per_ad.ed.pop_event(
- characteristic_write_request.format(gatt_server_callback),
- self.default_timeout)
+ expected_event = GattCbStrings.CHAR_WRITE_REQ.value.format(
+ gatt_server_callback)
+ try:
+ event = self.per_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ except Empty:
+ self.log.error(GattCbErr.CHAR_WRITE_REQ_ERR.value.format(
+ expected_event))
+ return False
request_id = event['data']['requestId']
bt_device_id = 0
@@ -774,8 +831,14 @@
request_id, status, offset,
test_value_return)
- self.cen_ad.ed.pop_event(
- characteristic_write.format(bluetooth_gatt), self.default_timeout)
+ try:
+ self.cen_ad.ed.pop_event(
+ GattCbStrings.CHAR_WRITE_REQ.value.format(bluetooth_gatt),
+ self.default_timeout)
+ except Empty:
+ self.log.error(GattCbErr.CHAR_WRITE_REQ_ERR.value.format(
+ expected_event))
+ return False
return True
@BluetoothBaseTest.bt_test_wrap
@@ -815,9 +878,15 @@
orchestrate_gatt_connection(self.cen_ad, self.per_ad))
self.adv_instances.append(adv_callback)
if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
- event = self.cen_ad.ed.pop_event(
- gatt_services_discovered.format(gatt_callback),
- self.default_timeout)
+ expected_event = GattCbStrings.GATT_SERV_DISC.value.format(
+ gatt_callback)
+ try:
+ event = self.cen_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ except Empty:
+ self.log.error(GattCbErr.GATT_SERV_DISC_ERR.value.format(
+ expected_event))
+ return False
discovered_services_index = event['data']['ServicesIndex']
else:
self.log.info("Failed to discover services.")
@@ -856,12 +925,18 @@
self.cen_ad.droid.gattClientWriteCharacteristic(
bluetooth_gatt, discovered_services_index, i,
characteristic)
- time.sleep(8)
- event = self.per_ad.ed.pop_event(
- characteristic_write_request.format(
- gatt_server_callback), self.default_timeout)
+ expected_event = GattCbStrings.CHAR_WRITE_REQ.value.format(
+ gatt_server_callback)
+ try:
+ event = self.per_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ except Empty:
+ self.log.error(
+ GattCbErr.CHAR_WRITE_REQ_ERR.value.format(
+ expected_event))
+ return False
self.log.info("{} event found: {}".format(
- characteristic_write_request.format(
+ GattCbStrings.CHAR_WRITE_REQ.value.format(
gatt_server_callback), event))
request_id = event['data']['requestId']
found_value = event['data']['value']
@@ -873,13 +948,16 @@
self.per_ad.droid.gattServerSendResponse(
gatt_server, bt_device_id, request_id, status, offset,
test_value_return)
- self.log.info(
- "onCharacteristicWrite event found: {}".format(
- characteristic_write_request.format(
- gatt_server_callback),
- self.cen_ad.ed.pop_event(
- characteristic_write.format(
- bluetooth_gatt), self.default_timeout)))
+ expected_event = GattCbStrings.CHAR_WRITE_REQ.value.format(
+ gatt_server_callback)
+ try:
+ self.cen_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ except Empty:
+ self.log.error(
+ GattCbErr.CHAR_WRITE_REQ_ERR.value.format(
+ expected_event))
+ return False
return True
def test_gatt_connect_mitm_attack(self):
@@ -894,8 +972,8 @@
1. Create a GATT server and server callback on the peripheral device.
2. Create a unique service and characteristic uuid on the peripheral.
3. Create a characteristic on the peripheral with these properties:
- BluetoothGattCharacteristic.PROPERTY_WRITE.value,
- BluetoothGattCharacteristic.PERMISSION_WRITE_ENCRYPTED_MITM.value
+ GattCharacteristic.PROPERTY_WRITE.value,
+ GattCharacteristic.PERMISSION_WRITE_ENCRYPTED_MITM.value
4. Create a GATT service on the peripheral.
5. Add the characteristic to the GATT service.
6. Create a GATT connection between your central and peripheral device.
@@ -928,11 +1006,12 @@
service_uuid = "3846D7A0-69C8-11E4-BA00-0002A5D5C51B"
test_uuid = "aa7edd5a-4d1d-4f0e-883a-d145616a1630"
bonded = False
- characteristic = self.per_ad.droid.gattServerCreateBluetoothGattCharacteristic(
- test_uuid, BluetoothGattCharacteristic.PROPERTY_WRITE.value,
- BluetoothGattCharacteristic.PERMISSION_WRITE_ENCRYPTED_MITM.value)
+ characteristic = self.per_ad.droid.gattServerCreateBluetoo
+ thGattCharacteristic(
+ test_uuid, GattCharacteristic.PROPERTY_WRITE.value,
+ GattCharacteristic.PERMISSION_WRITE_ENCRYPTED_MITM.value)
gatt_service = self.per_ad.droid.gattServerCreateService(
- service_uuid, BluetoothGattService.SERVICE_TYPE_PRIMARY.value)
+ service_uuid, GattService.SERVICE_TYPE_PRIMARY.value)
self.per_ad.droid.gattServerAddCharacteristicToService(gatt_service,
characteristic)
self.per_ad.droid.gattServerAddService(gatt_server, gatt_service)
@@ -944,9 +1023,15 @@
orchestrate_gatt_connection(self.cen_ad, self.per_ad))
self.adv_instances.append(adv_callback)
if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
- event = self.cen_ad.ed.pop_event(
- gatt_services_discovered.format(gatt_callback),
- self.default_timeout)
+ expected_event = GattCbStrings.GATT_SERV_DISC.value.format(
+ gatt_callback)
+ try:
+ event = self.cen_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ except Empty:
+ self.log.error(GattCbErr.GATT_SERV_DISC_ERR.value.format(
+ expected_event))
+ return False
discovered_services_index = event['data']['ServicesIndex']
else:
self.log.info("Failed to discover services.")
@@ -992,8 +1077,8 @@
1. Create a GATT server and server callback on the peripheral device.
2. Create a unique service and characteristic uuid on the peripheral.
3. Create a characteristic on the peripheral with these properties:
- BluetoothGattCharacteristic.PROPERTY_WRITE.value,
- BluetoothGattCharacteristic.PERMISSION_WRITE_ENCRYPTED_MITM.value
+ GattCharacteristic.PROPERTY_WRITE.value,
+ GattCharacteristic.PERMISSION_WRITE_ENCRYPTED_MITM.value
4. Create a GATT service on the peripheral.
5. Add the characteristic to the GATT service.
6. Create a GATT connection between your central and peripheral device.
@@ -1020,57 +1105,62 @@
Priority: 1
"""
gatt_server_callback = (
- self.per_droid.gattServerCreateGattServerCallback())
- gatt_server = self.per_droid.gattServerOpenGattServer(
+ self.per_ad.droid.gattServerCreateGattServerCallback())
+ gatt_server = self.per_ad.droid.gattServerOpenGattServer(
gatt_server_callback)
service_uuid = "3846D7A0-69C8-11E4-BA00-0002A5D5C51B"
test_uuid = "aa7edd5a-4d1d-4f0e-883a-d145616a1630"
bonded = False
- characteristic = self.per_droid.gattServerCreateBluetoothGattCharacteristic(
- test_uuid, BluetoothGattCharacteristic.PROPERTY_WRITE.value,
- BluetoothGattCharacteristic.PERMISSION_WRITE_ENCRYPTED_MITM.value)
- gatt_service = self.per_droid.gattServerCreateService(
- service_uuid, BluetoothGattService.SERVICE_TYPE_PRIMARY.value)
- self.per_droid.gattServerAddCharacteristicToService(gatt_service,
- characteristic)
- self.per_droid.gattServerAddService(gatt_server, gatt_service)
+ characteristic = self.per_ad.droid.gattServerCreateBluetoothGattCharacteristic(
+ test_uuid, GattCharacteristic.PROPERTY_WRITE.value,
+ GattCharacteristic.PERMISSION_WRITE_ENCRYPTED_MITM.value)
+ gatt_service = self.per_ad.droid.gattServerCreateService(
+ service_uuid, GattService.SERVICE_TYPE_PRIMARY.value)
+ self.per_ad.droid.gattServerAddCharacteristicToService(gatt_service,
+ characteristic)
+ self.per_ad.droid.gattServerAddService(gatt_server, gatt_service)
result = self._find_service_added_event(gatt_server_callback,
service_uuid)
if not result:
return False
bluetooth_gatt, gatt_callback, adv_callback = (
- orchestrate_gatt_connection(self.cen_droid, self.cen_ed,
- self.per_droid, self.per_ed))
+ orchestrate_gatt_connection(self.cen_ad, self.per_ad))
self.adv_instances.append(adv_callback)
- if self.cen_droid.gattClientDiscoverServices(bluetooth_gatt):
- event = self.cen_ed.pop_event(
- gatt_services_discovered.format(gatt_callback),
- self.default_timeout)
+ if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
+ expected_event = GattCbStrings.GATT_SERV_DISC.value.format(
+ gatt_callback)
+ try:
+ event = self.cen_ad.ed.pop_event(expected_event,
+ self.default_timeout)
+ except Empty:
+ self.log.error(GattCbErr.GATT_SERV_DISC_ERR.value.format(
+ expected_event))
+ return False
discovered_services_index = event['data']['ServicesIndex']
else:
self.log.info("Failed to discover services.")
return False
test_value = "1,2,3,4,5,6,7"
- services_count = self.cen_droid.gattClientGetDiscoveredServicesCount(
+ services_count = self.cen_ad.droid.gattClientGetDiscoveredServicesCount(
discovered_services_index)
for i in range(services_count):
characteristic_uuids = (
- self.cen_droid.gattClientGetDiscoveredCharacteristicUuids(
+ self.cen_ad.droid.gattClientGetDiscoveredCharacteristicUuids(
discovered_services_index, i))
for characteristic_uuid in characteristic_uuids:
if characteristic_uuid == test_uuid:
- self.cen_droid.bluetoothStartPairingHelper()
- self.per_droid.bluetoothStartPairingHelper()
- self.cen_droid.gattClientCharacteristicSetValue(
+ self.cen_ad.droid.bluetoothStartPairingHelper()
+ self.per_ad.droid.bluetoothStartPairingHelper()
+ self.cen_ad.droid.gattClientCharacteristicSetValue(
bluetooth_gatt, discovered_services_index, i,
characteristic_uuid, test_value)
- self.cen_droid.gattClientWriteCharacteristic(
+ self.cen_ad.droid.gattClientWriteCharacteristic(
bluetooth_gatt, discovered_services_index, i,
characteristic_uuid)
start_time = time.time() + self.default_timeout
- target_name = self.per_droid.bluetoothGetLocalName()
+ target_name = self.per_ad.droid.bluetoothGetLocalName()
while time.time() < start_time and bonded == False:
- bonded_devices = self.cen_droid.bluetoothGetBondedDevices(
+ bonded_devices = self.cen_ad.droid.bluetoothGetBondedDevices(
)
for device in bonded_devices:
if 'name' in device.keys() and device[
diff --git a/acts/tests/google/ble/scan/BleBackgroundScanTest.py b/acts/tests/google/ble/scan/BleBackgroundScanTest.py
index 992f969..cc3df3a 100644
--- a/acts/tests/google/ble/scan/BleBackgroundScanTest.py
+++ b/acts/tests/google/ble/scan/BleBackgroundScanTest.py
@@ -32,7 +32,6 @@
class BleBackgroundScanTest(BluetoothBaseTest):
- tests = None
default_timeout = 10
max_scan_instances = 28
report_delay = 2000
@@ -49,8 +48,6 @@
if self.droid_list[1]['max_advertisements'] == 0:
self.tests = ()
return
- self.tests = ("test_background_scan",
- "test_background_scan_ble_disabled", )
def setup_test(self):
self.log.debug(log_energy_info(self.android_devices, "Start"))
@@ -86,10 +83,6 @@
self.log.info("No scan result found as expected.")
return True
- def _delete_me(self):
- import time
- time.sleep(5)
-
@BluetoothBaseTest.bt_test_wrap
def test_background_scan(self):
"""Test generic background scan.
@@ -117,18 +110,36 @@
import time
self._setup_generic_advertisement()
self.scn_ad.droid.bluetoothToggleState(False)
- self.scn_ad.ed.pop_event(bluetooth_off, self.default_timeout)
+ try:
+ self.scn_ad.ed.pop_event(bluetooth_off, self.default_timeout)
+ except Empty:
+ self.log.error("Bluetooth Off event not found. Expected {}".format(
+ bluetooth_off))
+ return False
self.scn_ad.droid.bluetoothDisableBLE()
- self.scn_ad.ed.pop_event(bluetooth_off, self.default_timeout)
+ try:
+ self.scn_ad.ed.pop_event(bluetooth_off, self.default_timeout)
+ except Empty:
+ self.log.error("Bluetooth Off event not found. Expected {}".format(
+ bluetooth_off))
+ return False
self.scn_ad.droid.bluetoothEnableBLE()
- self._delete_me()
- self.scn_ad.ed.pop_event(bluetooth_on, self.default_timeout * 2)
+ try:
+ self.scn_ad.ed.pop_event(bluetooth_off, self.default_timeout*2)
+ except Empty:
+ self.log.error("Bluetooth On event not found. Expected {}".format(
+ bluetooth_on))
+ return False
filter_list, scan_settings, scan_callback = generate_ble_scan_objects(
self.scn_ad.droid)
self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
scan_callback)
- self.scn_ad.ed.pop_event(
- scan_result.format(scan_callback), self.default_timeout)
+ expected_event = scan_result.format(scan_callback)
+ try:
+ self.scn_ad.ed.pop_event(expected_event, self.default_timeout)
+ except Empty:
+ self.log.error("Scan Result event not found. Expected {}".format(expected_event))
+ return False
return True
@BluetoothBaseTest.bt_test_wrap
@@ -158,14 +169,23 @@
self._setup_generic_advertisement()
self.scn_ad.droid.bluetoothEnableBLE()
self.scn_ad.droid.bluetoothToggleState(False)
- self.scn_ad.ed.pop_event(bluetooth_off, self.default_timeout)
- self._delete_me()
+ try:
+ self.scn_ad.ed.pop_event(bluetooth_off, self.default_timeout)
+ except Empty:
+ self.log.error("Bluetooth Off event not found. Expected {}".format(
+ bluetooth_off))
+ return False
filter_list, scan_settings, scan_callback = generate_ble_scan_objects(
self.scn_ad.droid)
try:
self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
scan_callback)
- self.scn_ad.ed.pop_event(scan_result.format(scan_callback))
+ expected_event = scan_result.format(scan_callback)
+ try:
+ self.scn_ad.ed.pop_event(expected_event, self.default_timeout)
+ except Empty:
+ self.log.error("Scan Result event not found. Expected {}".format(expected_event))
+ return False
self.log.info("Was able to start background scan even though ble "
"was disabled.")
return False
diff --git a/acts/tests/google/ble/scan/BleOnLostOnFoundTest.py b/acts/tests/google/ble/scan/BleOnLostOnFoundTest.py
index c17d00c..4cd72dc 100644
--- a/acts/tests/google/ble/scan/BleOnLostOnFoundTest.py
+++ b/acts/tests/google/ble/scan/BleOnLostOnFoundTest.py
@@ -33,7 +33,6 @@
class BleOnLostOnFoundTest(BluetoothBaseTest):
- tests = None
default_timeout = 10
max_scan_instances = 28
active_scan_callback_list = []
@@ -47,9 +46,6 @@
if self.droid_list[1]['max_advertisements'] == 0:
self.tests = ()
return
- self.tests = ("test_onlost_onfound_defaults",
- "test_onlost_onfound_match_mode_sticky",
- "test_onlost_onfound_match_num_few", )
def teardown_test(self):
self.log.info(log_energy_info(self.android_devices, "End"))
diff --git a/acts/tests/google/ble/scan/BleOpportunisticScanTest.py b/acts/tests/google/ble/scan/BleOpportunisticScanTest.py
index 3f948f4..f6c0ce5 100644
--- a/acts/tests/google/ble/scan/BleOpportunisticScanTest.py
+++ b/acts/tests/google/ble/scan/BleOpportunisticScanTest.py
@@ -36,7 +36,6 @@
class BleOpportunisticScanTest(BluetoothBaseTest):
- tests = None
default_timeout = 10
max_scan_instances = 28
report_delay = 2000
diff --git a/acts/tests/google/ble/system_tests/BleOnLostOnFoundStressTest.py b/acts/tests/google/ble/system_tests/BleOnLostOnFoundStressTest.py
index 3965f2c..04d5cfd 100644
--- a/acts/tests/google/ble/system_tests/BleOnLostOnFoundStressTest.py
+++ b/acts/tests/google/ble/system_tests/BleOnLostOnFoundStressTest.py
@@ -29,14 +29,13 @@
from acts.test_utils.bt.BleEnum import ScanSettingsMatchNum
from acts.test_utils.bt.bt_test_utils import cleanup_scanners_and_advertisers
from acts.test_utils.bt.bt_test_utils import get_advanced_droid_list
-from acts.test_utils.bt.bt_test_utils import orchestrate_gatt_connection
+from acts.test_utils.bt.bt_gatt_utils import orchestrate_gatt_connection
from acts.test_utils.bt.bt_test_utils import reset_bluetooth
-from acts.test_utils.bt.bt_test_utils import run_continuous_write_descriptor
-from acts.test_utils.bt.bt_test_utils import setup_multiple_services
+from acts.test_utils.bt.bt_gatt_utils import run_continuous_write_descriptor
+from acts.test_utils.bt.bt_gatt_utils import setup_multiple_services
class BleOnLostOnFoundStressTest(BluetoothBaseTest):
- tests = None
default_timeout = 10
max_scan_instances = 28
report_delay = 2000
@@ -53,8 +52,6 @@
if self.droid_list[1]['max_advertisements'] == 0:
self.tests = ()
return
- self.tests = ("test_on_star_while_polling_energy_stats",
- "test_more_stress_test", )
def teardown_test(self):
cleanup_scanners_and_advertisers(
@@ -156,8 +153,7 @@
gatt_server_callback, gatt_server = setup_multiple_services(
self.adv_ad)
bluetooth_gatt, gatt_callback, adv_callback = (
- orchestrate_gatt_connection(self.scn_ad.droid, self.scn_ad.ed,
- self.adv_ad.droid, self.adv_ad.ed))
+ orchestrate_gatt_connection(self.scn_ad, self.adv_ad))
self.active_scan_callback_list.append(adv_callback)
if self.scn_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
event = self.scn_ad.ed.pop_event(
diff --git a/acts/tests/google/ble/system_tests/BleStressTest.py b/acts/tests/google/ble/system_tests/BleStressTest.py
index fcee3bc..22644ce 100644
--- a/acts/tests/google/ble/system_tests/BleStressTest.py
+++ b/acts/tests/google/ble/system_tests/BleStressTest.py
@@ -31,7 +31,6 @@
class BleStressTest(BluetoothBaseTest):
- tests = None
default_timeout = 10
def __init__(self, controllers):
@@ -270,7 +269,7 @@
self.scn_ad.droid)
self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
scan_callback)
- reset_bluetooth([self.scan_ad])
+ reset_bluetooth([self.scn_ad])
self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
scan_callback)
diff --git a/acts/tests/google/bt/BtBasicFunctionalityTest.py b/acts/tests/google/bt/BtBasicFunctionalityTest.py
index addc44d..5419674 100644
--- a/acts/tests/google/bt/BtBasicFunctionalityTest.py
+++ b/acts/tests/google/bt/BtBasicFunctionalityTest.py
@@ -40,14 +40,6 @@
BluetoothBaseTest.__init__(self, controllers)
self.droid_ad = self.android_devices[0]
self.droid1_ad = self.android_devices[1]
- self.tests = ("test_bluetooth_reset",
- "test_make_device_discoverable",
- "test_make_device_undiscoverable",
- "test_set_device_name",
- "test_scan_mode_off",
- "test_scan_mode_none",
- "test_scan_mode_connectable",
- "test_scan_mode_connectable_discoverable", )
def setup_class(self):
return setup_multiple_devices_for_bt_test(self.android_devices)
diff --git a/acts/tests/google/bt/SppTest.py b/acts/tests/google/bt/RfcommTest.py
similarity index 62%
rename from acts/tests/google/bt/SppTest.py
rename to acts/tests/google/bt/RfcommTest.py
index efb3c35..4d939fd 100644
--- a/acts/tests/google/bt/SppTest.py
+++ b/acts/tests/google/bt/RfcommTest.py
@@ -18,6 +18,7 @@
This test was designed to be run in a shield box.
"""
+from contextlib import suppress
import threading
import time
@@ -32,8 +33,9 @@
from acts.test_utils.bt.bt_test_utils import take_btsnoop_logs
-class SppTest(BluetoothBaseTest):
+class RfcommTest(BluetoothBaseTest):
default_timeout = 10
+ rf_client_th = 0
scan_discovery_time = 5
thread_list = []
message = (
@@ -44,9 +46,8 @@
def __init__(self, controllers):
BluetoothBaseTest.__init__(self, controllers)
- self.server_ad = self.android_devices[0]
- self.client_ad = self.android_devices[1]
- self.tests = ("test_spp_connection", )
+ self.client_ad = self.android_devices[0]
+ self.server_ad = self.android_devices[1]
def _clear_bonded_devices(self):
for a in self.android_devices:
@@ -73,8 +74,11 @@
reset_bluetooth(self.android_devices)
def teardown_test(self):
- for thread in self.thread_list:
- thread.join()
+ with suppress(Exception):
+ for thread in self.thread_list:
+ thread.join()
+ self.client_ad.droid.bluetoothRfcommStop()
+ self.server_ad.droid.bluetoothRfcommStop()
def orchestrate_rfcomm_connect(self, server_mac):
accept_thread = threading.Thread(target=rfcomm_accept,
@@ -84,21 +88,24 @@
connect_thread = threading.Thread(
target=rfcomm_connect,
args=(self.client_ad.droid, server_mac))
+ self.rf_client_th = connect_thread
self.thread_list.append(connect_thread)
connect_thread.start()
@BluetoothBaseTest.bt_test_wrap
- def test_spp_connection(self):
- """Test bluetooth SPP profile.
+ def test_rfcomm_connection_write_ascii(self):
+ """Test bluetooth RFCOMM writing and reading ascii data
- Test SPP profile though establishing an RFCOMM connection.
+ Test RFCOMM though establishing a connection.
Steps:
1. Get the mac address of the server device.
2. Establish an RFCOMM connection from the client to the server AD.
3. Verify that the RFCOMM connection is active from both the client and
server.
- 4. Disconnect the RFCOMM connection.
+ 4. Write data from the client and read received data from the server.
+ 5. Verify data matches from client and server
+ 6. Disconnect the RFCOMM connection.
Expected Result:
RFCOMM connection is established then disconnected succcessfully.
@@ -107,14 +114,10 @@
Pass if True
Fail if False
- TAGS: Classic, SPP, RFCOMM
+ TAGS: Classic, RFCOMM
Priority: 1
"""
- server_mac = get_bt_mac_address(self.client_ad.droid,
- self.server_ad.droid)
- # temporary workaround. Need to find out why I can't connect after I do
- # a device discovery from get_bt_mac_address
- reset_bluetooth([self.server_ad])
+ server_mac = self.server_ad.droid.bluetoothGetLocalAddress()
self.orchestrate_rfcomm_connect(server_mac)
self.log.info("Write message.")
self.client_ad.droid.bluetoothRfcommWrite(self.message)
@@ -128,6 +131,52 @@
if len(self.client_ad.droid.bluetoothRfcommActiveConnections()) == 0:
self.log.info("no rfcomm connections found on client.")
return False
+
+ self.client_ad.droid.bluetoothRfcommStop()
+ self.server_ad.droid.bluetoothRfcommStop()
+ return True
+
+ @BluetoothBaseTest.bt_test_wrap
+ def test_rfcomm_write_binary(self):
+ """Test bluetooth RFCOMM writing and reading binary data
+
+ Test profile though establishing an RFCOMM connection.
+
+ Steps:
+ 1. Get the mac address of the server device.
+ 2. Establish an RFCOMM connection from the client to the server AD.
+ 3. Verify that the RFCOMM connection is active from both the client and
+ server.
+ 4. Write data from the client and read received data from the server.
+ 5. Verify data matches from client and server
+ 6. Disconnect the RFCOMM connection.
+
+ Expected Result:
+ RFCOMM connection is established then disconnected succcessfully.
+
+ Returns:
+ Pass if True
+ Fail if False
+
+ TAGS: Classic, RFCOMM
+ Priority: 1
+ """
+ server_mac = self.server_ad.droid.bluetoothGetLocalAddress()
+ self.orchestrate_rfcomm_connect(server_mac)
+ binary_message = "11010101"
+ self.log.info("Write message.")
+ self.client_ad.droid.bluetoothRfcommWriteBinary(binary_message)
+ self.log.info("Read message.")
+ read_msg = self.server_ad.droid.bluetoothRfcommReadBinary().rstrip("\r\n")
+ self.log.info("Verify message.")
+ assert binary_message == read_msg, "Mismatch! Read {}".format(read_msg)
+ if len(self.server_ad.droid.bluetoothRfcommActiveConnections()) == 0:
+ self.log.info("No rfcomm connections found on server.")
+ return False
+ if len(self.client_ad.droid.bluetoothRfcommActiveConnections()) == 0:
+ self.log.info("no rfcomm connections found on client.")
+ return False
+
self.client_ad.droid.bluetoothRfcommStop()
self.server_ad.droid.bluetoothRfcommStop()
return True
diff --git a/acts/tests/google/bt/gatt/GattOverBrEdrTest.py b/acts/tests/google/bt/gatt/GattOverBrEdrTest.py
index 399d974..3241543 100644
--- a/acts/tests/google/bt/gatt/GattOverBrEdrTest.py
+++ b/acts/tests/google/bt/gatt/GattOverBrEdrTest.py
@@ -17,31 +17,27 @@
Test suite for GATT over BR/EDR.
"""
+import time
+
from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
from acts.test_utils.bt.bt_test_utils import reset_bluetooth
-from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
-from acts.test_utils.bt.BleEnum import BluetoothGattCharacteristic
-from acts.test_utils.bt.BleEnum import BluetoothGattDescriptor
-from acts.test_utils.bt.BleEnum import BluetoothGattService
-from acts.test_utils.bt.BleEnum import BluetoothMtuSize
-from acts.test_utils.bt.bt_test_utils import descriptor_write
-from acts.test_utils.bt.bt_test_utils import descriptor_write_request
-from acts.test_utils.bt.bt_test_utils import disconnect_gatt_connection
-from acts.test_utils.bt.bt_test_utils import gatt_services_discovered
+from acts.test_utils.bt.GattEnum import GattCharacteristic
+from acts.test_utils.bt.GattEnum import GattDescriptor
+from acts.test_utils.bt.GattEnum import GattService
+from acts.test_utils.bt.GattEnum import MtuSize
+from acts.test_utils.bt.GattEnum import GattCbStrings
+from acts.test_utils.bt.bt_gatt_utils import disconnect_gatt_connection
+from acts.test_utils.bt.bt_gatt_utils import orchestrate_gatt_connection
+from acts.test_utils.bt.bt_gatt_utils import setup_gatt_characteristics
+from acts.test_utils.bt.bt_gatt_utils import setup_gatt_connection
+from acts.test_utils.bt.bt_gatt_utils import setup_gatt_descriptors
from acts.test_utils.bt.bt_test_utils import get_advanced_droid_list
-from acts.test_utils.bt.bt_test_utils import get_bt_mac_address
-from acts.test_utils.bt.bt_test_utils import orchestrate_gatt_connection
-from acts.test_utils.bt.bt_test_utils import mtu_changed
-from acts.test_utils.bt.bt_test_utils import read_remote_rssi
-from acts.test_utils.bt.bt_test_utils import service_added
-from acts.test_utils.bt.bt_test_utils import setup_gatt_characteristics
-from acts.test_utils.bt.bt_test_utils import setup_gatt_descriptors
+from acts.test_utils.bt.bt_test_utils import log_energy_info
from acts.test_utils.bt.bt_test_utils import setup_multiple_devices_for_bt_test
from acts.test_utils.bt.bt_test_utils import take_btsnoop_logs
class GattOverBrEdrTest(BluetoothBaseTest):
- tests = None
default_timeout = 10
default_discovery_timeout = 3
droid_list = ()
@@ -52,25 +48,13 @@
self.droid_list = get_advanced_droid_list(self.android_devices)
self.cen_ad = self.android_devices[0]
self.per_ad = self.android_devices[1]
- self.tests = (
- "test_gatt_bredr_connect",
- "test_gatt_bredr_request_min_mtu",
- "test_gatt_bredr_request_max_mtu",
- "test_gatt_bredr_request_out_of_bounds_mtu",
- "test_gatt_bredr_connect_trigger_on_read_rssi",
- "test_gatt_bredr_connect_trigger_on_services_discovered",
- "test_gatt_bredr_connect_trigger_on_services_discovered_iterate_attributes",
- "test_gatt_bredr_connect_with_service_uuid_variations",
- "test_gatt_bredr_connect_multiple_iterations",
- "test_bredr_write_descriptor_stress",
- "test_gatt_bredr_connect_mitm_attack", )
def setup_class(self):
self.log.info("Setting up devices for bluetooth testing.")
if not setup_multiple_devices_for_bt_test(self.android_devices):
return False
- self.per_droid_mac_address = get_bt_mac_address(
- self.cen_ad.droid, self.per_ad.droid, False)
+ self.per_droid_mac_address = self.per_ad.droid.bluetoothGetLocalAddress(
+ )
if not self.per_droid_mac_address:
return False
return True
@@ -83,32 +67,32 @@
characteristic_input = [
{
'uuid': "aa7edd5a-4d1d-4f0e-883a-d145616a1630",
- 'property': BluetoothGattCharacteristic.PROPERTY_WRITE.value
- | BluetoothGattCharacteristic.PROPERTY_WRITE_NO_RESPONSE.value,
- 'permission': BluetoothGattCharacteristic.PROPERTY_WRITE.value
+ 'property': GattCharacteristic.PROPERTY_WRITE.value
+ | GattCharacteristic.PROPERTY_WRITE_NO_RESPONSE.value,
+ 'permission': GattCharacteristic.PROPERTY_WRITE.value
},
{
'uuid': "21c0a0bf-ad51-4a2d-8124-b74003e4e8c8",
- 'property': BluetoothGattCharacteristic.PROPERTY_NOTIFY.value
- | BluetoothGattCharacteristic.PROPERTY_READ.value,
- 'permission': BluetoothGattCharacteristic.PERMISSION_READ.value
+ 'property': GattCharacteristic.PROPERTY_NOTIFY.value
+ | GattCharacteristic.PROPERTY_READ.value,
+ 'permission': GattCharacteristic.PERMISSION_READ.value
},
{
'uuid': "6774191f-6ec3-4aa2-b8a8-cf830e41fda6",
- 'property': BluetoothGattCharacteristic.PROPERTY_NOTIFY.value
- | BluetoothGattCharacteristic.PROPERTY_READ.value,
- 'permission': BluetoothGattCharacteristic.PERMISSION_READ.value
+ 'property': GattCharacteristic.PROPERTY_NOTIFY.value
+ | GattCharacteristic.PROPERTY_READ.value,
+ 'permission': GattCharacteristic.PERMISSION_READ.value
},
]
descriptor_input = [
{
'uuid': "aa7edd5a-4d1d-4f0e-883a-d145616a1630",
- 'property': BluetoothGattDescriptor.PERMISSION_READ.value
- | BluetoothGattDescriptor.PERMISSION_WRITE.value,
+ 'property': GattDescriptor.PERMISSION_READ.value
+ | GattDescriptor.PERMISSION_WRITE.value,
}, {
'uuid': "76d5ed92-ca81-4edb-bb6b-9f019665fb32",
- 'property': BluetoothGattDescriptor.PERMISSION_READ.value
- | BluetoothGattCharacteristic.PERMISSION_WRITE.value,
+ 'property': GattDescriptor.PERMISSION_READ.value
+ | GattCharacteristic.PERMISSION_WRITE.value,
}
]
characteristic_list = setup_gatt_characteristics(droid,
@@ -147,7 +131,8 @@
def _find_service_added_event(self, gatt_server_callback, uuid):
event = self.per_ad.ed.pop_event(
- service_added.format(gatt_server_callback), self.default_timeout)
+ GattCbStrings.SERV_ADDED.value.format(gatt_server_callback),
+ self.default_timeout)
if event['data']['serviceUuid'].lower() != uuid.lower():
self.log.info("Uuid mismatch. Found: {}, Expected {}.".format(
event['data']['serviceUuid'], uuid))
@@ -167,13 +152,13 @@
characteristic_list[2], descriptor_list[1])
gatt_service = self.per_ad.droid.gattServerCreateService(
"00000000-0000-1000-8000-00805f9b34fb",
- BluetoothGattService.SERVICE_TYPE_PRIMARY.value)
+ GattService.SERVICE_TYPE_PRIMARY.value)
gatt_service2 = self.per_ad.droid.gattServerCreateService(
"FFFFFFFF-0000-1000-8000-00805f9b34fb",
- BluetoothGattService.SERVICE_TYPE_PRIMARY.value)
+ GattService.SERVICE_TYPE_PRIMARY.value)
gatt_service3 = self.per_ad.droid.gattServerCreateService(
"3846D7A0-69C8-11E4-BA00-0002A5D5C51B",
- BluetoothGattService.SERVICE_TYPE_PRIMARY.value)
+ GattService.SERVICE_TYPE_PRIMARY.value)
for characteristic in characteristic_list:
self.per_ad.droid.gattServerAddCharacteristicToService(
gatt_service, characteristic)
@@ -265,10 +250,11 @@
orchestrate_gatt_connection(self.cen_ad, self.per_ad, False,
self.per_droid_mac_address))
self.cen_ad.droid.gattClientRequestMtu(bluetooth_gatt,
- BluetoothMtuSize.MIN.value)
- mtu_event = self.cen_ad.ed.pop_event(mtu_changed.format(
- bluetooth_gatt))
- if mtu_event['data']['MTU'] != BluetoothMtuSize.MIN.value:
+ MtuSize.MIN.value)
+ mtu_event = self.cen_ad.ed.pop_event(
+ GattCbStrings.MTU_CHANGED.value.format(
+ bluetooth_gatt), self.default_timeout)
+ if mtu_event['data']['MTU'] != MtuSize.MIN.value:
return False
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -306,10 +292,11 @@
orchestrate_gatt_connection(self.cen_ad, self.per_ad, False,
self.per_droid_mac_address))
self.cen_ad.droid.gattClientRequestMtu(bluetooth_gatt,
- BluetoothMtuSize.MAX.value)
- mtu_event = self.cen_ad.ed.pop_event(mtu_changed.format(
- bluetooth_gatt))
- if mtu_event['data']['MTU'] != BluetoothMtuSize.MAX.value:
+ MtuSize.MAX.value)
+ mtu_event = self.cen_ad.ed.pop_event(
+ GattCbStrings.MTU_CHANGED.value.format(
+ bluetooth_gatt), self.default_timeout)
+ if mtu_event['data']['MTU'] != MtuSize.MAX.value:
return False
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -348,16 +335,17 @@
orchestrate_gatt_connection(self.cen_ad, self.per_ad, False,
self.per_droid_mac_address))
self.cen_ad.droid.gattClientRequestMtu(bluetooth_gatt,
- BluetoothMtuSize.MIN.value - 1)
+ MtuSize.MIN.value - 1)
try:
self.cen_ad.ed.pop_event(
- mtu_changed.format(bluetooth_gatt), self.default_timeout)
+ GattCbStrings.MTU_CHANGED.value.format(bluetooth_gatt),
+ self.default_timeout)
self.log.error("Found {} event when it wasn't expected".format(
- mtu_changed.format(bluetooth_gatt)))
+ GattCbStrings.MTU_CHANGED.format(bluetooth_gatt)))
return False
except Exception:
- self.log.debug("Successfully didn't find {} event"
- .format(mtu_changed.format(bluetooth_gatt)))
+ self.log.debug("Successfully didn't find {} event".format(
+ GattCbStrings.MTU_CHANGED.value.format(bluetooth_gatt)))
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -393,7 +381,8 @@
self.per_droid_mac_address))
if self.cen_ad.droid.gattClientReadRSSI(bluetooth_gatt):
self.cen_ad.ed.pop_event(
- read_remote_rssi.format(gatt_callback), self.default_timeout)
+ GattCbStrings.RD_REMOTE_RSSI.value.format(
+ gatt_callback), self.default_timeout)
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -430,7 +419,7 @@
discovered_services_index = -1
if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
event = self.cen_ad.ed.pop_event(
- gatt_services_discovered.format(gatt_callback),
+ GattCbStrings.GATT_SERV_DISC.value.format(gatt_callback),
self.default_timeout)
discovered_services_index = event['data']['ServicesIndex']
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
@@ -475,7 +464,7 @@
discovered_services_index = -1
if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
event = self.cen_ad.ed.pop_event(
- gatt_services_discovered.format(gatt_callback),
+ GattCbStrings.GATT_SERV_DISC.value.format(gatt_callback),
self.default_timeout)
discovered_services_index = event['data']['ServicesIndex']
self._iterate_attributes(discovered_services_index)
@@ -519,7 +508,7 @@
discovered_services_index = -1
if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
event = self.cen_ad.ed.pop_event(
- gatt_services_discovered.format(gatt_callback),
+ GattCbStrings.GATT_SERV_DISC.value.format(gatt_callback),
self.default_timeout)
discovered_services_index = event['data']['ServicesIndex']
self._iterate_attributes(discovered_services_index)
@@ -554,7 +543,7 @@
Priority: 1
"""
autoconnect = False
- mac_address = get_bt_mac_address(self.cen_ad.droid, self.per_ad.droid)
+ mac_address = self.per_ad.droid.bluetoothGetLocalAddress()
for i in range(20):
bluetooth_gatt, gatt_callback, adv_callback = (
orchestrate_gatt_connection(self.cen_ad, self.per_ad, False,
@@ -603,7 +592,7 @@
self.per_droid_mac_address))
if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
event = self.cen_ad.ed.pop_event(
- gatt_services_discovered.format(gatt_callback),
+ GattCbStrings.GATT_SERV_DISC.value.format(gatt_callback),
self.default_timeout)
discovered_services_index = event['data']['ServicesIndex']
else:
@@ -639,8 +628,8 @@
bluetooth_gatt, discovered_services_index, i,
characteristic, descriptor)
event = self.per_ad.ed.pop_event(
- descriptor_write_request.format(gatt_callback),
- self.default_timeout)
+ GattCbStrings.DESC_WRITE_REQ.value.format(
+ gatt_callback), self.default_timeout)
self.log.info(
"onDescriptorWriteRequest event found: {}".format(
event))
@@ -657,7 +646,8 @@
self.log.info(
"onDescriptorWrite event found: {}".format(
self.cen_ad.ed.pop_event(
- descriptor_write.format(bluetooth_gatt),
+ GattCbStrings.DESC_WRITE.value.format(
+ bluetooth_gatt),
self.default_timeout)))
return True
@@ -674,8 +664,8 @@
1. Create a GATT server and server callback on the peripheral device.
2. Create a unique service and characteristic uuid on the peripheral.
3. Create a characteristic on the peripheral with these properties:
- BluetoothGattCharacteristic.PROPERTY_WRITE.value,
- BluetoothGattCharacteristic.PERMISSION_WRITE_ENCRYPTED_MITM.value
+ GattCharacteristic.PROPERTY_WRITE.value,
+ GattCharacteristic.PERMISSION_WRITE_ENCRYPTED_MITM.value
4. Create a GATT service on the peripheral.
5. Add the characteristic to the GATT service.
6. Create a GATT connection between your central and peripheral device.
@@ -709,10 +699,10 @@
test_uuid = "aa7edd5a-4d1d-4f0e-883a-d145616a1630"
bonded = False
characteristic = self.per_ad.droid.gattServerCreateBluetoothGattCharacteristic(
- test_uuid, BluetoothGattCharacteristic.PROPERTY_WRITE.value,
- BluetoothGattCharacteristic.PERMISSION_WRITE_ENCRYPTED_MITM.value)
+ test_uuid, GattCharacteristic.PROPERTY_WRITE.value,
+ GattCharacteristic.PERMISSION_WRITE_ENCRYPTED_MITM.value)
gatt_service = self.per_ad.droid.gattServerCreateService(
- service_uuid, BluetoothGattService.SERVICE_TYPE_PRIMARY.value)
+ service_uuid, GattService.SERVICE_TYPE_PRIMARY.value)
self.per_ad.droid.gattServerAddCharacteristicToService(gatt_service,
characteristic)
self.per_ad.droid.gattServerAddService(gatt_server, gatt_service)
@@ -723,9 +713,11 @@
bluetooth_gatt, gatt_callback, adv_callback = (
orchestrate_gatt_connection(self.cen_ad, self.per_ad, False,
self.per_droid_mac_address))
+ if bluetooth_gatt is False:
+ return False
if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt):
event = self.cen_ad.ed.pop_event(
- gatt_services_discovered.format(gatt_callback),
+ GattCbStrings.GATT_SERV_DISC.value.format(gatt_callback),
self.default_timeout)
discovered_services_index = event['data']['ServicesIndex']
else:
diff --git a/acts/tests/google/bt/system_tests/BtStressTest.py b/acts/tests/google/bt/system_tests/BtStressTest.py
index d28ebc6..0bb0780 100644
--- a/acts/tests/google/bt/system_tests/BtStressTest.py
+++ b/acts/tests/google/bt/system_tests/BtStressTest.py
@@ -24,12 +24,10 @@
class BtStressTest(BaseTestClass):
- tests = None
default_timeout = 10
def __init__(self, controllers):
BaseTestClass.__init__(self, controllers)
- self.tests = ("test_toggle_bluetooth", )
def setup_class(self):
return setup_multiple_devices_for_bt_test(self.android_devices)
diff --git a/acts/tests/google/bt/system_tests/RfcommLongevityTest.py b/acts/tests/google/bt/system_tests/RfcommLongevityTest.py
new file mode 100644
index 0000000..bc02f19
--- /dev/null
+++ b/acts/tests/google/bt/system_tests/RfcommLongevityTest.py
@@ -0,0 +1,423 @@
+#/usr/bin/env python3.4
+#
+# 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.
+"""
+Test script to execute Bluetooth basic functionality test cases.
+This test was designed to be run in a shield box.
+"""
+
+import threading
+import time
+from contextlib import suppress
+from random import randint
+
+from queue import Empty
+from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
+from acts.test_utils.bt.bt_test_utils import reset_bluetooth
+from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
+from acts.test_utils.bt.bt_test_utils import get_bt_mac_address
+from acts.test_utils.bt.bt_test_utils import rfcomm_accept
+from acts.test_utils.bt.bt_test_utils import rfcomm_connect
+from acts.test_utils.bt.bt_test_utils import take_btsnoop_logs
+
+
+class RfcommLongevityTest(BluetoothBaseTest):
+ default_timeout = 10
+ longev_iterations = 200
+ thread_list = []
+ generic_message = (
+ "Space: the final frontier. These are the voyages of "
+ "the starship Enterprise. Its continuing mission: to explore "
+ "strange new worlds, to seek out new life and new civilizations,"
+ " to boldly go where no man has gone before.")
+
+ def __init__(self, controllers):
+ BluetoothBaseTest.__init__(self, controllers)
+ self.client_ad = self.android_devices[0]
+ self.server_ad = self.android_devices[1]
+
+ def on_fail(self, test_name, begin_time):
+ take_btsnoop_logs(self.android_devices, self, test_name)
+ reset_bluetooth(self.android_devices)
+
+ def teardown_test(self):
+ with suppress(Exception):
+ for thread in self.thread_list:
+ thread.join()
+ reset_bluetooth(self.android_devices)
+ time.sleep(20) #safeguard in case of connId errors
+
+ def orchestrate_rfcomm_connect(self, server_mac):
+ accept_thread = threading.Thread(target=rfcomm_accept,
+ args=(self.server_ad.droid, ))
+ self.thread_list.append(accept_thread)
+ accept_thread.start()
+ connect_thread = threading.Thread(
+ target=rfcomm_connect,
+ args=(self.client_ad.droid, server_mac))
+ self.thread_list.append(connect_thread)
+ connect_thread.start()
+
+ def test_rfcomm_longev_read_write_message(self):
+ """Longevity test an RFCOMM connection's I/O with a generic message
+
+ Test the longevity of RFCOMM with a basic read/write
+ connect/disconnect sequence.
+
+ Steps:
+ 1. Establish a bonding between two Android devices.
+ 2. Write data to RFCOMM from the client droid.
+ 3. Read data from RFCOMM from the server droid.
+ 4. Verify data written matches data read.
+ 5. Repeat steps 2-4 5000 times.
+ 6. Disconnect RFCOMM connection.
+ 7. Repeat steps 1-6 1000 times.
+
+ Expected Result:
+ Each iteration should read and write to the RFCOMM connection
+ successfully. Each connect and disconnect should be successful.
+
+ Returns:
+ Pass if True
+ Fail if False
+
+ TAGS: Classic, Longevity, RFCOMM
+ Priority: 2
+ """
+ server_mac = self.server_ad.droid.bluetoothGetLocalAddress()
+ write_iterations = 5000
+ for i in range(self.longev_iterations):
+ self.log.info("iteration {} connection".format(i))
+ self.orchestrate_rfcomm_connect(server_mac)
+ for n in range(write_iterations):
+ self.log.info("iteration {} data".format(((n + 1) + (
+ i * write_iterations))))
+ self.log.info("Write message.")
+ self.client_ad.droid.bluetoothRfcommWrite(self.generic_message)
+ self.log.info("Read message.")
+ read_msg = self.server_ad.droid.bluetoothRfcommRead()
+ self.log.info("Verify message.")
+ assert self.generic_message == read_msg, "Mismatch! Read {}".format(
+ read_msg)
+ self.log.info("Iteration {} completed".format(n))
+ self.client_ad.droid.bluetoothRfcommStop()
+ self.server_ad.droid.bluetoothRfcommStop()
+ for t in self.thread_list:
+ t.join()
+ self.thread_list.clear()
+ return True
+
+ def test_rfcomm_longev_read_write_small_message(self):
+ """Longevity test an RFCOMM connection's I/O with a small message
+
+ Test the longevity of RFCOMM with a basic read/write
+ connect/disconnect sequence. The data being transfered is only
+ one character in size.
+
+ Steps:
+ 1. Establish a bonding between two Android devices.
+ 2. Write data to RFCOMM from the client droid.
+ 3. Read data from RFCOMM from the server droid.
+ 4. Verify data written matches data read.
+ 5. Repeat steps 2-4 5000 times.
+ 6. Disconnect RFCOMM connection.
+ 7. Repeat steps 1-6 1000 times.
+
+ Expected Result:
+ Each iteration should read and write to the RFCOMM connection
+ successfully. Each connect and disconnect should be successful.
+
+ Returns:
+ Pass if True
+ Fail if False
+
+ TAGS: Classic, Longevity, RFCOMM
+ Priority: 2
+ """
+ server_mac = self.server_ad.droid.bluetoothGetLocalAddress()
+ message = "x"
+ write_iterations = 5000
+ for i in range(self.longev_iterations):
+ self.log.info("iteration {} connection".format(i))
+ self.orchestrate_rfcomm_connect(server_mac)
+ for n in range(write_iterations):
+ self.log.info("iteration {} data".format(((n + 1) + (
+ i * write_iterations))))
+ self.log.info("Write message.")
+ self.client_ad.droid.bluetoothRfcommWrite(message)
+ self.log.info("Read message.")
+ read_msg = self.server_ad.droid.bluetoothRfcommRead()
+ self.log.info("Verify message.")
+ assert message == read_msg, "Mismatch! Read {}".format(
+ read_msg)
+ self.log.info("Iteration {} completed".format(n))
+ self.client_ad.droid.bluetoothRfcommStop()
+ self.server_ad.droid.bluetoothRfcommStop()
+ for t in self.thread_list:
+ t.join()
+ self.thread_list.clear()
+ return True
+
+ def test_rfcomm_longev_read_write_binary_message(self):
+ """Longevity test an RFCOMM connection's I/O with a binary message
+
+ Test the longevity of RFCOMM with a basic read/write
+ connect/disconnect sequence. The data being transfered is in a
+ binary format.
+
+ Steps:
+ 1. Establish a bonding between two Android devices.
+ 2. Write data to RFCOMM from the client droid.
+ 3. Read data from RFCOMM from the server droid.
+ 4. Verify data written matches data read.
+ 5. Repeat steps 2-4 5000 times.
+ 6. Disconnect RFCOMM connection.
+ 7. Repeat steps 1-6 1000 times.
+
+ Expected Result:
+ Each iteration should read and write to the RFCOMM connection
+ successfully. Each connect and disconnect should be successful.
+
+ Returns:
+ Pass if True
+ Fail if False
+
+ TAGS: Classic, Longevity, RFCOMM
+ Priority: 2
+ """
+ server_mac = self.server_ad.droid.bluetoothGetLocalAddress()
+ #message = "1001010101" #todo: investigate why this fails...
+ binary_message = "11010101"
+ write_iterations = 5000
+ for i in range(self.longev_iterations):
+ self.log.info("iteration {} connection".format(i))
+ self.orchestrate_rfcomm_connect(server_mac)
+ for n in range(write_iterations):
+ self.log.info("iteration {} data".format(((n + 1) + (
+ i * write_iterations))))
+ self.log.info("Write message.")
+ self.client_ad.droid.bluetoothRfcommWriteBinary(binary_message)
+ self.log.info("Read message.")
+ read_msg = self.server_ad.droid.bluetoothRfcommReadBinary().rstrip("\r\n")
+ self.log.info("Verify message.")
+ assert binary_message == read_msg, "Mismatch! Read {}".format(
+ read_msg)
+ self.log.info("Iteration {} completed".format(n))
+ self.client_ad.droid.bluetoothRfcommStop()
+ self.server_ad.droid.bluetoothRfcommStop()
+ for t in self.thread_list:
+ t.join()
+ self.thread_list.clear()
+ return True
+
+ def test_rfcomm_longev_read_write_large_message(self):
+ """Longevity test an RFCOMM connection's I/O with a large message
+
+ Test the longevity of RFCOMM with a basic read/write
+ connect/disconnect sequence. The data being transfered is 990 chars
+ in size.
+
+ Steps:
+ 1. Establish a bonding between two Android devices.
+ 2. Write data to RFCOMM from the client droid.
+ 3. Read data from RFCOMM from the server droid.
+ 4. Verify data written matches data read.
+ 5. Repeat steps 2-4 5000 times.
+ 6. Disconnect RFCOMM connection.
+ 7. Repeat steps 1-6 1000 times.
+
+ Expected Result:
+ Each iteration should read and write to the RFCOMM connection
+ successfully. Each connect and disconnect should be successful.
+
+ Returns:
+ Pass if True
+ Fail if False
+
+ TAGS: Classic, Longevity, RFCOMM
+ Priority: 2
+ """
+ server_mac = self.server_ad.droid.bluetoothGetLocalAddress()
+ message = "x" * 990 #largest message size till sl4a fixed
+ write_iterations = 5000
+ for i in range(self.longev_iterations):
+ self.log.info("iteration {} connection".format(i))
+ self.orchestrate_rfcomm_connect(server_mac)
+ for n in range(write_iterations):
+ self.log.info("iteration {} data".format(((n + 1) + (
+ i * write_iterations))))
+ self.log.info("Write message.")
+ self.client_ad.droid.bluetoothRfcommWrite(message)
+ self.log.info("Read message.")
+ read_msg = self.server_ad.droid.bluetoothRfcommRead()
+ self.log.info("Verify message.")
+ assert message == read_msg, "Mismatch! Read {}".format(
+ read_msg)
+ self.log.info("Iteration {} completed".format(n))
+ self.client_ad.droid.bluetoothRfcommStop()
+ self.server_ad.droid.bluetoothRfcommStop()
+ for t in self.thread_list:
+ t.join()
+ self.thread_list.clear()
+ return True
+
+ def test_rfcomm_longev_connection_interuption(self):
+ """Longevity test an RFCOMM connection's with socket interuptions
+
+ Test the longevity of RFCOMM with a basic read/write
+ connect/disconnect sequence. Randomly in the sequence of reads and
+ writes the socket on the client side will close. There should be
+ an exception thrown for writing the next set of data and the
+ test should start up a new connection and continue.
+
+ Steps:
+ 1. Establish a bonding between two Android devices.
+ 2. Write data to RFCOMM from the client droid.
+ 3. Read data from RFCOMM from the server droid.
+ 4. Verify data written matches data read.
+ 5. Repeat steps 2-4 5000 times or until the random interupt occurs.
+ 6. Re-establish an RFCOMM connection.
+ 7. Repeat steps 1-6 1000 times.
+
+ Expected Result:
+ Each iteration should read and write to the RFCOMM connection
+ successfully. Each connect and disconnect should be successful.
+ Devices should recover a new connection after each interruption.
+
+ Returns:
+ Pass if True
+ Fail if False
+
+ TAGS: Classic, Longevity, RFCOMM
+ Priority: 2
+ """
+ server_mac = self.server_ad.droid.bluetoothGetLocalAddress()
+ write_iterations = 5000
+ for i in range(self.longev_iterations):
+ try:
+ self.log.info("iteration {} connection".format(i))
+ self.orchestrate_rfcomm_connect(server_mac)
+ random_interup_iteration = randint(0, write_iterations)
+ for n in range(write_iterations):
+ self.log.info("iteration {} data".format(((n + 1) + (
+ i * write_iterations))))
+ self.log.info("Write message.")
+ self.client_ad.droid.bluetoothRfcommWrite(
+ self.generic_message)
+ self.log.info("Read message.")
+ read_msg = self.server_ad.droid.bluetoothRfcommRead()
+ self.log.info("Verify message.")
+ assert self.generic_message == read_msg, "Mismatch! Read {}".format(
+ read_msg)
+ self.log.info("Iteration {} completed".format(n))
+ if n > random_interup_iteration:
+ self.client_ad.droid.bluetoothRfcommCloseSocket()
+ self.client_ad.droid.bluetoothRfcommStop()
+ self.server_ad.droid.bluetoothRfcommStop()
+ except Exception:
+ self.log.info("Exception found as expected. Continuing...")
+ try:
+ self.client_ad.droid.bluetoothRfcommStop()
+ except Exception as err:
+ self.log.error(
+ "Error closing client connection: {}".format(err))
+ return False
+ try:
+ self.server_ad.droid.bluetoothRfcommStop()
+ except Exception as err:
+ self.log.error(
+ "Error closing server connection: {}".format(err))
+ return False
+ for t in self.thread_list:
+ t.join()
+ self.thread_list.clear()
+ for t in self.thread_list:
+ t.join()
+ self.thread_list.clear()
+ return True
+
+ def test_rfcomm_longev_data_elasticity(self):
+ """Longevity test an RFCOMM connection's I/O with changing data size
+
+ Test the longevity of RFCOMM with a basic read/write
+ connect/disconnect sequence. The data being transfered changes
+ in size after each write/read sequence to increase up to 990
+ chars in size and decrease down to 1 in size. This repeats through
+ the entire test in order to exercise different size values being
+ written.
+
+ Steps:
+ 1. Establish a bonding between two Android devices.
+ 2. Write data to RFCOMM from the client droid.
+ 3. Read data from RFCOMM from the server droid.
+ 4. Verify data written matches data read.
+ 5. Change data size according to above description.
+ 6. Repeat steps 2-5 5000 times.
+ 7. Disconnect RFCOMM connection.
+ 8. Repeat steps 1-6 1000 times.
+
+ Expected Result:
+ Each iteration should read and write to the RFCOMM connection
+ successfully. Each connect and disconnect should be successful.
+
+ Returns:
+ Pass if True
+ Fail if False
+
+ TAGS: Classic, Longevity, RFCOMM
+ Priority: 2
+ """
+ server_mac = self.server_ad.droid.bluetoothGetLocalAddress()
+ message = "x"
+ resize_toggle = 1
+ write_iterations = 5000
+ for i in range(self.longev_iterations):
+ try:
+ self.log.info("iteration {} connection".format(i))
+ self.orchestrate_rfcomm_connect(server_mac)
+ for n in range(write_iterations):
+ self.log.info("iteration {} data".format(((n + 1) + (
+ i * write_iterations))))
+ self.log.info("Write message.")
+ self.client_ad.droid.bluetoothRfcommWrite(message)
+ self.log.info("Read message.")
+ read_msg = self.server_ad.droid.bluetoothRfcommRead()
+ self.log.info("Verify message.")
+ assert message == read_msg, "Mismatch! Read {}".format(
+ read_msg)
+ self.log.info("Iteration {} completed".format(n))
+ size_of_message = len(message)
+ #max size is 990 due to a bug in sl4a.
+ if size_of_message >= 990:
+ resize_toggle = 0
+ elif size_of_message <= 1:
+ resize_toggle = 1
+ if resize_toggle == 0:
+ message = "x" * (size_of_message - 1)
+ else:
+ message = "x" * (size_of_message + 1)
+ self.client_ad.droid.bluetoothRfcommStop()
+ self.server_ad.droid.bluetoothRfcommStop()
+ except Exception as err:
+ self.log.info("Error in longevity test: {}".format(err))
+ for t in self.thread_list:
+ t.join()
+ self.thread_list.clear()
+ return False
+
+ for t in self.thread_list:
+ t.join()
+ self.thread_list.clear()
+ return True
diff --git a/acts/tests/google/bt/system_tests/SppStressTest.py b/acts/tests/google/bt/system_tests/RfcommStressTest.py
similarity index 81%
rename from acts/tests/google/bt/system_tests/SppStressTest.py
rename to acts/tests/google/bt/system_tests/RfcommStressTest.py
index 23026a4..0f0b20e 100644
--- a/acts/tests/google/bt/system_tests/SppStressTest.py
+++ b/acts/tests/google/bt/system_tests/RfcommStressTest.py
@@ -32,7 +32,7 @@
from acts.test_utils.bt.bt_test_utils import take_btsnoop_logs
-class SppStressTest(BluetoothBaseTest):
+class RfcommStressTest(BluetoothBaseTest):
default_timeout = 10
scan_discovery_time = 5
thread_list = []
@@ -44,16 +44,8 @@
def __init__(self, controllers):
BluetoothBaseTest.__init__(self, controllers)
- self.server_ad = self.android_devices[0]
- self.client_ad = self.android_devices[1]
- self.tests = ("test_rfcomm_connection_stress",
- "test_rfcomm_read_write_stress", )
-
- def _clear_bonded_devices(self):
- for a in self.android_devices:
- bonded_device_list = a.droid.bluetoothGetConnectedDevices()
- for device in bonded_device_list:
- a.droid.bluetoothUnbond(device)
+ self.client_ad = self.android_devices[0]
+ self.server_ad = self.android_devices[1]
def on_fail(self, test_name, begin_time):
take_btsnoop_logs(self.android_devices, self, test_name)
@@ -96,15 +88,10 @@
Pass if True
Fail if False
- TAGS: Classic, Stress, RFCOMM, SPP
+ TAGS: Classic, Stress, RFCOMM
Priority: 1
"""
- server_mac = get_bt_mac_address(self.client_ad.droid,
- self.server_ad.droid)
- self._clear_bonded_devices()
- # temporary workaround. Need to find out why I can't connect after
- # I do a device discovery from get_bt_mac_address.
- reset_bluetooth([self.server_ad])
+ server_mac = self.server_ad.droid.bluetoothGetLocalAddress()
for n in range(1000):
self.orchestrate_rfcomm_connect(server_mac)
self.log.info("Write message.")
@@ -143,14 +130,10 @@
Pass if True
Fail if False
- TAGS: Classic, Stress, RFCOMM, SPP
+ TAGS: Classic, Stress, RFCOMM
Priority: 1
"""
- server_mac = get_bt_mac_address(self.client_ad.droid,
- self.server_ad.droid)
- self._clear_bonded_devices()
- # temporary workaround. Need to find out why I can't connect after
- # I do a device discovery from get_bt_mac_address.
+ server_mac = self.server_ad.droid.bluetoothGetLocalAddress()
reset_bluetooth([self.server_ad])
self.orchestrate_rfcomm_connect(server_mac)
for n in range(10000):
diff --git a/acts/tests/google/native/bt/BtNativeTest.py b/acts/tests/google/native/bt/BtNativeTest.py
index bba0891..9660a8e 100644
--- a/acts/tests/google/native/bt/BtNativeTest.py
+++ b/acts/tests/google/native/bt/BtNativeTest.py
@@ -16,27 +16,25 @@
import time
from acts.base_test import BaseTestClass
+from acts.controllers import native_android_device
from acts.test_utils.bt.native_bt_test_utils import setup_native_bluetooth
from acts.test_utils.bt.bt_test_utils import generate_id_by_size
+
class BtNativeTest(BaseTestClass):
tests = None
def __init__(self, controllers):
BaseTestClass.__init__(self, controllers)
- setup_native_bluetooth(self.native_android_devices)
- self.droid = self.native_android_devices[0].droid
- self.tests = (
- "test_binder_get_name",
- "test_binder_get_name_invalid_parameter",
- "test_binder_set_name_get_name",
- "test_binder_get_address",
- )
- if len(self.native_android_devices) > 1:
- self.droid1 = self.native_android_devices[1].droid
- self.tests = self.tests + (
- "test_two_devices_set_get_name",
- )
+ setup_native_bluetooth(self.native_devices)
+ self.droid = self.native_devices[0].droid
+ self.tests = ("test_binder_get_name",
+ "test_binder_get_name_invalid_parameter",
+ "test_binder_set_name_get_name",
+ "test_binder_get_address", )
+ if len(self.native_devices) > 1:
+ self.droid1 = self.native_devices[1].droid
+ self.tests = self.tests + ("test_two_devices_set_get_name", )
def test_binder_get_name(self):
result = self.droid.BtBinderGetName()
@@ -69,11 +67,10 @@
def test_two_devices_set_get_name(self):
test_name = generate_id_by_size(4)
- for n in self.native_android_devices:
+ for n in self.native_devices:
d = n.droid
d.BtBinderSetName(test_name)
name = d.BtBinderGetName()
if name != test_name:
return False
return True
-