#/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 exercises different ways Ble Advertisements can run in
concurrency. This test was designed to be run in a shield box.
"""
import concurrent
import os
import time
from queue import Empty
from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
from acts.test_utils.bt.BleEnum import AdvertiseSettingsAdvertiseMode
from acts.test_utils.bt.BleEnum import ScanSettingsCallbackType
from acts.test_utils.bt.BleEnum import ScanSettingsScanMode
from acts.test_utils.bt.bt_test_utils import adv_succ
from acts.test_utils.bt.bt_test_utils import generate_ble_advertise_objects
from acts.test_utils.bt.bt_test_utils import generate_ble_scan_objects
from acts.test_utils.bt.bt_test_utils import get_advanced_droid_list
from acts.test_utils.bt.bt_test_utils import reset_bluetooth
from acts.test_utils.bt.bt_test_utils import scan_result
from acts.test_utils.bt.bt_test_utils import take_btsnoop_logs
class ConcurrentBleAdvertisingTest(BluetoothBaseTest):
default_timeout = 10
max_advertisements = 4
def __init__(self, controllers):
BluetoothBaseTest.__init__(self, controllers)
self.droid_list = get_advanced_droid_list(self.android_devices)
self.scn_ad = self.android_devices[0]
self.adv_ad = self.android_devices[1]
self.max_advertisements = self.droid_list[1]['max_advertisements']
if self.max_advertisements == 0:
self.tests = ()
return
def on_fail(self, test_name, begin_time):
self.log.debug(
"Test {} failed. Gathering bugreport and btsnoop logs".format(
test_name))
take_btsnoop_logs(self.android_devices, self, test_name)
reset_bluetooth(self.android_devices)
def setup_test(self):
return reset_bluetooth(self.android_devices)
def _verify_n_advertisements(self, num_advertisements, filter_list):
test_result = False
address_list = []
self.scn_ad.droid.bleSetScanSettingsCallbackType(
ScanSettingsCallbackType.CALLBACK_TYPE_ALL_MATCHES.value)
self.scn_ad.droid.bleSetScanSettingsScanMode(
ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value)
self.adv_ad.droid.bleSetAdvertiseSettingsAdvertiseMode(
AdvertiseSettingsAdvertiseMode.ADVERTISE_MODE_LOW_LATENCY.value)
advertise_data = self.adv_ad.droid.bleBuildAdvertiseData()
advertise_settings = self.adv_ad.droid.bleBuildAdvertiseSettings()
advertise_callback_list = []
for i in range(num_advertisements):
advertise_callback = self.adv_ad.droid.bleGenBleAdvertiseCallback()
advertise_callback_list.append(advertise_callback)
self.adv_ad.droid.bleStartBleAdvertising(
advertise_callback, advertise_data, advertise_settings)
try:
self.adv_ad.ed.pop_event(
adv_succ.format(advertise_callback), self.default_timeout)
self.log.info("Advertisement {} started.".format(i + 1))
except Empty as error:
self.log.info("Advertisement {} failed to start.".format(i +
1))
self.log.debug("Test failed with Empty error: {}".format(
error))
return False
except concurrent.futures._base.TimeoutError as error:
self.log.debug(
"Test failed, filtering callback onSuccess never occurred: "
"{}".format(error))
return False
scan_settings = self.scn_ad.droid.bleBuildScanSetting()
scan_callback = self.scn_ad.droid.bleGenScanCallback()
self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
scan_callback)
start_time = time.time()
while (start_time + self.default_timeout) > time.time():
event = None
try:
event = self.scn_ad.ed.pop_event(
scan_result.format(scan_callback), self.default_timeout)
except Empty as error:
self.log.debug("Test failed with: {}".format(error))
return test_result
except concurrent.futures._base.TimeoutError as error:
self.log.debug("Test failed with: {}".format(error))
return test_result
address = event['data']['Result']['deviceInfo']['address']
if address not in address_list:
address_list.append(address)
if len(address_list) == num_advertisements:
test_result = True
break
for callback in advertise_callback_list:
self.adv_ad.droid.bleStopBleAdvertising(callback)
self.scn_ad.droid.bleStopBleScan(scan_callback)
return test_result
@BluetoothBaseTest.bt_test_wrap
def test_max_advertisements_defaults(self):
"""Testing max advertisements.
Test that a single device can have the max advertisements
concurrently advertising.
Steps:
1. Setup the scanning android device.
2. Setup the advertiser android device.
3. Start scanning on the max_advertisements as defined in the script.
4. Verify that all advertisements are found.
Expected Result:
All advertisements should start without errors.
Returns:
Pass if True
Fail if False
TAGS: LE, Advertising, Concurrency
Priority: 0
"""
test_result = True
filter_list = self.scn_ad.droid.bleGenFilterList()
self.scn_ad.droid.bleBuildScanFilter(filter_list)
test_result = self._verify_n_advertisements(self.max_advertisements,
filter_list)
return test_result
@BluetoothBaseTest.bt_test_wrap
def test_max_advertisements_include_device_name_and_filter_device_name(
self):
"""Testing max advertisement variant.
Test that a single device can have the max advertisements
concurrently advertising. Include the device name as a part of the filter
and advertisement data.
Steps:
1. Setup the scanning android device.
2. Setup the advertiser android device.
3. Include device name in each advertisement.
4. Include device name filter in the scanner.
5. Start scanning on the max_advertisements as defined in the script.
6. Verify that all advertisements are found.
Expected Result:
All advertisements should start without errors.
Returns:
Pass if True
Fail if False
TAGS: LE, Advertising, Concurrency
Priority: 2
"""
test_result = True
self.adv_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True)
filter_list = self.scn_ad.droid.bleGenFilterList()
self.scn_ad.droid.bleSetScanFilterDeviceName(
self.adv_ad.droid.bluetoothGetLocalName())
self.scn_ad.droid.bleBuildScanFilter(filter_list)
test_result = self._verify_n_advertisements(self.max_advertisements,
filter_list)
return test_result
@BluetoothBaseTest.bt_test_wrap
def test_max_advertisements_exclude_device_name_and_filter_device_name(
self):
"""Test max advertisement variant.
Test that a single device can have the max advertisements concurrently
advertising. Include the device name as a part of the filter but not the
advertisement data.
Steps:
1. Setup the scanning android device.
2. Setup the advertiser android device.
3. Include device name filter in the scanner.
4. Start scanning on the max_advertisements as defined in the script.
5. Verify that no advertisements are found.
Expected Result:
All advertisements should start without errors.
Returns:
Pass if True
Fail if False
TAGS: LE, Advertising, Concurrency
Priority: 2
"""
test_result = True
self.adv_ad.droid.bleSetAdvertiseDataIncludeDeviceName(False)
filter_list = self.scn_ad.droid.bleGenFilterList()
self.scn_ad.droid.bleSetScanFilterDeviceName(
self.adv_ad.droid.bluetoothGetLocalName())
self.scn_ad.droid.bleBuildScanFilter(filter_list)
test_result = self._verify_n_advertisements(self.max_advertisements,
filter_list)
return not test_result
@BluetoothBaseTest.bt_test_wrap
def test_max_advertisements_with_manufacturer_data(self):
"""Test max advertisement variant.
Test that a single device can have the max advertisements concurrently
advertising. Include the manufacturer data as a part of the filter and
advertisement data.
Steps:
1. Setup the scanning android device.
2. Setup the advertiser android device.
3. Include manufacturer data in each advertisement.
4. Include manufacturer data filter in the scanner.
5. Start scanning on the max_advertisements as defined in the script.
6. Verify that all advertisements are found.
Expected Result:
All advertisements should start without errors.
Returns:
Pass if True
Fail if False
TAGS: LE, Advertising, Concurrency
Priority: 2
"""
test_result = True
filter_list = self.scn_ad.droid.bleGenFilterList()
self.scn_ad.droid.bleSetScanFilterManufacturerData(1, "1")
self.scn_ad.droid.bleBuildScanFilter(filter_list)
self.adv_ad.droid.bleAddAdvertiseDataManufacturerId(1, "1")
test_result = self._verify_n_advertisements(self.max_advertisements,
filter_list)
return test_result
@BluetoothBaseTest.bt_test_wrap
def test_max_advertisements_with_manufacturer_data_mask(self):
"""Test max advertisements variant.
Test that a single device can have the max advertisements concurrently
advertising. Include the manufacturer data mask as a part of the filter
and advertisement data.
Steps:
1. Setup the scanning android device.
2. Setup the advertiser android device.
3. Include manufacturer data in each advertisement.
4. Include manufacturer data mask filter in the scanner.
5. Start scanning on the max_advertisements as defined in the script.
6. Verify that all advertisements are found.
Expected Result:
All advertisements should start without errors.
Returns:
Pass if True
Fail if False
TAGS: LE, Advertising, Concurrency
Priority: 2
"""
test_result = True
filter_list = self.scn_ad.droid.bleGenFilterList()
self.scn_ad.droid.bleSetScanFilterManufacturerData(1, "1", "1")
self.scn_ad.droid.bleBuildScanFilter(filter_list)
self.adv_ad.droid.bleAddAdvertiseDataManufacturerId(1, "1")
test_result = self._verify_n_advertisements(self.max_advertisements,
filter_list)
return test_result
@BluetoothBaseTest.bt_test_wrap
def test_max_advertisements_with_service_data(self):
"""Test max advertisement variant.
Test that a single device can have the max advertisements concurrently
advertising. Include the service data as a part of the filter and
advertisement data.
Steps:
1. Setup the scanning android device.
2. Setup the advertiser android device.
3. Include service data in each advertisement.
4. Include service data filter in the scanner.
5. Start scanning on the max_advertisements as defined in the script.
6. Verify that all advertisements are found.
Expected Result:
All advertisements should start without errors.
Returns:
Pass if True
Fail if False
TAGS: LE, Advertising, Concurrency
Priority: 2
"""
test_result = True
filter_list = self.scn_ad.droid.bleGenFilterList()
self.scn_ad.droid.bleSetScanFilterServiceData(
"0000110A-0000-1000-8000-00805F9B34FB", "11,17,80")
self.scn_ad.droid.bleBuildScanFilter(filter_list)
self.adv_ad.droid.bleAddAdvertiseDataServiceData(
"0000110A-0000-1000-8000-00805F9B34FB", "11,17,80")
test_result = self._verify_n_advertisements(self.max_advertisements,
filter_list)
return test_result
@BluetoothBaseTest.bt_test_wrap
def test_max_advertisements_with_manufacturer_data_mask_and_include_device_name(
self):
"""Test max advertisement variant.
Test that a single device can have the max advertisements concurrently
advertising. Include the device name and manufacturer data as a part of
the filter and advertisement data.
Steps:
1. Setup the scanning android device.
2. Setup the advertiser android device.
3. Include device name and manufacturer data in each advertisement.
4. Include device name and manufacturer data filter in the scanner.
5. Start scanning on the max_advertisements as defined in the script.
6. Verify that all advertisements are found.
Expected Result:
All advertisements should start without errors.
Returns:
Pass if True
Fail if False
TAGS: LE, Advertising, Concurrency
Priority: 2
"""
test_result = True
filter_list = self.scn_ad.droid.bleGenFilterList()
self.adv_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True)
self.scn_ad.droid.bleSetScanFilterDeviceName(
self.adv_ad.droid.bluetoothGetLocalName())
self.scn_ad.droid.bleSetScanFilterManufacturerData(1, "1", "1")
self.scn_ad.droid.bleBuildScanFilter(filter_list)
self.adv_ad.droid.bleAddAdvertiseDataManufacturerId(1, "1")
test_result = self._verify_n_advertisements(self.max_advertisements,
filter_list)
return test_result
@BluetoothBaseTest.bt_test_wrap
def test_max_advertisements_with_service_uuids(self):
"""Test max advertisement variant.
Test that a single device can have the max advertisements concurrently
advertising. Include the service uuid as a part of the filter and
advertisement data.
Steps:
1. Setup the scanning android device.
2. Setup the advertiser android device.
3. Include service uuid in each advertisement.
4. Include service uuid filter in the scanner.
5. Start scanning on the max_advertisements as defined in the script.
6. Verify that all advertisements are found.
Expected Result:
All advertisements should start without errors.
Returns:
Pass if True
Fail if False
TAGS: LE, Advertising, Concurrency
Priority: 1
"""
test_result = True
filter_list = self.scn_ad.droid.bleGenFilterList()
self.scn_ad.droid.bleSetScanFilterServiceUuid(
"00000000-0000-1000-8000-00805f9b34fb")
self.scn_ad.droid.bleBuildScanFilter(filter_list)
self.adv_ad.droid.bleSetAdvertiseDataSetServiceUuids(
["00000000-0000-1000-8000-00805f9b34fb"])
test_result = self._verify_n_advertisements(self.max_advertisements,
filter_list)
return test_result
@BluetoothBaseTest.bt_test_wrap
def test_max_advertisements_with_service_uuid_and_service_mask(self):
"""Test max advertisements variant.
Test that a single device can have the max advertisements concurrently
advertising. Include the service mask as a part of the filter and
advertisement data.
Steps:
1. Setup the scanning android device.
2. Setup the advertiser android device.
3. Include service uuid in each advertisement.
4. Include service mask filter in the scanner.
5. Start scanning on the max_advertisements as defined in the script.
6. Verify that all advertisements are found.
Expected Result:
All advertisements should start without errors.
Returns:
Pass if True
Fail if False
TAGS: LE, Advertising, Concurrency
Priority: 2
"""
test_result = True
filter_list = self.scn_ad.droid.bleGenFilterList()
self.scn_ad.droid.bleSetScanFilterServiceUuid(
"00000000-0000-1000-8000-00805f9b34fb",
"00000000-0000-1000-8000-00805f9b34fb")
self.scn_ad.droid.bleBuildScanFilter(filter_list)
self.adv_ad.droid.bleSetAdvertiseDataSetServiceUuids(
["00000000-0000-1000-8000-00805f9b34fb"])
test_result = self._verify_n_advertisements(self.max_advertisements,
filter_list)
return test_result
@BluetoothBaseTest.bt_test_wrap
def test_max_advertisements_plus_one(self):
"""Test max advertisements plus one.
Test that a single device can have the max advertisements concurrently
advertising but fail on starting the max advertisements plus one.
filter and advertisement data.
Steps:
1. Setup the scanning android device.
2. Setup the advertiser android device.
3. Start max_advertisements + 1.
Expected Result:
The last advertisement should fail.
Returns:
Pass if True
Fail if False
TAGS: LE, Advertising, Concurrency
Priority: 0
"""
test_result = True
filter_list = self.scn_ad.droid.bleGenFilterList()
self.scn_ad.droid.bleBuildScanFilter(filter_list)
test_result = self._verify_n_advertisements(
self.max_advertisements + 1, filter_list)
return not test_result
@BluetoothBaseTest.bt_test_wrap
def test_start_two_advertisements_on_same_callback(self):
"""Test invalid advertisement scenario.
Test that a single device cannot have two advertisements start on the
same callback.
Steps:
1. Setup the scanning android device.
2. Setup the advertiser android device.
3. Call start ble advertising on the same callback.
Expected Result:
The second call of start advertising on the same callback should fail.
Returns:
Pass if True
Fail if False
TAGS: LE, Advertising, Concurrency
Priority: 1
"""
test_result = True
advertise_callback, advertise_data, advertise_settings = (
generate_ble_advertise_objects(self.adv_ad.droid))
self.adv_ad.droid.bleStartBleAdvertising(
advertise_callback, advertise_data, advertise_settings)
try:
self.adv_ad.ed.pop_event(
adv_succ.format(advertise_callback), self.default_timeout)
except Empty as error:
self.log.debug("Test failed with Empty error: {}".format(error))
return False
except concurrent.futures._base.TimeoutError as error:
self.log.debug(
"Test failed, filtering callback onSuccess never occurred: {}"
.format(error))
try:
self.adv_ad.droid.bleStartBleAdvertising(
advertise_callback, advertise_data, advertise_settings)
self.adv_ad.ed.pop_event(
adv_succ.format(advertise_callback), self.default_timeout)
test_result = False
except Empty as error:
self.log.debug("Test passed with Empty error: {}".format(error))
except concurrent.futures._base.TimeoutError as error:
self.log.debug(
"Test passed, filtering callback onSuccess never occurred: {}"
.format(error))
return test_result
@BluetoothBaseTest.bt_test_wrap
def test_toggle_advertiser_bt_state(self):
"""Test forcing stopping advertisements.
Test that a single device resets its callbacks when the bluetooth state is
reset. There should be no advertisements.
Steps:
1. Setup the scanning android device.
2. Setup the advertiser android device.
3. Call start ble advertising.
4. Toggle bluetooth on and off.
5. Scan for any advertisements.
Expected Result:
No advertisements should be found after toggling Bluetooth on the
advertising device.
Returns:
Pass if True
Fail if False
TAGS: LE, Advertising, Concurrency
Priority: 2
"""
test_result = True
advertise_callback, advertise_data, advertise_settings = (
generate_ble_advertise_objects(self.adv_ad.droid))
self.adv_ad.droid.bleStartBleAdvertising(
advertise_callback, advertise_data, advertise_settings)
try:
self.adv_ad.ed.pop_event(
adv_succ.format(advertise_callback), self.default_timeout)
except Empty as error:
self.log.debug("Test failed with Empty error: {}".format(error))
return False
except concurrent.futures._base.TimeoutError as error:
self.log.debug(
"Test failed, filtering callback onSuccess never occurred: {}".format(
error))
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)
try:
self.scn_ad.ed.pop_event(
scan_result.format(scan_callback), self.default_timeout)
except Empty as error:
self.log.debug("Test failed with: {}".format(error))
return False
except concurrent.futures._base.TimeoutError as error:
self.log.debug("Test failed with: {}".format(error))
return False
self.scn_ad.droid.bleStopBleScan(scan_callback)
test_result = reset_bluetooth([self.android_devices[1]])
self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
scan_callback)
if not test_result:
return test_result
try:
self.scn_ad.ed.pop_event(
scan_result.format(scan_callback), self.default_timeout)
return False
except Empty as error:
self.log.debug("Test passed with: {}".format(error))
except concurrent.futures._base.TimeoutError as error:
self.log.debug("Test passed with: {}".format(error))
self.scn_ad.droid.bleStopBleScan(scan_callback)
self.adv_ad.droid.bleStopBleAdvertising(advertise_callback)
return test_result
@BluetoothBaseTest.bt_test_wrap
def test_restart_advertise_callback_after_bt_toggle(self):
"""Test starting an advertisement on a cleared out callback.
Test that a single device resets its callbacks when the bluetooth state
is reset.
Steps:
1. Setup the scanning android device.
2. Setup the advertiser android device.
3. Call start ble advertising.
4. Toggle bluetooth on and off.
5. Call start ble advertising on the same callback.
Expected Result:
Starting an advertisement on a callback id after toggling bluetooth
should fail.
Returns:
Pass if True
Fail if False
TAGS: LE, Advertising, Concurrency
Priority: 1
"""
test_result = True
advertise_callback, advertise_data, advertise_settings = (
generate_ble_advertise_objects(self.adv_ad.droid))
self.adv_ad.droid.bleStartBleAdvertising(
advertise_callback, advertise_data, advertise_settings)
try:
self.adv_ad.ed.pop_event(
adv_succ.format(advertise_callback), self.default_timeout)
except Empty as error:
self.log.debug("Test failed with Empty error: {}".format(error))
test_result = False
except concurrent.futures._base.TimeoutError as error:
self.log.debug(
"Test failed, filtering callback onSuccess never occurred: {}".format(
error))
test_result = reset_bluetooth([self.android_devices[1]])
if not test_result:
return test_result
self.adv_ad.droid.bleStartBleAdvertising(
advertise_callback, advertise_data, advertise_settings)
try:
self.adv_ad.ed.pop_event(
adv_succ.format(advertise_callback), self.default_timeout)
except Empty as error:
self.log.debug("Test failed with Empty error: {}".format(error))
test_result = False
except concurrent.futures._base.TimeoutError as error:
self.log.debug(
"Test failed, filtering callback onSuccess never occurred: {}".format(
error))
return test_result