Snap for 4826885 from 8375a937d006cdec4ceca0110288471ea6a69bd9 to pi-release
Change-Id: I2d8ebfc2c0bc4448bc4393c6fb374c4e6466f7e4
diff --git a/acts/framework/acts/test_utils/bt/BluetoothBaseTest.py b/acts/framework/acts/test_utils/bt/BluetoothBaseTest.py
index cded9f5..01cb76a 100644
--- a/acts/framework/acts/test_utils/bt/BluetoothBaseTest.py
+++ b/acts/framework/acts/test_utils/bt/BluetoothBaseTest.py
@@ -29,6 +29,13 @@
from acts.test_utils.bt.bt_test_utils import reset_bluetooth
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
+from acts.test_utils.bt.ble_lib import BleLib
+from acts.test_utils.bt.bta_lib import BtaLib
+from acts.test_utils.bt.config_lib import ConfigLib
+from acts.test_utils.bt.gattc_lib import GattClientLib
+from acts.test_utils.bt.gatts_lib import GattServerLib
+from acts.test_utils.bt.rfcomm_lib import RfcommLib
+from acts.test_utils.bt.shell_commands_lib import ShellCommands
class BluetoothBaseTest(BaseTestClass):
@@ -38,6 +45,8 @@
def __init__(self, controllers):
BaseTestClass.__init__(self, controllers)
+ for ad in self.android_devices:
+ self._setup_bt_libs(ad)
# Use for logging in the test cases to facilitate
# faster log lookup and reduce ambiguity in logging.
@@ -109,8 +118,9 @@
return True
def on_fail(self, test_name, begin_time):
- self.log.debug("Test {} failed. Gathering bugreport and btsnoop logs".
- format(test_name))
+ self.log.debug(
+ "Test {} failed. Gathering bugreport and btsnoop logs".format(
+ test_name))
take_btsnoop_logs(self.android_devices, self, test_name)
self._take_bug_report(test_name, begin_time)
for _ in range(5):
@@ -142,3 +152,26 @@
self.log.info("Total items in list {}".format(
len(self.timer_list)))
self.timer_list = []
+
+ def _setup_bt_libs(self, android_device):
+ # Bluetooth Low Energy library.
+ setattr(android_device, "ble", BleLib(
+ log=self.log, dut=android_device))
+ # Bluetooth Adapter library.
+ setattr(android_device, "bta", BtaLib(
+ log=self.log, dut=android_device))
+ # Bluetooth stack config library.
+ setattr(android_device, "config",
+ ConfigLib(log=self.log, dut=android_device))
+ # GATT Client library.
+ setattr(android_device, "gattc",
+ GattClientLib(log=self.log, dut=android_device))
+ # GATT Server library.
+ setattr(android_device, "gatts",
+ GattServerLib(log=self.log, dut=android_device))
+ # RFCOMM library.
+ setattr(android_device, "rfcomm",
+ RfcommLib(log=self.log, dut=android_device))
+ # Shell command library
+ setattr(android_device, "shell",
+ ShellCommands(log=self.log, dut=android_device))
diff --git a/acts/framework/acts/test_utils/bt/ble_lib.py b/acts/framework/acts/test_utils/bt/ble_lib.py
index 379f182..3bbaede 100644
--- a/acts/framework/acts/test_utils/bt/ble_lib.py
+++ b/acts/framework/acts/test_utils/bt/ble_lib.py
@@ -21,10 +21,10 @@
from acts.test_utils.bt.bt_constants import ble_advertise_settings_tx_powers
from acts.test_utils.bt.bt_constants import ble_scan_settings_modes
from acts.test_utils.bt.bt_constants import small_timeout
-from acts.test_utils.bt.bt_test_utils import adv_fail
+from acts.test_utils.bt.bt_constants import adv_fail
from acts.test_utils.bt.bt_constants import adv_succ
-from acts.test_utils.bt.bt_test_utils import advertising_set_on_own_address_read
-from acts.test_utils.bt.bt_test_utils import advertising_set_started
+from acts.test_utils.bt.bt_constants import advertising_set_on_own_address_read
+from acts.test_utils.bt.bt_constants import advertising_set_started
from acts.test_utils.bt.bt_test_utils import generate_ble_advertise_objects
import time
@@ -32,11 +32,10 @@
class BleLib():
- def __init__(self, log, mac_addr, dut):
+ def __init__(self, log, dut):
self.advertisement_list = []
self.dut = dut
self.log = log
- self.mac_addr = mac_addr
self.default_timeout = 5
self.set_advertisement_list = []
self.generic_uuid = "0000{}-0000-1000-8000-00805f9b34fb"
@@ -178,3 +177,35 @@
return
self.dut.droid.bleStopBleAdvertising(callback_id)
self.advertisement_list.remove(callback_id)
+
+ def start_max_advertisements(self, line):
+ scan_response = None
+ if line:
+ scan_response = bool(line)
+ while (True):
+ try:
+ self.dut.droid.bleSetAdvertiseSettingsAdvertiseMode(
+ ble_advertise_settings_modes['low_latency'])
+ self.dut.droid.bleSetAdvertiseSettingsIsConnectable(True)
+ advertise_callback, advertise_data, advertise_settings = (
+ generate_ble_advertise_objects(self.dut.droid))
+ if scan_response:
+ self.dut.droid.bleStartBleAdvertisingWithScanResponse(
+ advertise_callback, advertise_data, advertise_settings,
+ advertise_data)
+ else:
+ self.dut.droid.bleStartBleAdvertising(
+ advertise_callback, advertise_data, advertise_settings)
+ if self._verify_ble_adv_started(advertise_callback):
+ self.log.info(
+ "Tracking Callback ID: {}".format(advertise_callback))
+ self.advertisement_list.append(advertise_callback)
+ self.log.info(self.advertisement_list)
+ else:
+ self.log.info("Advertisements active: {}".format(
+ len(self.advertisement_list)))
+ return False
+ except Exception as err:
+ self.log.info("Advertisements active: {}".format(
+ len(self.advertisement_list)))
+ return True
diff --git a/acts/framework/acts/test_utils/bt/bt_carkit_lib.py b/acts/framework/acts/test_utils/bt/bt_carkit_lib.py
new file mode 100644
index 0000000..f1fa9fa
--- /dev/null
+++ b/acts/framework/acts/test_utils/bt/bt_carkit_lib.py
@@ -0,0 +1,830 @@
+#/usr/bin/env python3.4
+#
+# Copyright (C) 2018 The Android Open Source Project
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+
+import time
+import os
+
+from acts.keys import Config
+from acts.utils import rand_ascii_str
+from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
+from acts.test_utils.bt.bt_constants import logcat_strings
+from acts.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
+from acts.test_utils.tel.tel_defines import AUDIO_ROUTE_BLUETOOTH
+from acts.test_utils.tel.tel_defines import AUDIO_ROUTE_EARPIECE
+from acts.test_utils.tel.tel_defines import AUDIO_ROUTE_SPEAKER
+from acts.test_utils.tel.tel_defines import WAIT_TIME_IN_CALL
+from acts.test_utils.tel.tel_test_utils import get_phone_number
+from acts.test_utils.tel.tel_test_utils import hangup_call
+from acts.test_utils.tel.tel_test_utils import initiate_call
+from acts.test_utils.tel.tel_test_utils import num_active_calls
+from acts.test_utils.tel.tel_test_utils import sms_send_receive_verify
+from acts.test_utils.tel.tel_test_utils import wait_and_answer_call
+from acts.test_utils.tel.tel_voice_utils import get_audio_route
+from acts.test_utils.tel.tel_voice_utils import set_audio_route
+from acts.test_utils.tel.tel_voice_utils import swap_calls
+from acts.test_utils.tel.tel_test_utils import toggle_airplane_mode
+from acts.test_utils.tel.tel_test_utils import call_setup_teardown
+from acts.utils import exe_cmd
+from acts.utils import get_current_epoch_time
+
+KEYCODE_VOLUME_UP = "input keyevent 24"
+KEYCODE_VOLUME_DOWN = "input keyevent 25"
+KEYCODE_EVENT_PLAY_PAUSE = "input keyevent 85"
+KEYCODE_MEDIA_STOP = "input keyevent 86"
+KEYCODE_EVENT_NEXT = "input keyevent 87"
+KEYCODE_EVENT_PREVIOUS = "input keyevent 88"
+KEYCODE_MEDIA_REWIND = "input keyevent 89"
+KEYCODE_MEDIA_FAST_FORWARD = "input keyevent 90"
+KEYCODE_MUTE = "input keyevent 91"
+
+default_timeout = 10
+
+
+class E2eBtCarkitLib():
+
+ android_devices = []
+ short_timeout = 3
+ active_call_id = None
+ hold_call_id = None
+ log = None
+ mac_address = None
+
+ def __init__(self, log, target_mac_address=None):
+ self.log = log
+ self.target_mac_address = target_mac_address
+
+ def connect_hsp_helper(self, ad):
+ end_time = time.time() + default_timeout + 10
+ connected_hsp_devices = len(ad.droid.bluetoothHspGetConnectedDevices())
+ while connected_hsp_devices != 1 and time.time() < end_time:
+ try:
+ ad.droid.bluetoothHspConnect(self.target_mac_address)
+ time.sleep(3)
+ if len(ad.droid.bluetoothHspGetConnectedDevices() == 1):
+ break
+ except Exception:
+ self.log.debug("Failed to connect hsp trying again...")
+ try:
+ ad.droid.bluetoothConnectBonded(self.target_mac_address)
+ except Exception:
+ self.log.info("Failed to connect to bonded device...")
+ connected_hsp_devices = len(
+ ad.droid.bluetoothHspGetConnectedDevices())
+ if connected_hsp_devices != 1:
+ self.log.error("Failed to reconnect to HSP service...")
+ return False
+ self.log.info("Connected to HSP service...")
+ return True
+
+ def setup_multi_call(self, caller0, caller1, callee):
+ outgoing_num = get_phone_number(self.log, callee)
+ if not initiate_call(self.log, caller0, outgoing_num):
+ self.log.error("Failed to initiate call")
+ return False
+ if not wait_and_answer_call(self.log, callee):
+ self.log.error("Failed to answer call.")
+ return False
+ time.sleep(self.short_timeout)
+ if not initiate_call(self.log, caller1, outgoing_num):
+ self.log.error("Failed to initiate call")
+ return False
+ if not wait_and_answer_call(self.log, callee):
+ self.log.error("Failed to answer call.")
+ return False
+ return True
+
+ def process_tests(self, tests):
+ for test in tests:
+ try:
+ test()
+ except Exception as err:
+ self.log.error(err)
+
+ def run_suite_hfp_tests(self):
+ tests = [
+ self.outgoing_call_private_number,
+ self.outgoing_call_unknown_contact,
+ self.incomming_call_private_number,
+ self.incomming_call_unknown_contact,
+ self.outgoing_call_multiple_iterations,
+ self.outgoing_call_hsp_disabled_then_enabled_during_call,
+ self.call_audio_routes,
+ self.sms_during_incomming_call,
+ self.multi_incomming_call,
+ self.multi_call_audio_routing,
+ self.multi_call_swap_multiple_times,
+ self.outgoing_call_a2dp_play_before_and_after,
+ ]
+ _process_tests(tests)
+
+ def run_suite_hfp_conf_tests(self):
+ tests = [
+ self.multi_call_join_conference_call,
+ self.multi_call_join_conference_call_hangup_conf_call,
+ self.outgoing_multi_call_join_conference_call,
+ self.multi_call_join_conference_call_audio_routes,
+ ]
+ _process_tests(tests)
+
+ def run_suite_map_tests(self):
+ tests = [
+ self.sms_receive_different_sizes,
+ self.sms_receive_multiple,
+ self.sms_send_outgoing_texts,
+ ]
+ _process_tests(tests)
+
+ def run_suite_avrcp_tests(self):
+ tests = [
+ self.avrcp_play_pause,
+ self.avrcp_next_previous_song,
+ self.avrcp_next_previous,
+ self.avrcp_next_repetative,
+ ]
+ _process_tests(tests)
+
+ def disconnect_reconnect_multiple_iterations(self, pri_dut):
+ iteration_count = 5
+ self.log.info(
+ "Test disconnect-reconnect scenario from phone {} times.".format(
+ iteration_count))
+ self.log.info(
+ "This test will prompt for user interaction after each reconnect.")
+ input("Press enter to execute this testcase...")
+ #Assumes only one devices connected
+ grace_timeout = 4 #disconnect and reconnect timeout
+ for n in range(iteration_count):
+ self.log.info("Test iteration {}.".format(n + 1))
+ self.log.info("Disconnecting device {}...".format(
+ self.target_mac_address))
+ pri_dut.droid.bluetoothDisconnectConnected(self.target_mac_address)
+ # May have to do a longer sleep for carkits.... need to test
+ time.sleep(grace_timeout)
+ self.log.info("Connecting device {}...".format(
+ self.target_mac_address))
+ pri_dut.droid.bluetoothConnectBonded(self.target_mac_address)
+ if not self.connect_hsp_helper(pri_dut):
+ return False
+ start_time = time.time()
+ connected_devices = pri_dut.droid.bluetoothGetConnectedDevices()
+ self.log.info(
+ "Waiting up to 10 seconds for device to reconnect...")
+ while time.time() < start_time + 10 and len(connected_devices) != 1:
+ connected_devices = pri_dut.droid.bluetoothGetConnectedDevices(
+ )
+ time.sleep(1)
+ if len(connected_devices) != 1:
+ self.log.error(
+ "Failed to reconnect at iteration {}... continuing".format(
+ n))
+ return False
+ input("Continue to next iteration?")
+ return True
+
+ def disconnect_a2dp_only_then_reconnect(self, pri_dut):
+ self.log.info(
+ "Test disconnect-reconnect a2dp only scenario from phone.")
+ input("Press enter to execute this testcase...")
+ if not pri_dut.droid.bluetoothA2dpDisconnect(self.target_mac_address):
+ self.log.error("Failed to disconnect A2DP service...")
+ return False
+ time.sleep(self.short_timeout)
+ result = input("Confirm A2DP disconnected? (Y/n) ")
+ if result == "n":
+ self.log.error(
+ "Tester confirmed that A2DP did not disconnect. Failing test.")
+ return False
+ if len(pri_dut.droid.bluetoothA2dpGetConnectedDevices()) != 0:
+ self.log.error("Failed to disconnect from A2DP service")
+ return False
+ pri_dut.droid.bluetoothA2dpConnect(self.target_mac_address)
+ time.sleep(self.short_timeout)
+ if len(pri_dut.droid.bluetoothA2dpGetConnectedDevices()) != 1:
+ self.log.error("Failed to reconnect to A2DP service...")
+ return False
+ return True
+
+ def disconnect_hsp_only_then_reconnect(self, pri_dut):
+ self.log.info(
+ "Test disconnect-reconnect hsp only scenario from phone.")
+ input("Press enter to execute this testcase...")
+ if not pri_dut.droid.bluetoothHspDisconnect(self.target_mac_address):
+ self.log.error("Failed to disconnect HSP service...")
+ return False
+ time.sleep(self.short_timeout)
+ result = input("Confirm HFP disconnected? (Y/n) ")
+ pri_dut.droid.bluetoothHspConnect(self.target_mac_address)
+ time.sleep(self.short_timeout)
+ if len(pri_dut.droid.bluetoothHspGetConnectedDevices()) != 1:
+ self.log.error("Failed to connect from HSP service")
+ return False
+ return True
+
+ def disconnect_both_hsp_and_a2dp_then_reconnect(self, pri_dut):
+ self.log.info(
+ "Test disconnect-reconnect hsp and a2dp scenario from phone.")
+ input("Press enter to execute this testcase...")
+ if not pri_dut.droid.bluetoothA2dpDisconnect(self.target_mac_address):
+ self.log.error("Failed to disconnect A2DP service...")
+ return False
+ if not pri_dut.droid.bluetoothHspDisconnect(self.target_mac_address):
+ self.log.error("Failed to disconnect HSP service...")
+ return False
+ time.sleep(self.short_timeout)
+ if len(pri_dut.droid.bluetoothA2dpGetConnectedDevices()) != 0:
+ self.log.error("Failed to disconnect from A2DP service")
+ return False
+ if len(pri_dut.droid.bluetoothHspGetConnectedDevices()) != 0:
+ self.log.error("Failed to disconnect from HSP service")
+ return False
+ result = input("Confirm HFP and A2DP disconnected? (Y/n) ")
+ pri_dut.droid.bluetoothConnectBonded(self.target_mac_address)
+ time.sleep(self.short_timeout)
+ if len(pri_dut.droid.bluetoothA2dpGetConnectedDevices()) != 1:
+ self.log.error("Failed to reconnect to A2DP service...")
+ return False
+ if not self.connect_hsp_helper(pri_dut):
+ return False
+ return True
+
+ def outgoing_call_private_number(self, pri_dut, ter_dut):
+ self.log.info(
+ "Test outgoing call scenario from phone to private number")
+ input("Press enter to execute this testcase...")
+ outgoing_num = "*67" + get_phone_number(self.log, ter_dut)
+ if not initiate_call(self.log, pri_dut, outgoing_num):
+ self.log.error("Failed to initiate call")
+ return False
+ if not wait_and_answer_call(self.log, ter_dut):
+ self.log.error("Failed to answer call.")
+ return False
+ time.sleep(self.short_timeout)
+ input("Press enter to hangup call...")
+ if not hangup_call(self.log, pri_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return True
+
+ def outgoing_call_a2dp_play_before_and_after(self, pri_dut, sec_dut):
+ self.log.info(
+ "Test outgoing call scenario while playing music. Music should resume after call."
+ )
+ pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
+ input(
+ "Press enter to execute this testcase when music is in a play state..."
+ )
+ outgoing_num = get_phone_number(self.log, sec_dut)
+ if not initiate_call(self.log, pri_dut, outgoing_num):
+ self.log.error("Failed to initiate call")
+ return False
+ if not wait_and_answer_call(self.log, sec_dut):
+ self.log.error("Failed to answer call.")
+ return False
+ time.sleep(self.short_timeout)
+ input("Press enter to hangup call...")
+ if not hangup_call(self.log, pri_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ input("Press enter when music continues to play.")
+ self.log.info("Pausing Music...")
+ pri_dut.adb.shell(KEYCODE_EVENT_PLAY_PAUSE)
+ return True
+
+ def outgoing_call_unknown_contact(self, pri_dut, ter_dut):
+ self.log.info(
+ "Test outgoing call scenario from phone to unknow contact")
+ input("Press enter to execute this testcase...")
+ outgoing_num = get_phone_number(self.log, ter_dut)
+ if not initiate_call(self.log, pri_dut, outgoing_num):
+ self.log.error("Failed to initiate call")
+ return False
+ if not wait_and_answer_call(self.log, ter_dut):
+ self.log.error("Failed to answer call.")
+ return False
+ time.sleep(self.short_timeout)
+ input("Press enter to hangup call...")
+ if not hangup_call(self.log, pri_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return True
+
+ def incomming_call_private_number(self, pri_dut, ter_dut):
+ self.log.info(
+ "Test incomming call scenario to phone from private number")
+ input("Press enter to execute this testcase...")
+ outgoing_num = "*67" + get_phone_number(self.log, pri_dut)
+ if not initiate_call(self.log, ter_dut, outgoing_num):
+ self.log.error("Failed to initiate call")
+ return False
+ if not wait_and_answer_call(self.log, pri_dut):
+ self.log.error("Failed to answer call.")
+ return False
+ time.sleep(self.short_timeout)
+ input("Press enter to hangup call...")
+ if not hangup_call(self.log, ter_dut):
+ self.log.error("Failed to hangup call")
+ return False
+
+ return True
+
+ def incomming_call_unknown_contact(self, pri_dut, ter_dut):
+ self.log.info(
+ "Test incomming call scenario to phone from unknown contact")
+ input("Press enter to execute this testcase...")
+ outgoing_num = get_phone_number(self.log, pri_dut)
+ if not initiate_call(self.log, ter_dut, outgoing_num):
+ self.log.error("Failed to initiate call")
+ return False
+ if not wait_and_answer_call(self.log, pri_dut):
+ self.log.error("Failed to answer call.")
+ return False
+ time.sleep(self.short_timeout)
+ input("Press enter to hangup call...")
+ if not hangup_call(self.log, ter_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return True
+
+ def outgoing_call_multiple_iterations(self, pri_dut, sec_dut):
+ iteration_count = 3
+ self.log.info(
+ "Test outgoing call scenario from phone {} times from known contact".
+ format(iteration_count))
+ input("Press enter to execute this testcase...")
+ outgoing_num = get_phone_number(self.log, sec_dut)
+ for _ in range(iteration_count):
+ if not initiate_call(self.log, pri_dut, outgoing_num):
+ self.log.error("Failed to initiate call")
+ return False
+ if not wait_and_answer_call(self.log, sec_dut):
+ self.log.error("Failed to answer call.")
+ return False
+ time.sleep(self.short_timeout)
+ if not hangup_call(self.log, pri_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return True
+
+ def outgoing_call_hsp_disabled_then_enabled_during_call(
+ self, pri_dut, sec_dut):
+ self.log.info(
+ "Test outgoing call hsp disabled then enable during call.")
+ input("Press enter to execute this testcase...")
+ outgoing_num = get_phone_number(self.log, sec_dut)
+ if not pri_dut.droid.bluetoothHspDisconnect(self.target_mac_address):
+ self.log.error("Failed to disconnect HSP service...")
+ return False
+ time.sleep(self.short_timeout)
+ if len(pri_dut.droid.bluetoothHspGetConnectedDevices()) != 0:
+ self.log.error("Failed to disconnect from HSP service")
+ return False
+ if not initiate_call(self.log, pri_dut, outgoing_num):
+ self.log.error("Failed to initiate call")
+ return False
+ time.sleep(default_timeout)
+ pri_dut.droid.bluetoothConnectBonded(self.target_mac_address)
+ time.sleep(self.short_timeout)
+ test_result = True
+ if len(pri_dut.droid.bluetoothHspGetConnectedDevices()) != 1:
+ self.log.error("Failed to reconnect to HSP service...")
+ return
+ if not hangup_call(self.log, pri_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return test_result
+
+ def call_audio_routes(self, pri_dut, sec_dut):
+ self.log.info("Test various audio routes scenario from phone.")
+ input("Press enter to execute this testcase...")
+ outgoing_num = get_phone_number(self.log, sec_dut)
+ if not initiate_call(self.log, pri_dut, outgoing_num):
+ self.log.error("Failed to initiate call")
+ return False
+ if not wait_and_answer_call(self.log, sec_dut):
+ self.log.error("Failed to answer call.")
+ return False
+ time.sleep(self.short_timeout)
+ call_id = pri_dut.droid.telecomCallGetCallIds()[0]
+ pri_dut.droid.telecomCallPlayDtmfTone(call_id, "9")
+ input("Press enter to switch to speaker...")
+ self.log.info("Switching to speaker.")
+ set_audio_route(self.log, pri_dut, AUDIO_ROUTE_SPEAKER)
+ time.sleep(self.short_timeout)
+ if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_SPEAKER:
+ self.log.error(
+ "Audio Route not set to {}".format(AUDIO_ROUTE_SPEAKER))
+ return False
+ input("Press enter to switch to earpiece...")
+ self.log.info("Switching to earpiece.")
+ set_audio_route(self.log, pri_dut, AUDIO_ROUTE_EARPIECE)
+ time.sleep(self.short_timeout)
+ if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_EARPIECE:
+ self.log.error(
+ "Audio Route not set to {}".format(AUDIO_ROUTE_EARPIECE))
+ return False
+ input("Press enter to switch to Bluetooth...")
+ self.log.info("Switching to Bluetooth...")
+ set_audio_route(self.log, pri_dut, AUDIO_ROUTE_BLUETOOTH)
+ time.sleep(self.short_timeout)
+ if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_BLUETOOTH:
+ self.log.error(
+ "Audio Route not set to {}".format(AUDIO_ROUTE_BLUETOOTH))
+ return False
+ input("Press enter to hangup call...")
+ self.log.info("Hanging up call...")
+ pri_dut.droid.telecomCallStopDtmfTone(call_id)
+ if not hangup_call(self.log, pri_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return True
+
+ def sms_receive_different_sizes(self, pri_dut, sec_dut):
+ self.log.info("Test recieve sms.")
+ input("Press enter to execute this testcase...")
+ msg = [rand_ascii_str(50), rand_ascii_str(1), rand_ascii_str(500)]
+ if not sms_send_receive_verify(self.log, sec_dut, pri_dut, msg):
+ return False
+ else:
+ self.log.info("Successfully sent sms. Please verify on carkit.")
+ return True
+
+ def sms_receive_multiple(self, pri_dut, sec_dut):
+ text_count = 10
+ self.log.info(
+ "Test sending {} sms messages to phone.".format(text_count))
+ input("Press enter to execute this testcase...")
+ for _ in range(text_count):
+ msg = [rand_ascii_str(50)]
+ if not sms_send_receive_verify(self.log, sec_dut, pri_dut, msg):
+ return False
+ else:
+ self.log.info(
+ "Successfully sent sms. Please verify on carkit.")
+ return True
+
+ def sms_send_outgoing_texts(self, pri_dut, sec_dut):
+ self.log.info("Test send sms of different sizes.")
+ input("Press enter to execute this testcase...")
+ msg = [rand_ascii_str(50), rand_ascii_str(1), rand_ascii_str(500)]
+ if not sms_send_receive_verify(self.log, pri_dut, sec_dut, msg):
+ return False
+ else:
+ self.log.info("Successfully sent sms. Please verify on carkit.")
+ return True
+
+ def sms_during_incomming_call(self, pri_dut, sec_dut):
+ self.log.info(
+ "Test incomming call scenario to phone from unknown contact")
+ input("Press enter to execute this testcase...")
+ outgoing_num = get_phone_number(self.log, pri_dut)
+ if not initiate_call(self.log, sec_dut, outgoing_num):
+ self.log.error("Failed to initiate call")
+ return False
+ if not wait_and_answer_call(self.log, pri_dut):
+ self.log.error("Failed to answer call.")
+ return False
+ time.sleep(self.short_timeout)
+ msg = [rand_ascii_str(10)]
+ if not sms_send_receive_verify(self.log, sec_dut, pri_dut, msg):
+ return False
+ else:
+ self.log.info("Successfully sent sms. Please verify on carkit.")
+ input("Press enter to hangup call...")
+ if not hangup_call(self.log, sec_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return True
+
+ def multi_incomming_call(self, pri_dut, sec_dut, ter_dut):
+ self.log.info("Test 2 incomming calls scenario to phone.")
+ input("Press enter to execute this testcase...")
+ if not self.setup_multi_call(sec_dut, ter_dut, pri_dut):
+ return False
+ input("Press enter to hangup call 1...")
+ if not hangup_call(self.log, sec_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ input("Press enter to hangup call 2...")
+ if not hangup_call(self.log, ter_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return True
+
+ def multi_call_audio_routing(self, pri_dut, sec_dut, ter_dut):
+ self.log.info(
+ "Test 2 incomming calls scenario to phone, then test audio routing."
+ )
+ input("Press enter to execute this testcase...")
+ if not self.setup_multi_call(sec_dut, ter_dut, pri_dut):
+ return False
+ input("Press enter to switch to earpiece...")
+ self.log.info("Switching to earpiece.")
+ set_audio_route(self.log, pri_dut, AUDIO_ROUTE_EARPIECE)
+ time.sleep(self.short_timeout)
+ if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_EARPIECE:
+ self.log.error(
+ "Audio Route not set to {}".format(AUDIO_ROUTE_EARPIECE))
+ return False
+ input("Press enter to switch to Bluetooth...")
+ self.log.info("Switching to Bluetooth...")
+ set_audio_route(self.log, pri_dut, AUDIO_ROUTE_BLUETOOTH)
+ time.sleep(self.short_timeout)
+ if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_BLUETOOTH:
+ self.log.error(
+ "Audio Route not set to {}".format(AUDIO_ROUTE_BLUETOOTH))
+ return False
+ input("Press enter to hangup call 1...")
+ if not hangup_call(self.log, sec_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ input("Press enter to hangup call 2...")
+ if not hangup_call(self.log, ter_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return True
+
+ def multi_call_swap_multiple_times(self, pri_dut, sec_dut, ter_dut):
+ self.log.info(
+ "Test 2 incomming calls scenario to phone, then test audio routing."
+ )
+ input("Press enter to execute this testcase...")
+ if not self.setup_multi_call(sec_dut, ter_dut, pri_dut):
+ return False
+ input("Press enter to swap active calls...")
+ calls = pri_dut.droid.telecomCallGetCallIds()
+ if not swap_calls(self.log, [pri_dut, sec_dut, ter_dut], calls[0],
+ calls[1], 5):
+ return False
+ input("Press enter to hangup call 1...")
+ if not hangup_call(self.log, sec_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ input("Press enter to hangup call 2...")
+ if not hangup_call(self.log, ter_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return True
+
+ def multi_call_join_conference_call(self, pri_dut, sec_dut, ter_dut):
+ self.log.info(
+ "Test 2 incomming calls scenario to phone then join the calls.")
+ input("Press enter to execute this testcase...")
+ if not self.setup_multi_call(sec_dut, ter_dut, pri_dut):
+ return False
+ input("Press enter to join active calls...")
+ calls = pri_dut.droid.telecomCallGetCallIds()
+ pri_dut.droid.telecomCallJoinCallsInConf(calls[0], calls[1])
+ time.sleep(WAIT_TIME_IN_CALL)
+ if num_active_calls(self.log, pri_dut) != 4:
+ self.log.error("Total number of call ids in {} is not 4.".format(
+ pri_dut.serial))
+ return False
+ input("Press enter to hangup call 1...")
+ if not hangup_call(self.log, sec_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ input("Press enter to hangup call 2...")
+ if not hangup_call(self.log, ter_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return True
+
+ def multi_call_join_conference_call_hangup_conf_call(
+ self, pri_dut, sec_dut, ter_dut):
+ self.log.info(
+ "Test 2 incomming calls scenario to phone then join the calls, then terminate the call from the primary dut."
+ )
+ input("Press enter to execute this testcase...")
+ if not self.setup_multi_call(sec_dut, ter_dut, pri_dut):
+ return False
+ input("Press enter to join active calls...")
+ calls = pri_dut.droid.telecomCallGetCallIds()
+ pri_dut.droid.telecomCallJoinCallsInConf(calls[0], calls[1])
+ time.sleep(WAIT_TIME_IN_CALL)
+ if num_active_calls(self.log, pri_dut) != 4:
+ self.log.error("Total number of call ids in {} is not 4.".format(
+ pri_dut.serial))
+ return False
+ input("Press enter to hangup conf call...")
+ if not hangup_call(self.log, pri_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return True
+
+ def outgoing_multi_call_join_conference_call(self, pri_dut, sec_dut,
+ ter_dut):
+ self.log.info(
+ "Test 2 outgoing calls scenario from phone then join the calls.")
+ input("Press enter to execute this testcase...")
+ outgoing_num = get_phone_number(self.log, sec_dut)
+ if not initiate_call(self.log, pri_dut, outgoing_num):
+ self.log.error("Failed to initiate call")
+ return False
+ if not wait_and_answer_call(self.log, sec_dut):
+ self.log.error("Failed to answer call.")
+ return False
+ time.sleep(self.short_timeout)
+ outgoing_num = get_phone_number(self.log, ter_dut)
+ if not initiate_call(self.log, pri_dut, outgoing_num):
+ self.log.error("Failed to initiate call")
+ return False
+ if not wait_and_answer_call(self.log, ter_dut):
+ self.log.error("Failed to answer call.")
+ return False
+ input("Press enter to join active calls...")
+ calls = pri_dut.droid.telecomCallGetCallIds()
+ pri_dut.droid.telecomCallJoinCallsInConf(calls[0], calls[1])
+ time.sleep(WAIT_TIME_IN_CALL)
+ if num_active_calls(self.log, pri_dut) != 4:
+ self.log.error("Total number of call ids in {} is not 4.".format(
+ pri_dut.serial))
+ return False
+ input("Press enter to hangup call 1...")
+ if not hangup_call(self.log, sec_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ input("Press enter to hangup call 2...")
+ if not hangup_call(self.log, ter_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return True
+
+ def multi_call_join_conference_call_audio_routes(self, pri_dut, sec_dut,
+ ter_dut):
+ self.log.info(
+ "Test 2 incomming calls scenario to phone then join the calls, then test different audio routes."
+ )
+ input("Press enter to execute this testcase...")
+ if not self.setup_multi_call(sec_dut, ter_dut, pri_dut):
+ return False
+ input("Press enter to join active calls...")
+ calls = pri_dut.droid.telecomCallGetCallIds()
+ pri_dut.droid.telecomCallJoinCallsInConf(calls[0], calls[1])
+ time.sleep(WAIT_TIME_IN_CALL)
+ if num_active_calls(self.log, pri_dut) != 4:
+ self.log.error("Total number of call ids in {} is not 4.".format(
+ pri_dut.serial))
+ return False
+ input("Press enter to switch to phone speaker...")
+ self.log.info("Switching to earpiece.")
+ set_audio_route(self.log, pri_dut, AUDIO_ROUTE_EARPIECE)
+ time.sleep(self.short_timeout)
+ if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_EARPIECE:
+ self.log.error(
+ "Audio Route not set to {}".format(AUDIO_ROUTE_EARPIECE))
+ return False
+ input("Press enter to switch to Bluetooth...")
+ self.log.info("Switching to Bluetooth...")
+ set_audio_route(self.log, pri_dut, AUDIO_ROUTE_BLUETOOTH)
+ time.sleep(self.short_timeout)
+ if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_BLUETOOTH:
+ self.log.error(
+ "Audio Route not set to {}".format(AUDIO_ROUTE_BLUETOOTH))
+ return False
+ input("Press enter to hangup conf call...")
+ if not hangup_call(self.log, pri_dut):
+ self.log.error("Failed to hangup call")
+ return False
+ return True
+
+ def avrcp_play_pause(self, pri_dut):
+ play_pause_count = 5
+ self.log.info(
+ "Test AVRCP play/pause {} times.".format(play_pause_count))
+ pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
+ input(
+ "Press enter to execute this testcase when music is in the play state..."
+ )
+ for i in range(play_pause_count):
+ input("Execute iteration {}?".format(i + 1))
+ pri_dut.adb.shell(KEYCODE_EVENT_PLAY_PAUSE)
+ self.log.info("Test should end in a paused state.")
+ return True
+
+ def avrcp_next_previous_song(self, pri_dut):
+ self.log.info("Test AVRCP go to the next song then the previous song.")
+ pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
+ input(
+ "Press enter to execute this testcase when music is in the play state..."
+ )
+ self.log.info("Hitting Next input event...")
+ pri_dut.adb.shell(KEYCODE_EVENT_NEXT)
+ input("Press enter to go to the previous song")
+ pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
+ pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
+ self.log.info("Test should end on original song.")
+ return True
+
+ def avrcp_next_previous(self, pri_dut):
+ self.log.info(
+ "Test AVRCP go to the next song then the press previous after a few seconds."
+ )
+ pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
+ input(
+ "Press enter to execute this testcase when music is in the play state..."
+ )
+ self.log.info("Hitting Next input event...")
+ pri_dut.adb.shell(KEYCODE_EVENT_NEXT)
+ time.sleep(5)
+ self.log.info("Hitting Previous input event...")
+ pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
+ self.log.info("Test should end on \"next\" song.")
+ return True
+
+ def avrcp_next_repetative(self, pri_dut):
+ iterations = 10
+ self.log.info("Test AVRCP go to the next {} times".format(iterations))
+ pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
+ input(
+ "Press enter to execute this testcase when music is in the play state..."
+ )
+ for i in range(iterations):
+ self.log.info(
+ "Hitting Next input event, iteration {}...".format(i + 1))
+ pri_dut.adb.shell(KEYCODE_EVENT_NEXT)
+ # Allow time for the carkit to update.
+ time.sleep(1)
+ return True
+
+ def _cycle_aboslute_volume_control_helper(self, volume_step,
+ android_volume_steps, pri_dut):
+ begin_time = get_current_epoch_time()
+ pri_dut.droid.setMediaVolume(volume_step)
+ percentage_to_set = int((volume_step / android_volume_steps) * 100)
+ self.log.info("Setting phone volume to {}%".format(percentage_to_set))
+ volume_info_logcat = pri_dut.search_logcat(
+ logcat_strings['media_playback_vol_changed'], begin_time)
+ if len(volume_info_logcat) > 1:
+ self.log.info("Instant response detected.")
+ carkit_response = volume_info_logcat[-1]['log_message'].split(',')
+ for item in carkit_response:
+ if " volume=" in item:
+ carkit_vol_response = int((
+ int(item.split("=")[-1]) / android_volume_steps) * 100)
+ self.log.info(
+ "Carkit set volume to {}%".format(carkit_vol_response))
+ result = input(
+ "Did volume change reflect properly on carkit and phone? (Y/n) "
+ ).lower()
+
+ def cycle_absolute_volume_control(self, pri_dut):
+ result = input(
+ "Does carkit support Absolute Volume Control? (Y/n) ").lower()
+ if result is "n":
+ return True
+ android_volume_steps = 25
+ for i in range(android_volume_steps):
+ self._cycle_aboslute_volume_control_helper(i, android_volume_steps,
+ pri_dut)
+ for i in reversed(range(android_volume_steps)):
+ self._cycle_aboslute_volume_control_helper(i, android_volume_steps,
+ pri_dut)
+ return True
+
+ def cycle_battery_level(self, pri_dut):
+ for i in range(11):
+ level = i * 10
+ pri_dut.shell.set_battery_level(level)
+ question = "Phone battery level {}. Has the carkit indicator " \
+ "changed? (Y/n) "
+ result = input(question.format(level)).lower()
+
+ def test_voice_recognition_from_phone(self, pri_dut):
+ result = input(
+ "Does carkit support voice recognition (BVRA)? (Y/n) ").lower()
+ if result is "n":
+ return True
+ input("Press enter to start voice recognition from phone.")
+ self.pri_dut.droid.bluetoothHspStartVoiceRecognition(
+ self.target_mac_address)
+ input("Press enter to stop voice recognition from phone.")
+ self.pri_dut.droid.bluetoothHspStopVoiceRecognition(
+ self.target_mac_address)
+
+ def test_audio_and_voice_recognition_from_phone(self, pri_dut):
+ result = input(
+ "Does carkit support voice recognition (BVRA)? (Y/n) ").lower()
+ if result is "n":
+ return True
+ # Start playing music here
+ input("Press enter to start voice recognition from phone.")
+ self.pri_dut.droid.bluetoothHspStartVoiceRecognition(
+ self.target_mac_address)
+ input("Press enter to stop voice recognition from phone.")
+ self.pri_dut.droid.bluetoothHspStopVoiceRecognition(
+ self.target_mac_address)
+ time.sleep(2)
+ result = input("Did carkit continue music playback after? (Y/n) ")
diff --git a/acts/framework/acts/test_utils/bt/bt_coc_test_utils.py b/acts/framework/acts/test_utils/bt/bt_coc_test_utils.py
index f7d4867..306291f 100644
--- a/acts/framework/acts/test_utils/bt/bt_coc_test_utils.py
+++ b/acts/framework/acts/test_utils/bt/bt_coc_test_utils.py
@@ -162,7 +162,7 @@
client_ad.log.info(
"Orchestrate_coc_connection: Start BLE advertisement and"
"scanning. Secured Connection={}".format(secured_conn))
- mac_address, adv_callback = (
+ mac_address, adv_callback, scan_callback = (
get_mac_address_of_generic_advertisement(client_ad, server_ad))
except BtTestUtilsError as err:
raise BtCoCTestUtilsError(
@@ -187,6 +187,7 @@
False,
transport=gatt_transport['le'],
opportunistic=False)
+ client_ad.droid.bleStopBleScan(scan_callback)
except GattTestUtilsError as err:
client_ad.log.error(err)
if (adv_callback != None):
diff --git a/acts/framework/acts/test_utils/bt/bt_constants.py b/acts/framework/acts/test_utils/bt/bt_constants.py
index 20f68dd..f093a1f 100644
--- a/acts/framework/acts/test_utils/bt/bt_constants.py
+++ b/acts/framework/acts/test_utils/bt/bt_constants.py
@@ -589,3 +589,10 @@
headphone_bus_endpoint = 'Cros device headphone'
### Chameleon Constants End ###
+
+### Begin logcat strings dict"""
+logcat_strings = {
+ "media_playback_vol_changed": "onRouteVolumeChanged",
+}
+
+### End logcat strings dict"""
diff --git a/acts/framework/acts/test_utils/bt/bt_gatt_utils.py b/acts/framework/acts/test_utils/bt/bt_gatt_utils.py
index f3fb6bc..e3d410e 100644
--- a/acts/framework/acts/test_utils/bt/bt_gatt_utils.py
+++ b/acts/framework/acts/test_utils/bt/bt_gatt_utils.py
@@ -106,7 +106,7 @@
if mac_address is None:
if transport == gatt_transport['le']:
try:
- mac_address, adv_callback = (
+ mac_address, adv_callback, scan_callback = (
get_mac_address_of_generic_advertisement(cen_ad, per_ad))
except BtTestUtilsError as err:
raise GattTestUtilsError(
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 5ddc9a3..9a930ab 100644
--- a/acts/framework/acts/test_utils/bt/bt_test_utils.py
+++ b/acts/framework/acts/test_utils/bt/bt_test_utils.py
@@ -66,6 +66,7 @@
from acts.test_utils.bt.bt_constants import scan_result
from acts.test_utils.bt.bt_constants import scan_failed
from acts.test_utils.bt.bt_constants import hid_id_keyboard
+
from acts.test_utils.tel.tel_test_utils import toggle_airplane_mode_by_adb
from acts.test_utils.tel.tel_test_utils import verify_http_connection
from acts.utils import exe_cmd
@@ -573,8 +574,7 @@
raise BtTestUtilsError(
"Scanner did not find advertisement {}".format(err))
mac_address = event['data']['Result']['deviceInfo']['address']
- scan_ad.droid.bleStopBleScan(scan_callback)
- return mac_address, advertise_callback
+ return mac_address, advertise_callback, scan_callback
def enable_bluetooth(droid, ed):
if droid.bluetoothCheckState() is True:
diff --git a/acts/framework/acts/test_utils/bt/bta_lib.py b/acts/framework/acts/test_utils/bt/bta_lib.py
index 723dfdb..661a7e8 100644
--- a/acts/framework/acts/test_utils/bt/bta_lib.py
+++ b/acts/framework/acts/test_utils/bt/bta_lib.py
@@ -24,11 +24,14 @@
class BtaLib():
- def __init__(self, log, mac_addr, dut):
+ def __init__(self, log, dut, target_mac_address=None):
self.advertisement_list = []
self.dut = dut
self.log = log
- self.mac_addr = mac_addr
+ self.target_mac_addr = target_mac_address
+
+ def set_target_mac_addr(self, mac_addr):
+ self.target_mac_addr = mac_addr
def set_scan_mode(self, scan_mode):
"""Set the Scan mode of the Bluetooth Adapter"""
@@ -48,7 +51,7 @@
def init_bond(self):
"""Initiate bond to PTS device"""
- self.dut.droid.bluetoothDiscoverAndBond(self.mac_addr)
+ self.dut.droid.bluetoothDiscoverAndBond(self.target_mac_addr)
def start_discovery(self):
"""Start BR/EDR Discovery"""
@@ -67,15 +70,15 @@
def bond(self):
"""Bond to PTS device"""
- self.dut.droid.bluetoothBond(self.mac_addr)
+ self.dut.droid.bluetoothBond(self.target_mac_addr)
def disconnect(self):
"""BTA disconnect"""
- self.dut.droid.bluetoothDisconnectConnected(self.mac_addr)
+ self.dut.droid.bluetoothDisconnectConnected(self.target_mac_addr)
def unbond(self):
"""Unbond from PTS device"""
- self.dut.droid.bluetoothUnbond(self.mac_addr)
+ self.dut.droid.bluetoothUnbond(self.target_mac_addr)
def start_pairing_helper(self, line):
"""Start or stop Bluetooth Pairing Helper"""
@@ -96,12 +99,13 @@
def fetch_uuids_with_sdp(self):
"""BTA fetch UUIDS with SDP"""
- self.log.info(self.dut.droid.bluetoothFetchUuidsWithSdp(self.mac_addr))
+ self.log.info(
+ self.dut.droid.bluetoothFetchUuidsWithSdp(self.target_mac_addr))
def connect_profiles(self):
"""Connect available profiles"""
- self.dut.droid.bluetoothConnectBonded(self.mac_addr)
+ self.dut.droid.bluetoothConnectBonded(self.target_mac_addr)
def tts_speak(self):
"""Open audio channel by speaking characters"""
- self.dut.droid.ttsSpeak(self.mac_addr)
+ self.dut.droid.ttsSpeak(self.target_mac_addr)
diff --git a/acts/framework/acts/test_utils/bt/gattc_lib.py b/acts/framework/acts/test_utils/bt/gattc_lib.py
index 4dfc076..349c9ee 100644
--- a/acts/framework/acts/test_utils/bt/gattc_lib.py
+++ b/acts/framework/acts/test_utils/bt/gattc_lib.py
@@ -17,13 +17,19 @@
GATT Client Libraries
"""
+from acts.test_utils.bt.bt_constants import default_le_connection_interval_ms
+from acts.test_utils.bt.bt_constants import default_bluetooth_socket_timeout_ms
from acts.test_utils.bt.bt_gatt_utils import disconnect_gatt_connection
from acts.test_utils.bt.bt_gatt_utils import setup_gatt_connection
from acts.test_utils.bt.bt_gatt_utils import setup_gatt_mtu
+from acts.test_utils.bt.bt_constants import ble_scan_settings_modes
from acts.test_utils.bt.bt_constants import gatt_cb_strings
from acts.test_utils.bt.bt_constants import gatt_char_desc_uuids
from acts.test_utils.bt.bt_constants import gatt_descriptor
from acts.test_utils.bt.bt_constants import gatt_transport
+from acts.test_utils.bt.bt_constants import le_default_supervision_timeout
+from acts.test_utils.bt.bt_constants import le_connection_interval_time_step_ms
+from acts.test_utils.bt.bt_constants import scan_result
from acts.test_utils.bt.bt_gatt_utils import log_gatt_server_uuids
import time
@@ -31,20 +37,47 @@
class GattClientLib():
- def __init__(self, log, mac_addr, dut):
+ def __init__(self, log, dut, target_mac_addr=None):
self.dut = dut
self.log = log
- self.mac_addr = mac_addr
self.gatt_callback = None
self.bluetooth_gatt = None
self.discovered_services_index = None
+ self.target_mac_addr = target_mac_addr
self.generic_uuid = "0000{}-0000-1000-8000-00805f9b34fb"
+ def set_target_mac_addr(self, mac_addr):
+ self.target_mac_addr = mac_addr
+
+ def connect_over_le_based_off_name(self, autoconnect, name):
+ """Perform GATT connection over LE"""
+ self.dut.droid.bleSetScanSettingsScanMode(
+ ble_scan_settings_modes['low_latency'])
+ filter_list = self.dut.droid.bleGenFilterList()
+ scan_settings = self.dut.droid.bleBuildScanSetting()
+ scan_callback = self.dut.droid.bleGenScanCallback()
+ event_name = scan_result.format(scan_callback)
+ self.dut.droid.bleSetScanFilterDeviceName("BLE Rect")
+ self.dut.droid.bleBuildScanFilter(filter_list)
+ self.dut.droid.bleStartBleScan(filter_list, scan_settings,
+ scan_callback)
+
+ try:
+ event = self.dut.ed.pop_event(event_name, 10)
+ self.log.info("Found scan result: {}".format(event))
+ except Exception:
+ self.log.info("Didn't find any scan results.")
+ mac_addr = event['data']['Result']['deviceInfo']['address']
+ self.bluetooth_gatt, self.gatt_callback = setup_gatt_connection(
+ self.dut, mac_addr, autoconnect, transport=gatt_transport['le'])
+ self.dut.droid.bleStopBleScan(scan_callback)
+ self.discovered_services_index = None
+
def connect_over_le(self, autoconnect):
"""Perform GATT connection over LE"""
self.bluetooth_gatt, self.gatt_callback = setup_gatt_connection(
self.dut,
- self.mac_addr,
+ self.target_mac_addr,
autoconnect,
transport=gatt_transport['le'])
self.discovered_services_index = None
@@ -52,10 +85,14 @@
def connect_over_bredr(self):
"""Perform GATT connection over BREDR"""
self.bluetooth_gatt, self.gatt_callback = setup_gatt_connection(
- self.dut, self.mac_addr, False, transport=gatt_transport['bredr'])
+ self.dut,
+ self.target_mac_addr,
+ False,
+ transport=gatt_transport['bredr'])
def disconnect(self):
"""Perform GATT disconnect"""
+ cmd = "Disconnect GATT connection"
try:
disconnect_gatt_connection(self.dut, self.bluetooth_gatt,
self.gatt_callback)
@@ -296,15 +333,6 @@
self._setup_discovered_services_index()
services_count = self.dut.droid.gattClientGetDiscoveredServicesCount(
self.discovered_services_index)
- """
- self.log.info(
- CMD_LOG.format(
- cmd,
- self.dut.droid.gattClientWriteDescriptorByInstanceId(
- self.bluetooth_gatt, self.discovered_services_index,
- int(instance_id, 16),
- gatt_descriptor['enable_notification_value'])))
- """
for i in range(services_count):
characteristic_uuids = (
self.dut.droid.gattClientGetDiscoveredCharacteristicUuids(
@@ -332,6 +360,39 @@
self.bluetooth_gatt,
self.discovered_services_index, i, j, True)
+ def enable_indication_desc_by_instance_id(self, line):
+ """GATT Client Enable indication on Descriptor by instance ID"""
+ instance_id = line
+ self._setup_discovered_services_index()
+ services_count = self.dut.droid.gattClientGetDiscoveredServicesCount(
+ self.discovered_services_index)
+ for i in range(services_count):
+ characteristic_uuids = (
+ self.dut.droid.gattClientGetDiscoveredCharacteristicUuids(
+ self.discovered_services_index, i))
+ for j in range(len(characteristic_uuids)):
+ descriptor_uuids = (
+ self.dut.droid.
+ gattClientGetDiscoveredDescriptorUuidsByIndex(
+ self.discovered_services_index, i, j))
+ for k in range(len(descriptor_uuids)):
+ desc_inst_id = self.dut.droid.gattClientGetDescriptorInstanceId(
+ self.bluetooth_gatt, self.discovered_services_index, i,
+ j, k)
+ if desc_inst_id == int(instance_id, 16):
+ self.dut.droid.gattClientDescriptorSetValueByIndex(
+ self.bluetooth_gatt,
+ self.discovered_services_index, i, j, k,
+ gatt_descriptor['enable_indication_value'])
+ time.sleep(2) #Necessary for PTS
+ self.dut.droid.gattClientWriteDescriptorByIndex(
+ self.bluetooth_gatt,
+ self.discovered_services_index, i, j, k)
+ time.sleep(2) #Necessary for PTS
+ self.dut.droid.gattClientSetCharacteristicNotificationByIndex(
+ self.bluetooth_gatt,
+ self.discovered_services_index, i, j, True)
+
def char_enable_all_notifications(self):
self._setup_discovered_services_index()
services_count = self.dut.droid.gattClientGetDiscoveredServicesCount(
@@ -486,3 +547,30 @@
uuid = self.generic_uuid.format(line)
self.dut.droid.gattClientDiscoverServiceByUuid(self.bluetooth_gatt,
uuid)
+
+ def request_le_connection_parameters(self):
+ le_min_ce_len = 0
+ le_max_ce_len = 0
+ le_connection_interval = 0
+ minInterval = default_le_connection_interval_ms / le_connection_interval_time_step_ms
+ maxInterval = default_le_connection_interval_ms / le_connection_interval_time_step_ms
+ return_status = self.dut.droid.gattClientRequestLeConnectionParameters(
+ self.bluetooth_gatt, minInterval, maxInterval, 0,
+ le_default_supervision_timeout, le_min_ce_len, le_max_ce_len)
+ self.log.info(
+ "Result of request le connection param: {}".format(return_status))
+
+ def socket_conn_begin_connect_thread_psm(self, line):
+ args = line.split()
+ is_ble = bool(int(args[0]))
+ secured_conn = bool(int(args[1]))
+ psm_value = int(args[2]) # 1
+ self.dut.droid.bluetoothSocketConnBeginConnectThreadPsm(
+ self.target_mac_addr, is_ble, psm_value, secured_conn)
+
+ def socket_conn_begin_accept_thread_psm(self, line):
+ accept_timeout_ms = default_bluetooth_socket_timeout_ms
+ is_ble = True
+ secured_conn = False
+ self.dut.droid.bluetoothSocketConnBeginAcceptThreadPsm(
+ accept_timeout_ms, is_ble, secured_conn)
diff --git a/acts/framework/acts/test_utils/bt/gatts_lib.py b/acts/framework/acts/test_utils/bt/gatts_lib.py
index 7916a92..cbffb18 100644
--- a/acts/framework/acts/test_utils/bt/gatts_lib.py
+++ b/acts/framework/acts/test_utils/bt/gatts_lib.py
@@ -117,12 +117,15 @@
self.gatt_server_callback)
char_read = gatt_event['char_read_req']['evt'].format(
self.gatt_server_callback)
+ char_write_req = gatt_event['char_write_req']['evt'].format(
+ self.gatt_server_callback)
char_write = gatt_event['char_write']['evt'].format(
self.gatt_server_callback)
execute_write = gatt_event['exec_write']['evt'].format(
self.gatt_server_callback)
- regex = "({}|{}|{}|{}|{})".format(desc_read, desc_write, char_read,
- char_write, execute_write)
+ regex = "({}|{}|{}|{}|{}|{})".format(desc_read, desc_write, char_read,
+ char_write, execute_write,
+ char_write_req)
events = self.dut.ed.pop_events(regex, 5, small_timeout)
status = 0
if user_input:
@@ -147,7 +150,8 @@
continue
offset = event['data']['offset']
instance_id = event['data']['instanceId']
- if (event['name'] == desc_write or event['name'] == char_write):
+ if (event['name'] == desc_write or event['name'] == char_write
+ or event['name'] == char_write_req):
if ('preparedWrite' in event['data']
and event['data']['preparedWrite'] == True):
value = event['data']['value']
diff --git a/acts/framework/acts/test_utils/bt/rfcomm_lib.py b/acts/framework/acts/test_utils/bt/rfcomm_lib.py
index 49b35b6..ee1fd04 100644
--- a/acts/framework/acts/test_utils/bt/rfcomm_lib.py
+++ b/acts/framework/acts/test_utils/bt/rfcomm_lib.py
@@ -24,11 +24,14 @@
class RfcommLib():
- def __init__(self, log, mac_addr, dut):
+ def __init__(self, log, dut, target_mac_addr=None):
self.advertisement_list = []
self.dut = dut
self.log = log
- self.mac_addr = mac_addr
+ self.target_mac_addr = target_mac_addr
+
+ def set_target_mac_addr(self, mac_addr):
+ self.target_mac_addr = mac_addr
def connect(self, line):
"""Perform an RFCOMM connect"""
@@ -37,17 +40,18 @@
uuid = line
if uuid:
self.dut.droid.bluetoothRfcommBeginConnectThread(
- self.mac_addr, uuid)
+ self.target_mac_addr, uuid)
else:
- self.dut.droid.bluetoothRfcommBeginConnectThread(self.mac_addr)
+ self.dut.droid.bluetoothRfcommBeginConnectThread(
+ self.target_mac_addr)
def open_rfcomm_socket(self):
"""Open rfcomm socket"""
- self.dut.droid.rfcommCreateRfcommSocket(self.mac_addr, 1)
+ self.dut.droid.rfcommCreateRfcommSocket(self.target_mac_addr, 1)
def open_l2cap_socket(self):
"""Open L2CAP socket"""
- self.dut.droid.rfcommCreateL2capSocket(self.mac_addr, 1)
+ self.dut.droid.rfcommCreateL2capSocket(self.target_mac_addr, 1)
def write(self, line):
"""Write String data over an RFCOMM connection"""
@@ -78,4 +82,4 @@
def open_l2cap_socket(self):
"""Open L2CAP socket"""
- self.dut.droid.rfcommCreateL2capSocket(self.mac_addr, 1)
+ self.dut.droid.rfcommCreateL2capSocket(self.target_mac_addr, 1)
diff --git a/acts/framework/acts/test_utils/bt/shell_commands_lib.py b/acts/framework/acts/test_utils/bt/shell_commands_lib.py
new file mode 100644
index 0000000..f44e123
--- /dev/null
+++ b/acts/framework/acts/test_utils/bt/shell_commands_lib.py
@@ -0,0 +1,44 @@
+#/usr/bin/env python3.4
+#
+# Copyright (C) 2018 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.
+"""
+Shell command library.
+"""
+
+
+class ShellCommands():
+ def __init__(self, log, dut):
+ self.dut = dut
+ self.log = log
+
+ def set_battery_level(self, level):
+ """Set the battery level via ADB shell
+ Args:
+ level: the percent level to set
+ """
+ self.dut.adb.shell("dumpsys battery set level {}".format(level))
+
+ def disable_ble_scanning(self):
+ """Disable BLE scanning via ADB shell"""
+ self.dut.adb.shell("settings put global ble_scan_always_enabled 0")
+
+ def enable_ble_scanning(self):
+ """Enable BLE scanning via ADB shell"""
+ self.dut.adb.shell("settings put global ble_scan_always_enabled 1")
+
+ def consume_cpu_core(self):
+ """Consume a CPU core on the Android device via ADB shell"""
+ self.dut.adb.shell("echo $$ > /dev/cpuset/top-app/tasks")
+ self.dut.adb.shell("cat /dev/urandom > /dev/null &")
diff --git a/acts/tests/google/ble/gatt/GattConnectTest.py b/acts/tests/google/ble/gatt/GattConnectTest.py
index 07c8e1b..0b160a2 100644
--- a/acts/tests/google/ble/gatt/GattConnectTest.py
+++ b/acts/tests/google/ble/gatt/GattConnectTest.py
@@ -103,8 +103,8 @@
event = self.per_ad.ed.pop_event(expected_event,
self.default_timeout)
except Empty:
- self.log.error(gatt_cb_strings['serv_added_err'].format(
- expected_event))
+ self.log.error(
+ gatt_cb_strings['serv_added_err'].format(expected_event))
return False
if event['data']['serviceUuid'].lower() != uuid.lower():
self.log.error("Uuid mismatch. Found: {}, Expected {}.".format(
@@ -124,8 +124,8 @@
mtu_size_found, expected_mtu))
return False
except Empty:
- self.log.error(gatt_cb_err['mtu_changed_err'].format(
- expected_event))
+ self.log.error(
+ gatt_cb_err['mtu_changed_err'].format(expected_event))
return False
expected_event = gatt_cb_strings['mtu_serv_changed'].format(
@@ -139,8 +139,8 @@
mtu_size_found, expected_mtu))
return False
except Empty:
- self.log.error(gatt_cb_err['mtu_serv_changed_err'].format(
- expected_event))
+ self.log.error(
+ gatt_cb_err['mtu_serv_changed_err'].format(expected_event))
return False
return True
@@ -228,9 +228,9 @@
return False
self.per_ad.droid.bleStopBleAdvertising(adv_callback)
try:
- event = self.cen_ad.ed.pop_event(gatt_cb_strings[
- 'gatt_conn_change'].format(gatt_callback,
- self.default_timeout))
+ event = self.cen_ad.ed.pop_event(
+ gatt_cb_strings['gatt_conn_change'].format(
+ gatt_callback, self.default_timeout))
self.log.error(
"Connection event found when not expected: {}".format(event))
return False
@@ -277,11 +277,12 @@
gatt_server_cb)
self.gatt_server_list.append(gatt_server)
autoconnect = False
- mac_address, adv_callback = (
+ mac_address, adv_callback, scan_callback = (
get_mac_address_of_generic_advertisement(self.cen_ad, self.per_ad))
try:
bluetooth_gatt, gatt_callback = setup_gatt_connection(
self.cen_ad, mac_address, autoconnect)
+ self.cen_ad.droid.bleStopBleScan(scan_callback)
self.bluetooth_gatt_list.append(bluetooth_gatt)
except GattTestUtilsError as err:
self.log.error(err)
@@ -306,8 +307,8 @@
event = self.cen_ad.ed.pop_event(expected_event,
self.default_timeout)
except Empty:
- self.log.error(gatt_cb_err['gatt_conn_change_err'].format(
- expected_event))
+ self.log.error(
+ gatt_cb_err['gatt_conn_change_err'].format(expected_event))
test_result = False
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -345,7 +346,7 @@
gatt_server = self.per_ad.droid.gattServerOpenGattServer(
gatt_server_cb)
self.gatt_server_list.append(gatt_server)
- mac_address, adv_callback = (
+ mac_address, adv_callback, scan_callback = (
get_mac_address_of_generic_advertisement(self.cen_ad, self.per_ad))
# Make GATT connection 1
try:
@@ -355,6 +356,7 @@
False,
transport=gatt_transport['auto'],
opportunistic=False)
+ self.cen_ad.droid.bleStopBleScan(scan_callback)
self.bluetooth_gatt_list.append(bluetooth_gatt_1)
except GattTestUtilsError as err:
self.log.error(err)
@@ -585,8 +587,8 @@
try:
self.cen_ad.ed.pop_event(expected_event, self.default_timeout)
except Empty:
- self.log.error(gatt_cb_err['rd_remote_rssi_err'].format(
- expected_event))
+ self.log.error(
+ gatt_cb_err['rd_remote_rssi_err'].format(expected_event))
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -637,8 +639,8 @@
event = self.cen_ad.ed.pop_event(expected_event,
self.default_timeout)
except Empty:
- self.log.error(gatt_cb_err['gatt_serv_disc'].format(
- expected_event))
+ self.log.error(
+ gatt_cb_err['gatt_serv_disc'].format(expected_event))
return False
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -697,8 +699,8 @@
self.default_timeout)
discovered_services_index = event['data']['ServicesIndex']
except Empty:
- self.log.error(gatt_cb_err['gatt_serv_disc'].format(
- expected_event))
+ self.log.error(
+ gatt_cb_err['gatt_serv_disc'].format(expected_event))
return False
log_gatt_server_uuids(self.cen_ad, discovered_services_index)
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
@@ -754,8 +756,8 @@
event = self.cen_ad.ed.pop_event(expected_event,
self.default_timeout)
except Empty:
- self.log.error(gatt_cb_err['gatt_serv_disc'].format(
- expected_event))
+ self.log.error(
+ gatt_cb_err['gatt_serv_disc'].format(expected_event))
return False
discovered_services_index = event['data']['ServicesIndex']
log_gatt_server_uuids(self.cen_ad, discovered_services_index)
@@ -795,7 +797,7 @@
gatt_server = self.per_ad.droid.gattServerOpenGattServer(
gatt_server_cb)
self.gatt_server_list.append(gatt_server)
- mac_address, adv_callback = get_mac_address_of_generic_advertisement(
+ mac_address, adv_callback, scan_callback = get_mac_address_of_generic_advertisement(
self.cen_ad, self.per_ad)
autoconnect = False
for i in range(1000):
@@ -803,11 +805,12 @@
try:
bluetooth_gatt, gatt_callback = setup_gatt_connection(
self.cen_ad, mac_address, autoconnect)
+ self.cen_ad.droid.bleStopBleScan(scan_callback)
except GattTestUtilsError as err:
self.log.error(err)
return False
- test_result = self._orchestrate_gatt_disconnection(bluetooth_gatt,
- gatt_callback)
+ test_result = self._orchestrate_gatt_disconnection(
+ bluetooth_gatt, gatt_callback)
if not test_result:
self.log.info("Failed to disconnect from peripheral device.")
return False
@@ -867,8 +870,8 @@
gatt_characteristic['permission_write_encrypted_mitm'])
gatt_service = self.per_ad.droid.gattServerCreateService(
service_uuid, gatt_service_types['primary'])
- self.per_ad.droid.gattServerAddCharacteristicToService(gatt_service,
- characteristic)
+ 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_cb, service_uuid)
if not result:
@@ -884,8 +887,8 @@
event = self.cen_ad.ed.pop_event(expected_event,
self.default_timeout)
except Empty:
- self.log.error(gatt_cb_err['gatt_serv_disc'].format(
- expected_event))
+ self.log.error(
+ gatt_cb_err['gatt_serv_disc'].format(expected_event))
return False
discovered_services_index = event['data']['ServicesIndex']
else:
@@ -914,8 +917,8 @@
bonded_devices = \
self.cen_ad.droid.bluetoothGetBondedDevices()
for device in bonded_devices:
- if ('name' in device.keys() and
- device['name'] == target_name):
+ if ('name' in device.keys()
+ and device['name'] == target_name):
bonded = True
break
bonded = False
@@ -924,8 +927,8 @@
bonded_devices = \
self.per_ad.droid.bluetoothGetBondedDevices()
for device in bonded_devices:
- if ('name' in device.keys() and
- device['name'] == target_name):
+ if ('name' in device.keys()
+ and device['name'] == target_name):
bonded = True
break
for ad in [self.cen_ad, self.per_ad]:
@@ -935,8 +938,8 @@
time.sleep(2)
bonded_devices = ad.droid.bluetoothGetBondedDevices()
if len(bonded_devices) > 0:
- self.log.error("Failed to unbond devices: {}".format(
- bonded_devices))
+ self.log.error(
+ "Failed to unbond devices: {}".format(bonded_devices))
return False
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
gatt_callback)
@@ -1064,8 +1067,8 @@
self.cen_ad.droid.bleSetScanFilterDeviceName(
self.per_ad.droid.bluetoothGetLocalName())
self.cen_ad.droid.bleBuildScanFilter(filter_list)
- self.cen_ad.droid.bleSetScanSettingsScanMode(ble_scan_settings_modes[
- 'low_latency'])
+ self.cen_ad.droid.bleSetScanSettingsScanMode(
+ ble_scan_settings_modes['low_latency'])
# Setup necessary scan objects.
scan_settings = self.cen_ad.droid.bleBuildScanSetting()
@@ -1106,8 +1109,8 @@
# Steps 4: Try to connect to the first mac address
gatt_callback = self.cen_ad.droid.gattCreateGattCallback()
- self.log.info("Gatt Connect to mac address {}.".format(
- mac_address_pre_restart))
+ self.log.info(
+ "Gatt Connect to mac address {}.".format(mac_address_pre_restart))
bluetooth_gatt = self.cen_ad.droid.gattClientConnectGatt(
gatt_callback, mac_address_pre_restart, autoconnect, transport,
opportunistic, gatt_phy_mask['1m_mask'])
@@ -1127,8 +1130,8 @@
self.cen_ad.droid.gattClientDisconnect(bluetooth_gatt)
# Step 6: Connect to second mac address.
- self.log.info("Gatt Connect to mac address {}.".format(
- mac_address_post_restart))
+ self.log.info(
+ "Gatt Connect to mac address {}.".format(mac_address_post_restart))
bluetooth_gatt = self.cen_ad.droid.gattClientConnectGatt(
gatt_callback, mac_address_post_restart, autoconnect, transport,
opportunistic, gatt_phy_mask['1m_mask'])
@@ -1149,4 +1152,4 @@
except Empty:
self.log.error("No connection update was found.")
return False
- return self.cen_ad.droid.gattClientDisconnect(bluetooth_gatt)
\ No newline at end of file
+ return self.cen_ad.droid.gattClientDisconnect(bluetooth_gatt)
diff --git a/acts/tests/google/ble/gatt/GattToolTest.py b/acts/tests/google/ble/gatt/GattToolTest.py
index d9921b4..2b0b619 100644
--- a/acts/tests/google/ble/gatt/GattToolTest.py
+++ b/acts/tests/google/ble/gatt/GattToolTest.py
@@ -27,19 +27,19 @@
from queue import Empty
import time
-from acts.test_utils.bt.bt_constants import ble_scan_settings_modes
from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
+from acts.test_utils.bt.bt_constants import ble_scan_settings_modes
from acts.test_utils.bt.bt_constants import gatt_cb_err
from acts.test_utils.bt.bt_constants import gatt_cb_strings
from acts.test_utils.bt.bt_constants import gatt_descriptor
from acts.test_utils.bt.bt_constants import gatt_transport
+from acts.test_utils.bt.bt_constants import scan_result
from acts.test_utils.bt.bt_gatt_utils import GattTestUtilsError
from acts.test_utils.bt.bt_gatt_utils import disconnect_gatt_connection
from acts.test_utils.bt.bt_test_utils import generate_ble_scan_objects
from acts.test_utils.bt.bt_gatt_utils import setup_gatt_connection
from acts.test_utils.bt.bt_gatt_utils import log_gatt_server_uuids
from acts.test_utils.bt.bt_test_utils import reset_bluetooth
-from acts.test_utils.bt.bt_constants import scan_result
class GattToolTest(BluetoothBaseTest):
@@ -84,8 +84,8 @@
def _is_peripheral_advertising(self):
self.cen_ad.droid.bleSetScanFilterDeviceAddress(self.ble_mac_address)
- self.cen_ad.droid.bleSetScanSettingsScanMode(ble_scan_settings_modes[
- 'low_latency'])
+ self.cen_ad.droid.bleSetScanSettingsScanMode(
+ ble_scan_settings_modes['low_latency'])
filter_list, scan_settings, scan_callback = generate_ble_scan_objects(
self.cen_ad.droid)
self.cen_ad.droid.bleBuildScanFilter(filter_list)
@@ -96,8 +96,8 @@
test_result = True
try:
self.cen_ad.ed.pop_event(expected_event_name, self.DEFAULT_TIMEOUT)
- self.log.info("Peripheral found with event: {}".format(
- expected_event_name))
+ self.log.info(
+ "Peripheral found with event: {}".format(expected_event_name))
except Empty:
self.log.info("Peripheral not advertising or not found: {}".format(
self.ble_mac_address))
@@ -110,8 +110,8 @@
time.sleep(2) #Grace timeout for unbonding to finish
bonded_devices = self.cen_ad.droid.bluetoothGetBondedDevices()
if bonded_devices:
- self.log.error("Failed to unbond device... found: {}".format(
- bonded_devices))
+ self.log.error(
+ "Failed to unbond device... found: {}".format(bonded_devices))
return False
return True
@@ -145,9 +145,9 @@
self.AUTOCONNECT = False
start_time = self._get_time_in_milliseconds()
try:
- bluetooth_gatt, gatt_callback = (
- setup_gatt_connection(self.cen_ad, self.ble_mac_address,
- self.AUTOCONNECT, gatt_transport['le']))
+ bluetooth_gatt, gatt_callback = (setup_gatt_connection(
+ self.cen_ad, self.ble_mac_address, self.AUTOCONNECT,
+ gatt_transport['le']))
except GattTestUtilsError as err:
self.log.error(err)
return False
@@ -247,9 +247,9 @@
Priority: 2
"""
try:
- bluetooth_gatt, gatt_callback = (
- setup_gatt_connection(self.cen_ad, self.ble_mac_address,
- self.AUTOCONNECT, gatt_transport['le']))
+ bluetooth_gatt, gatt_callback = (setup_gatt_connection(
+ self.cen_ad, self.ble_mac_address, self.AUTOCONNECT,
+ gatt_transport['le']))
except GattTestUtilsError as err:
self.log.error(err)
return False
@@ -261,8 +261,8 @@
self.DEFAULT_TIMEOUT)
discovered_services_index = event['data']['ServicesIndex']
except Empty:
- self.log.error(gatt_cb_err['gatt_serv_disc'].format(
- expected_event))
+ self.log.error(
+ gatt_cb_err['gatt_serv_disc'].format(expected_event))
return False
log_gatt_server_uuids(self.cen_ad, discovered_services_index)
try:
@@ -382,9 +382,9 @@
if not self._pair_with_peripheral():
return False
try:
- bluetooth_gatt, gatt_callback = (
- setup_gatt_connection(self.cen_ad, self.ble_mac_address,
- self.AUTOCONNECT, gatt_transport['le']))
+ bluetooth_gatt, gatt_callback = (setup_gatt_connection(
+ self.cen_ad, self.ble_mac_address, self.AUTOCONNECT,
+ gatt_transport['le']))
except GattTestUtilsError as err:
self.log.error(err)
return False
@@ -396,8 +396,8 @@
self.DEFAULT_TIMEOUT)
discovered_services_index = event['data']['ServicesIndex']
except Empty:
- self.log.error(gatt_cb_err['gatt_serv_disc'].format(
- expected_event))
+ self.log.error(
+ gatt_cb_err['gatt_serv_disc'].format(expected_event))
return False
# TODO: in setup save service_cound and discovered_services_index
# programatically
@@ -431,7 +431,7 @@
# set 15 minute notification test time
notification_test_time = 900
end_time = time.time() + notification_test_time
- expected_event = GattCbStrings.CHAR_CHANGE.value.format(gatt_callback)
+ expected_event = gatt_cb_strings['char_change'].format(gatt_callback)
while time.time() < end_time:
try:
event = self.cen_ad.ed.pop_event(expected_event,
@@ -440,6 +440,6 @@
except Empty as err:
print(err)
self.log.error(
- GattCbStrings.CHAR_CHANGE_ERR.value.format(expected_event))
+ gatt_cb_err['char_change_err'].format(expected_event))
return False
return True
diff --git a/acts/tests/google/ble/system_tests/BleStressTest.py b/acts/tests/google/ble/system_tests/BleStressTest.py
index a1387ab..5be082a 100644
--- a/acts/tests/google/ble/system_tests/BleStressTest.py
+++ b/acts/tests/google/ble/system_tests/BleStressTest.py
@@ -168,7 +168,8 @@
self.adv_ad.droid.bleStartBleAdvertising(
advertise_callback, advertise_data, advertise_settings)
expected_advertise_event_name = "".join(
- ["BleAdvertise", str(advertise_callback), "onSuccess"])
+ ["BleAdvertise",
+ str(advertise_callback), "onSuccess"])
worker = self.adv_ad.ed.handle_event(
self.bleadvertise_verify_onsuccess_handler,
expected_advertise_event_name, ([]), self.default_timeout)
@@ -176,7 +177,8 @@
self.log.debug(worker.result(self.default_timeout))
except Empty as error:
self.log.debug(" ".join(
- ["Test failed with Empty error:", str(error)]))
+ ["Test failed with Empty error:",
+ str(error)]))
test_result = False
except concurrent.futures._base.TimeoutError as error:
self.log.debug(" ".join([
@@ -219,7 +221,8 @@
self.adv_ad.droid.bleStartBleAdvertising(
advertise_callback, advertise_data, advertise_settings)
expected_advertise_event_name = "".join(
- ["BleAdvertise", str(advertise_callback), "onSuccess"])
+ ["BleAdvertise",
+ str(advertise_callback), "onSuccess"])
worker = self.adv_ad.ed.handle_event(
self.bleadvertise_verify_onsuccess_handler,
expected_advertise_event_name, ([]), self.default_timeout)
@@ -227,7 +230,8 @@
self.log.debug(worker.result(self.default_timeout))
except Empty as error:
self.log.debug(" ".join(
- ["Test failed with Empty error:", str(error)]))
+ ["Test failed with Empty error:",
+ str(error)]))
test_result = False
except concurrent.futures._base.TimeoutError as error:
self.log.debug(" ".join([
@@ -247,7 +251,8 @@
self.log.debug(worker.result(self.default_timeout))
except Empty as error:
self.log.debug(" ".join(
- ["Test failed with Empty error:", str(error)]))
+ ["Test failed with Empty error:",
+ str(error)]))
test_result = False
except concurrent.futures._base.TimeoutError as error:
self.log.debug(" ".join([
@@ -318,7 +323,7 @@
iterations = 100
for i in range(iterations):
try:
- target_address, adv_callback = get_mac_address_of_generic_advertisement(
+ target_address, adv_callback, scan_callback = get_mac_address_of_generic_advertisement(
self.scn_ad, self.adv_ad)
except BtTestUtilsError as err:
self.log.error(err)
@@ -339,6 +344,7 @@
self.log.error("Failed to unbond device from advertiser.")
return False
self.adv_ad.droid.bleStopBleAdvertising(adv_callback)
+ self.scn_ad.droid.bleStopBleScan(scan_callback)
# Magic sleep to let unbonding finish
time.sleep(2)
return True
diff --git a/acts/tests/google/bt/pts/bta_lib.py b/acts/tests/google/bt/pts/bta_lib.py
deleted file mode 100644
index f75ff6b..0000000
--- a/acts/tests/google/bt/pts/bta_lib.py
+++ /dev/null
@@ -1,110 +0,0 @@
-#/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.
-"""
-Bluetooth adapter libraries
-"""
-
-from acts.test_utils.bt.bt_constants import bt_scan_mode_types
-from acts.test_utils.bt.bt_test_utils import set_bt_scan_mode
-
-import pprint
-
-
-class BtaLib():
- def __init__(self, log, mac_addr, dut):
- self.advertisement_list = []
- self.dut = dut
- self.log = log
- self.mac_addr = mac_addr
-
- def set_scan_mode(self, scan_mode):
- """Set the Scan mode of the Bluetooth Adapter"""
- for mode in bt_scan_mode_types:
- if scan_mode == mode.name:
- set_bt_scan_mode(self.dut, mode.value)
- return
-
- def set_device_name(self, line):
- """Set Bluetooth Adapter Name"""
- self.dut.droid.bluetoothSetLocalName(line)
-
- def enable(self):
- """Enable Bluetooth Adapter"""
- self.dut.droid.bluetoothToggleState(True)
-
- def disable(self):
- """Disable Bluetooth Adapter"""
- self.dut.droid.bluetoothToggleState(False)
-
- def init_bond(self):
- """Initiate bond to PTS device"""
- self.dut.droid.bluetoothDiscoverAndBond(self.mac_addr)
-
- def start_discovery(self):
- """Start BR/EDR Discovery"""
- self.dut.droid.bluetoothStartDiscovery()
-
- def stop_discovery(self):
- """Stop BR/EDR Discovery"""
- self.dut.droid.bluetoothCancelDiscovery()
-
- def get_discovered_devices(self):
- """Get Discovered Br/EDR Devices"""
- if self.dut.droid.bluetoothIsDiscovering():
- self.dut.droid.bluetoothCancelDiscovery()
- self.log.info(
- pprint.pformat(self.dut.droid.bluetoothGetDiscoveredDevices()))
-
- def bond(self):
- """Bond to PTS device"""
- self.dut.droid.bluetoothBond(self.mac_addr)
-
- def disconnect(self):
- """BTA disconnect"""
- self.dut.droid.bluetoothDisconnectConnected(self.mac_addr)
-
- def unbond(self):
- """Unbond from PTS device"""
- self.dut.droid.bluetoothUnbond(self.mac_addr)
-
- def start_pairing_helper(self, line):
- """Start or stop Bluetooth Pairing Helper"""
- if line:
- self.dut.droid.bluetoothStartPairingHelper(bool(line))
- else:
- self.dut.droid.bluetoothStartPairingHelper()
-
- def push_pairing_pin(self, line):
- """Push pairing pin to the Android Device"""
- self.dut.droid.eventPost("BluetoothActionPairingRequestUserConfirm",
- line)
-
- def get_pairing_pin(self):
- """Get pairing PIN"""
- self.log.info(
- self.dut.ed.pop_event("BluetoothActionPairingRequest", 1))
-
- def fetch_uuids_with_sdp(self):
- """BTA fetch UUIDS with SDP"""
- self.log.info(self.dut.droid.bluetoothFetchUuidsWithSdp(self.mac_addr))
-
- def connect_profiles(self):
- """Connect available profiles"""
- self.dut.droid.bluetoothConnectBonded(self.mac_addr)
-
- def tts_speak(self):
- """Open audio channel by speaking characters"""
- self.dut.droid.ttsSpeak(self.mac_addr)
diff --git a/acts/tests/google/bt/pts/cmd_input.py b/acts/tests/google/bt/pts/cmd_input.py
index 10d2839..7cf7f20 100644
--- a/acts/tests/google/bt/pts/cmd_input.py
+++ b/acts/tests/google/bt/pts/cmd_input.py
@@ -19,13 +19,9 @@
from acts.test_utils.bt.bt_constants import bt_scan_mode_types
from acts.test_utils.bt.bt_constants import gatt_server_responses
import acts.test_utils.bt.gatt_test_database as gatt_test_database
-from acts.test_utils.bt.ble_lib import BleLib
-from acts.test_utils.bt.bta_lib import BtaLib
-from acts.test_utils.bt.config_lib import ConfigLib
-from acts.test_utils.bt.gattc_lib import GattClientLib
-from acts.test_utils.bt.gatts_lib import GattServerLib
-from acts.test_utils.bt.rfcomm_lib import RfcommLib
+from acts.test_utils.bt.bt_carkit_lib import E2eBtCarkitLib
+import threading
import time
import cmd
"""Various Global Strings"""
@@ -34,8 +30,8 @@
class CmdInput(cmd.Cmd):
+ android_devices = []
"""Simple command processor for Bluetooth PTS Testing"""
- gattc_lib = None
def connect_hsp_helper(self, ad):
"""A helper function for making HSP connections"""
@@ -65,17 +61,16 @@
self.pri_dut = android_devices[0]
if len(android_devices) > 1:
self.sec_dut = android_devices[1]
+ if len(android_devices) > 2:
self.ter_dut = android_devices[2]
+ if len(android_devices) > 3:
+ self.qua_dut = android_devices[3]
self.mac_addr = mac_addr
self.log = log
-
- # Initialize libraries
- self.config_lib = ConfigLib(log, self.pri_dut)
- self.bta_lib = BtaLib(log, mac_addr, self.pri_dut)
- self.ble_lib = BleLib(log, mac_addr, self.pri_dut)
- self.gattc_lib = GattClientLib(log, mac_addr, self.pri_dut)
- self.gatts_lib = GattServerLib(log, mac_addr, self.pri_dut)
- self.rfcomm_lib = RfcommLib(log, mac_addr, self.pri_dut)
+ self.pri_dut.bta.set_target_mac_addr(mac_addr)
+ self.pri_dut.gattc.set_target_mac_addr(mac_addr)
+ self.pri_dut.rfcomm.set_target_mac_addr(mac_addr)
+ self.bt_carkit_lib = E2eBtCarkitLib(self.log, mac_addr)
def emptyline(self):
pass
@@ -84,8 +79,38 @@
"End Script"
return True
+ def do_reset_mac_address(self, line):
+ """Reset target mac address for libraries based on the current connected device"""
+ try:
+ device = self.pri_dut.droid.bluetoothGetConnectedDevices()[0]
+ #self.setup_vars(self.android_devices, device['address'], self.log)
+ self.log.info("New device is {}".format(device))
+ except Exception as err:
+ self.log.info("Failed to setup new vars with {}".format(err))
+
"""Begin GATT Client wrappers"""
+ def do_gattc_socket_conn_begin_connect_thread_psm(self, line):
+ cmd = ""
+ try:
+ self.pri_dut.gattc.socket_conn_begin_connect_thread_psm(line)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_gattc_socket_conn_begin_accept_thread_psm(self, line):
+ cmd = ""
+ try:
+ self.pri_dut.gattc.socket_conn_begin_accept_thread_psm(line)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_gattc_request_le_connection_parameters(self, line):
+ cmd = ""
+ try:
+ self.pri_dut.gattc.request_le_connection_parameters()
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
def do_gattc_connect_over_le(self, line):
"""Perform GATT connection over LE"""
cmd = "Gatt connect over LE"
@@ -93,7 +118,7 @@
autoconnect = False
if line:
autoconnect = bool(line)
- self.gattc_lib.connect_over_le(autoconnect)
+ self.pri_dut.gattc.connect_over_le(autoconnect)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -101,7 +126,7 @@
"""Perform GATT connection over BREDR"""
cmd = "Gatt connect over BR/EDR"
try:
- self.gattc_lib.connect_over_bredr()
+ self.pri_dut.gattc.connect_over_bredr()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -109,7 +134,7 @@
"""Perform GATT disconnect"""
cmd = "Gatt Disconnect"
try:
- self.gattc_lib.disconnect()
+ self.pri_dut.gattc.disconnect()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -117,7 +142,7 @@
"""GATT client read Characteristic by UUID."""
cmd = "GATT client read Characteristic by UUID."
try:
- self.gattc_lib.read_char_by_uuid(line)
+ self.pri_dut.gattc.read_char_by_uuid(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -125,7 +150,7 @@
"""Request MTU Change of input value"""
cmd = "Request MTU Value"
try:
- self.gattc_lib.request_mtu(line)
+ self.pri_dut.gattc.request_mtu(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -135,7 +160,7 @@
"""
cmd = "Discovery Services and list all UUIDS"
try:
- self.gattc_lib.list_all_uuids()
+ self.pri_dut.gattc.list_all_uuids()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -143,7 +168,7 @@
"""GATT Client discover services of GATT Server"""
cmd = "Discovery Services of GATT Server"
try:
- self.gattc_lib.discover_services()
+ self.pri_dut.gattc.discover_services()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -151,7 +176,7 @@
"""Perform Gatt Client Refresh"""
cmd = "GATT Client Refresh"
try:
- self.gattc_lib.refresh()
+ self.pri_dut.gattc.refresh()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -161,7 +186,7 @@
"""
cmd = "GATT Client Read By Instance ID"
try:
- self.gattc_lib.read_char_by_instance_id(line)
+ self.pri_dut.gattc.read_char_by_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -169,7 +194,7 @@
"""GATT Client Write to Characteristic by instance ID"""
cmd = "GATT Client write to Characteristic by instance ID"
try:
- self.gattc_lib.write_char_by_instance_id(line)
+ self.pri_dut.gattc.write_char_by_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -177,7 +202,7 @@
"""GATT Client Write to Characteristic by instance ID"""
cmd = "GATT Client write to Characteristic by instance ID"
try:
- self.gattc_lib.write_char_by_instance_id_value(line)
+ self.pri_dut.gattc.write_char_by_instance_id_value(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -185,14 +210,14 @@
"""GATT Client Write to Char that doesn't have write permission"""
cmd = "GATT Client Write to Char that doesn't have write permission"
try:
- self.gattc_lib.mod_write_char_by_instance_id(line)
+ self.pri_dut.gattc.mod_write_char_by_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
def do_gattc_write_invalid_char_by_instance_id(self, line):
"""GATT Client Write to Char that doesn't exists"""
try:
- self.gattc_lib.write_invalid_char_by_instance_id(line)
+ self.pri_dut.gattc.write_invalid_char_by_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -200,7 +225,7 @@
"""GATT Client Read Char that doesn't have write permission"""
cmd = "GATT Client Read Char that doesn't have write permission"
try:
- self.gattc_lib.mod_read_char_by_instance_id(line)
+ self.pri_dut.gattc.mod_read_char_by_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -208,7 +233,7 @@
"""GATT Client Read Char that doesn't exists"""
cmd = "GATT Client Read Char that doesn't exists"
try:
- self.gattc_lib.read_invalid_char_by_instance_id(line)
+ self.pri_dut.gattc.read_invalid_char_by_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -216,7 +241,7 @@
"""GATT Client Write to Desc that doesn't have write permission"""
cmd = "GATT Client Write to Desc that doesn't have write permission"
try:
- self.gattc_lib.mod_write_desc_by_instance_id(line)
+ self.pri_dut.gattc.mod_write_desc_by_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -224,7 +249,7 @@
"""GATT Client Write to Desc that doesn't exists"""
cmd = "GATT Client Write to Desc that doesn't exists"
try:
- self.gattc_lib.write_invalid_desc_by_instance_id(line)
+ self.pri_dut.gattc.write_invalid_desc_by_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -232,7 +257,7 @@
"""GATT Client Read Desc that doesn't have write permission"""
cmd = "GATT Client Read Desc that doesn't have write permission"
try:
- self.gattc_lib.mod_read_desc_by_instance_id(line)
+ self.pri_dut.gattc.mod_read_desc_by_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -240,7 +265,7 @@
"""GATT Client Read Desc that doesn't exists"""
cmd = "GATT Client Read Desc that doesn't exists"
try:
- self.gattc_lib.read_invalid_desc_by_instance_id(line)
+ self.pri_dut.gattc.read_invalid_desc_by_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -248,7 +273,7 @@
"""GATT Client Read Char that doesn't have write permission"""
cmd = "GATT Client Read Char that doesn't have write permission"
try:
- self.gattc_lib.mod_read_char_by_uuid_and_instance_id(line)
+ self.pri_dut.gattc.mod_read_char_by_uuid_and_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -256,7 +281,7 @@
"""GATT Client Read Char that doesn't exist"""
cmd = "GATT Client Read Char that doesn't exist"
try:
- self.gattc_lib.read_invalid_char_by_uuid(line)
+ self.pri_dut.gattc.read_invalid_char_by_uuid(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -264,7 +289,7 @@
"""GATT Client Write to Descriptor by instance ID"""
cmd = "GATT Client Write to Descriptor by instance ID"
try:
- self.gattc_lib.write_desc_by_instance_id(line)
+ self.pri_dut.gattc.write_desc_by_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -272,7 +297,15 @@
"""GATT Client Enable Notification on Descriptor by instance ID"""
cmd = "GATT Client Enable Notification on Descriptor by instance ID"
try:
- self.gattc_lib.enable_notification_desc_by_instance_id(line)
+ self.pri_dut.gattc.enable_notification_desc_by_instance_id(line)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_gattc_enable_indication_desc_by_instance_id(self, line):
+ """GATT Client Enable Indication on Descriptor by instance ID"""
+ cmd = "GATT Client Enable Indication on Descriptor by instance ID"
+ try:
+ self.pri_dut.gattc.enable_indication_desc_by_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -280,7 +313,7 @@
"""GATT Client enable all notifications"""
cmd = "GATT Client enable all notifications"
try:
- self.gattc_lib.char_enable_all_notifications()
+ self.pri_dut.gattc.char_enable_all_notifications()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -288,7 +321,7 @@
"""GATT Client read char by non-existant instance id"""
cmd = "GATT Client read char by non-existant instance id"
try:
- self.gattc_lib.read_char_by_invalid_instance_id(line)
+ self.pri_dut.gattc.read_char_by_invalid_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -296,7 +329,7 @@
"""Begin a reliable write on the Bluetooth Gatt Client"""
cmd = "GATT Client Begin Reliable Write"
try:
- self.gattc_lib.begin_reliable_write()
+ self.pri_dut.gattc.begin_reliable_write()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -304,7 +337,7 @@
"""Abort a reliable write on the Bluetooth Gatt Client"""
cmd = "GATT Client Abort Reliable Write"
try:
- self.gattc_lib.abort_reliable_write()
+ self.pri_dut.gattc.abort_reliable_write()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -312,7 +345,7 @@
"""Execute a reliable write on the Bluetooth Gatt Client"""
cmd = "GATT Client Execute Reliable Write"
try:
- self.gattc_lib.execute_reliable_write()
+ self.pri_dut.gattc.execute_reliable_write()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -320,7 +353,7 @@
"""GATT Client read all Characteristic values"""
cmd = "GATT Client read all Characteristic values"
try:
- self.gattc_lib.read_all_char()
+ self.pri_dut.gattc.read_all_char()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -328,7 +361,7 @@
"""Write to every Characteristic on the GATT server"""
cmd = "GATT Client Write All Characteristics"
try:
- self.gattc_lib.write_all_char(line)
+ self.pri_dut.gattc.write_all_char(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -336,7 +369,7 @@
""" Write to every Descriptor on the GATT server """
cmd = "GATT Client Write All Descriptors"
try:
- self.gattc_lib.write_all_desc(line)
+ self.pri_dut.gattc.write_all_desc(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -344,7 +377,7 @@
"""Write 0x00 or 0x02 to notification descriptor [instance_id]"""
cmd = "Write 0x00 0x02 to notification descriptor"
try:
- self.gattc_lib.write_desc_notification_by_instance_id(line)
+ self.pri_dut.gattc.write_desc_notification_by_instance_id(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -352,7 +385,7 @@
"""Discover service by uuid"""
cmd = "Discover service by uuid"
try:
- self.gattc_lib.discover_service_by_uuid(line)
+ self.pri_dut.gattc.discover_service_by_uuid(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -360,7 +393,7 @@
"""Read all Descriptor values"""
cmd = "Read all Descriptor values"
try:
- self.gattc_lib.read_all_desc()
+ self.pri_dut.gattc.read_all_desc()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -371,7 +404,7 @@
"""Close Bluetooth Gatt Servers"""
cmd = "Close Bluetooth Gatt Servers"
try:
- self.gatts_lib.close_bluetooth_gatt_servers()
+ self.pri_dut.gatts.close_bluetooth_gatt_servers()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -424,7 +457,7 @@
"""
cmd = "Discovery Services and list all UUIDS"
try:
- self.gatts_lib.list_all_uuids()
+ self.pri_dut.gatts.list_all_uuids()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -432,7 +465,7 @@
"""Send a response to the GATT Client"""
cmd = "GATT server send response"
try:
- self.gatts_lib.send_response(line)
+ self.pri_dut.gatts.send_response(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -446,14 +479,15 @@
confirm = False
if confirm_str.lower() == 'true':
confirm = True
- self.gatts_lib.notify_characteristic_changed(instance_id, confirm)
+ self.pri_dut.gatts.notify_characteristic_changed(
+ instance_id, confirm)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
def do_gatts_setup_database(self, line):
cmd = "Setup GATT Server database: {}".format(line)
try:
- self.gatts_lib.setup_gatts_db(
+ self.pri_dut.gatts.setup_gatts_db(
gatt_test_database.GATT_SERVER_DB_MAPPING.get(line))
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -468,7 +502,7 @@
value = []
for i in range(size):
value.append(i % 256)
- self.gatts_lib.gatt_server_characteristic_set_value_by_instance_id(
+ self.pri_dut.gatts.characteristic_set_value_by_instance_id(
instance_id, value)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -480,7 +514,8 @@
info = line.split()
instance_id = info[0]
confirm = bool(info[1])
- self.gatts_lib.notify_characteristic_changed(instance_id, confirm)
+ self.pri_dut.gatts.notify_characteristic_changed(
+ instance_id, confirm)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -488,7 +523,7 @@
"""Open a GATT Server instance"""
cmd = "Open an empty GATT Server"
try:
- self.gatts_lib.open()
+ self.pri_dut.gatts.open()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -496,7 +531,7 @@
"""Clear BluetoothGattServices from BluetoothGattServer"""
cmd = "Clear BluetoothGattServices from BluetoothGattServer"
try:
- self.gatts_lib.gatt_server_clear_services()
+ self.pri_dut.gatts.gatt_server_clear_services()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -504,7 +539,7 @@
"""Send continous response with random data"""
cmd = "Send continous response with random data"
try:
- self.gatts_lib.send_continuous_response(line)
+ self.pri_dut.gatts.send_continuous_response(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -512,7 +547,7 @@
"""Send continous response including requested data"""
cmd = "Send continous response including requested data"
try:
- self.gatts_lib.send_continuous_response_data(line)
+ self.pri_dut.gatts.send_continuous_response_data(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -538,7 +573,7 @@
return completions
def complete_ble_stop_advertisement(self, text, line, begidx, endidx):
- str_adv_list = list(map(str, self.ble_lib.ADVERTISEMENT_LIST))
+ str_adv_list = list(map(str, self.pri_dut.ble.ADVERTISEMENT_LIST))
if not text:
completions = str_adv_list[:]
else:
@@ -549,21 +584,39 @@
"""Start a connectable LE advertisement"""
cmd = "Start a connectable LE advertisement"
try:
- self.ble_lib.start_generic_connectable_advertisement(line)
+ self.pri_dut.ble.start_generic_connectable_advertisement(line)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def _start_max_advertisements(self, ad):
+ self.pri_dut.ble.start_max_advertisements(ad)
+
+ def do_ble_start_generic_connectable_beacon_swarm(self, line):
+ """Start a connectable LE advertisement"""
+ cmd = "Start as many advertisements as possible on all devices"
+ try:
+ threads = []
+ for ad in self.android_devices:
+ thread = threading.Thread(
+ target=self._start_max_advertisements, args=([ad]))
+ threads.append(thread)
+ thread.start()
+ for t in threads:
+ t.join()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
def do_ble_start_connectable_advertisement_set(self, line):
"""Start a connectable advertisement set"""
try:
- self.ble_lib.start_connectable_advertisement_set(line)
+ self.pri_dut.ble.start_connectable_advertisement_set(line)
except Exception as err:
self.log.error("Failed to start advertisement: {}".format(err))
def do_ble_stop_all_advertisement_set(self, line):
"""Stop all advertisement sets"""
try:
- self.ble_lib.stop_all_advertisement_set(line)
+ self.pri_dut.ble.stop_all_advertisement_set(line)
except Exception as err:
self.log.error("Failed to stop advertisement: {}".format(err))
@@ -572,7 +625,7 @@
[uuid1 uuid2 ... uuidN]"""
cmd = "Add a valid service UUID to the advertisement."
try:
- self.ble_lib.adv_add_service_uuid_list(line)
+ self.pri_dut.ble.adv_add_service_uuid_list(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -580,7 +633,7 @@
"""Include local name in the advertisement. inputs: [true|false]"""
cmd = "Include local name in the advertisement."
try:
- self.ble_lib.adv_data_include_local_name(line)
+ self.pri_dut.ble.adv_data_include_local_name(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -588,7 +641,7 @@
"""Include tx power level in the advertisement. inputs: [true|false]"""
cmd = "Include local name in the advertisement."
try:
- self.ble_lib.adv_data_include_tx_power_level(line)
+ self.pri_dut.ble.adv_data_include_tx_power_level(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -597,7 +650,7 @@
[id data1 data2 ... dataN]"""
cmd = "Include manufacturer id and data to the advertisment."
try:
- self.ble_lib.adv_data_add_manufacturer_data(line)
+ self.pri_dut.ble.adv_data_add_manufacturer_data(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -605,19 +658,19 @@
"""Start a nonconnectable LE advertisement"""
cmd = "Start a nonconnectable LE advertisement"
try:
- self.ble_lib.start_generic_nonconnectable_advertisement(line)
+ self.pri_dut.ble.start_generic_nonconnectable_advertisement(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
def do_ble_list_active_advertisement_ids(self, line):
"""List all active BLE advertisements"""
- self.log.info("IDs: {}".format(self.ble_lib.advertisement_list))
+ self.log.info("IDs: {}".format(self.pri_dut.ble.advertisement_list))
def do_ble_stop_all_advertisements(self, line):
"""Stop all LE advertisements"""
cmd = "Stop all LE advertisements"
try:
- self.ble_lib.stop_all_advertisements(line)
+ self.pri_dut.ble.stop_all_advertisements(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -625,7 +678,7 @@
"""Stop an LE advertisement"""
cmd = "Stop a connectable LE advertisement"
try:
- self.ble_lib.ble_stop_advertisement(line)
+ self.pri_dut.ble.ble_stop_advertisement(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -652,7 +705,7 @@
"""Set the Scan mode of the Bluetooth Adapter"""
cmd = "Set the Scan mode of the Bluetooth Adapter"
try:
- self.bta_lib.set_scan_mode(line)
+ self.pri_dut.bta.set_scan_mode(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -660,7 +713,7 @@
"""Set Bluetooth Adapter Name"""
cmd = "Set Bluetooth Adapter Name"
try:
- self.bta_lib.set_device_name(line)
+ self.pri_dut.bta.set_device_name(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -668,7 +721,7 @@
"""Enable Bluetooth Adapter"""
cmd = "Enable Bluetooth Adapter"
try:
- self.bta_lib.enable()
+ self.pri_dut.bta.enable()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -676,7 +729,7 @@
"""Disable Bluetooth Adapter"""
cmd = "Disable Bluetooth Adapter"
try:
- self.bta_lib.disable()
+ self.pri_dut.bta.disable()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -684,7 +737,7 @@
"""Initiate bond to PTS device"""
cmd = "Initiate Bond"
try:
- self.bta_lib.init_bond()
+ self.pri_dut.bta.init_bond()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -692,7 +745,7 @@
"""Start BR/EDR Discovery"""
cmd = "Start BR/EDR Discovery"
try:
- self.bta_lib.start_discovery()
+ self.pri_dut.bta.start_discovery()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -700,7 +753,7 @@
"""Stop BR/EDR Discovery"""
cmd = "Stop BR/EDR Discovery"
try:
- self.bta_lib.stop_discovery()
+ self.pri_dut.bta.stop_discovery()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -708,7 +761,7 @@
"""Get Discovered Br/EDR Devices"""
cmd = "Get Discovered Br/EDR Devices\n"
try:
- self.bta_lib.get_discovered_devices()
+ self.pri_dut.bta.get_discovered_devices()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -716,7 +769,7 @@
"""Bond to PTS device"""
cmd = "Bond to the PTS dongle directly"
try:
- self.bta_lib.bond()
+ self.pri_dut.bta.bond()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -724,7 +777,7 @@
"""BTA disconnect"""
cmd = "BTA disconnect"
try:
- self.bta_lib.disconnect()
+ self.pri_dut.bta.disconnect()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -732,7 +785,7 @@
"""Unbond from PTS device"""
cmd = "Unbond from the PTS dongle"
try:
- self.bta_lib.unbond()
+ self.pri_dut.bta.unbond()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -740,7 +793,7 @@
"""Start or stop Bluetooth Pairing Helper"""
cmd = "Start or stop BT Pairing helper"
try:
- self.bta_lib.start_pairing_helper(line)
+ self.pri_dut.bta.start_pairing_helper(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -748,7 +801,7 @@
"""Push pairing pin to the Android Device"""
cmd = "Push the pin to the Android Device"
try:
- self.bta_lib.push_pairing_pin(line)
+ self.pri_dut.bta.push_pairing_pin(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -756,7 +809,7 @@
"""Get pairing PIN"""
cmd = "Get Pin Info"
try:
- self.bta_lib.get_pairing_pin()
+ self.pri_dut.bta.get_pairing_pin()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -764,7 +817,7 @@
"""BTA fetch UUIDS with SDP"""
cmd = "Fetch UUIDS with SDP"
try:
- self.bta_lib.fetch_uuids_with_sdp()
+ self.pri_dut.bta.fetch_uuids_with_sdp()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -772,14 +825,14 @@
"""Connect available profiles"""
cmd = "Connect all profiles possible"
try:
- self.bta_lib.connect_profiles()
+ self.pri_dut.bta.connect_profiles()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
def do_bta_tts_speak(self, line):
cmd = "Open audio channel by speaking characters"
try:
- self.bta_lib.tts_speak()
+ self.pri_dut.bta.tts_speak()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -789,7 +842,7 @@
def do_rfcomm_connect(self, line):
"""Perform an RFCOMM connect"""
try:
- self.rfcomm_lib.connect(line)
+ self.pri_dut.rfcomm.connect(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -797,7 +850,7 @@
"""Open rfcomm socket"""
cmd = "Open RFCOMM socket"
try:
- self.rfcomm_lib.open_rfcomm_socket()
+ self.pri_dut.rfcomm.open_rfcomm_socket()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -805,42 +858,42 @@
"""Open L2CAP socket"""
cmd = "Open L2CAP socket"
try:
- self.rfcomm_lib.open_l2cap_socket()
+ self.pri_dut.rfcomm.open_l2cap_socket()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
def do_rfcomm_write(self, line):
cmd = "Write String data over an RFCOMM connection"
try:
- self.rfcomm_lib.write(line)
+ self.pri_dut.rfcomm.write(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
def do_rfcomm_write_binary(self, line):
cmd = "Write String data over an RFCOMM connection"
try:
- self.rfcomm_lib.write_binary(line)
+ self.pri_dut.rfcomm.write_binary(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
def do_rfcomm_end_connect(self, line):
cmd = "End RFCOMM connection"
try:
- self.rfcomm_lib.end_connect()
+ self.pri_dut.rfcomm.end_connect()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
def do_rfcomm_accept(self, line):
cmd = "Accept RFCOMM connection"
try:
- self.rfcomm_lib.accept(line)
+ self.pri_dut.rfcomm.accept(line)
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
def do_rfcomm_stop(self, line):
cmd = "STOP RFCOMM Connection"
try:
- self.rfcomm_lib.stop()
+ self.pri_dut.rfcomm.stop()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -848,7 +901,7 @@
"""Open L2CAP socket"""
cmd = "Open L2CAP socket"
try:
- self.rfcomm_lib.open_l2cap_socket()
+ self.pri_dut.rfcomm.open_l2cap_socket()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -859,7 +912,7 @@
"""Reset Bluetooth Config file"""
cmd = "Reset Bluetooth Config file"
try:
- self.config_lib.reset()
+ self.pri_dut.config.reset()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -867,7 +920,7 @@
"""Set NonBondable Mode"""
cmd = "Set NonBondable Mode"
try:
- self.config_lib.set_nonbond()
+ self.pri_dut.config.set_nonbond()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
@@ -875,13 +928,32 @@
"""Set Disable MITM"""
cmd = "Set Disable MITM"
try:
- self.config_lib.set_disable_mitm()
+ self.pri_dut.config.set_disable_mitm()
except Exception as err:
self.log.info(FAILURE.format(cmd, err))
"""End Config wrappers"""
"""Begin HFP/HSP wrapper"""
+ def do_bta_hfp_start_voice_recognition(self, line):
+ self.pri_dut.droid.bluetoothHspStartVoiceRecognition(self.mac_addr)
+
+ def do_bta_hfp_stop_voice_recognition(self, line):
+ self.pri_dut.droid.bluetoothHspStopVoiceRecognition(self.mac_addr)
+
+ def do_test_clcc_response(self, line):
+ # Experimental
+ clcc_index = 1
+ clcc_direction = 1
+ clcc_status = 4
+ clcc_mode = 0
+ clcc_mpty = False
+ clcc_number = "18888888888"
+ clcc_type = 0
+ self.pri_dut.droid.bluetoothHspClccResponse(
+ clcc_index, clcc_direction, clcc_status, clcc_mode, clcc_mpty,
+ clcc_number, clcc_type)
+
def do_bta_hsp_force_sco_audio_on(self, line):
"""HFP/HSP Force SCO Audio ON"""
cmd = "HFP/HSP Force SCO Audio ON"
@@ -987,3 +1059,334 @@
self.log.info(FAILURE.format(cmd, err))
"""End HID wrappers"""
+ """Begin carkit test wrappers"""
+
+ def do_test_suite_generic_bt_tests(self, line):
+ """Run generic Bluetooth connection test suite"""
+ generic_bt_tests = [
+ tuple((self.bt_carkit_lib.disconnect_reconnect_multiple_iterations,
+ [self.pri_dut])),
+ tuple((self.bt_carkit_lib.disconnect_a2dp_only_then_reconnect,
+ [self.pri_dut])),
+ tuple((self.bt_carkit_lib.disconnect_hsp_only_then_reconnect,
+ [self.pri_dut])),
+ tuple((
+ self.bt_carkit_lib.disconnect_both_hsp_and_a2dp_then_reconnect,
+ [self.pri_dut])),
+ ]
+ try:
+ for func, param in generic_bt_tests:
+ try:
+ func(param)
+ except Exception:
+ self.log.info("Test {} failed.".format(func))
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_connect_hsp_helper(self, line):
+ """Connect to HSP/HFP with additional tries and help"""
+ cmd = "Connect to hsp with some help"
+ try:
+ self.bt_carkit_lib.connect_hsp_helper(self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_hfp_setup_multi_incomming_calls(self, line):
+ """Setup two incomming calls"""
+ cmd = "Setup multiple incomming calls"
+ try:
+ self.bt_carkit_lib.setup_multi_call(self.sec_dut, self.ter_dut,
+ self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_disconnect_reconnect_multiple_iterations(self, line):
+ """Quick disconnect/reconnect stress test"""
+ try:
+ self.bt_carkit_lib.disconnect_reconnect_multiple_iterations(
+ self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_disconnect_a2dp_only_then_reconnect(self, line):
+ """Test disconnect-reconnect a2dp only scenario from phone."""
+ cmd = "Test disconnect-reconnect a2dp only scenario from phone."
+ try:
+ self.bt_carkit_lib.disconnect_a2dp_only_then_reconnect(
+ self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_disconnect_hsp_only_then_reconnect(self, line):
+ """Test disconnect-reconnect hsp only scenario from phone."""
+ cmd = "Test disconnect-reconnect hsp only scenario from phone."
+ try:
+ self.bt_carkit_lib.disconnect_hsp_only_then_reconnect(self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_disconnect_both_hsp_and_a2dp_then_reconnect(self, line):
+ """Test disconnect-reconnect hsp and a2dp scenario from phone."""
+ cmd = "Test disconnect-reconnect hsp and a2dp scenario from phone."
+ try:
+ self.bt_carkit_lib.disconnect_both_hsp_and_a2dp_then_reconnect(
+ self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_outgoing_call_private_number(self, line):
+ """Test outgoing call scenario from phone to private number"""
+ cmd = "Test outgoing call scenario from phone to private number"
+ try:
+ self.bt_carkit_lib.outgoing_call_private_number(
+ self.pri_dut, self.ter_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_outgoing_call_a2dp_play_before_and_after(self, line):
+ """Test outgoing call scenario while playing music. Music should resume
+ after call."""
+ cmd = "Test outgoing call scenario while playing music. Music should " \
+ "resume after call."
+ try:
+ self.bt_carkit_lib.outgoing_call_a2dp_play_before_and_after(
+ self.pri_dut, self.sec_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_outgoing_call_unknown_contact(self, line):
+ """Test outgoing call scenario from phone to unknow contact"""
+ cmd = "Test outgoing call scenario from phone to unknow contact"
+ try:
+ self.bt_carkit_lib.outgoing_call_unknown_contact(
+ self.pri_dut, self.ter_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_incomming_call_private_number(self, line):
+ """Test incomming call scenario to phone from unknown contact"""
+ cmd = "Test incomming call scenario to phone from unknown contact"
+ try:
+ self.bt_carkit_lib.incomming_call_private_number(
+ self.pri_dut, self.ter_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_outgoing_call_multiple_iterations(self, line):
+ """Test outgoing call quickly 3 times"""
+ cmd = "Test outgoing call quickly 3 times"
+ try:
+ self.bt_carkit_lib.outgoing_call_multiple_iterations(
+ self.pri_dut, self.sec_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_outgoing_call_hsp_disabled_then_enabled_during_call(self, line):
+ """Test outgoing call hsp disabled then enable during call."""
+ cmd = "Test outgoing call hsp disabled then enable during call."
+ try:
+ self.bt_carkit_lib.outgoing_call_hsp_disabled_then_enabled_during_call(
+ self.pri_dut, self.sec_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_call_audio_routes(self, line):
+ """Test various audio routes scenario from phone."""
+ cmd = "Test various audio routes scenario from phone."
+ try:
+ self.bt_carkit_lib.call_audio_routes(self.pri_dut, self.sec_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_sms_receive_different_sizes(self, line):
+ """Test recieve sms of different sizes."""
+ cmd = "Test recieve sms of different sizes."
+ try:
+ self.bt_carkit_lib.sms_receive_different_sizes(
+ self.pri_dut, self.sec_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_sms_receive_multiple(self, line):
+ """Test recieve sms of different sizes."""
+ cmd = "Test recieve sms of different sizes."
+ try:
+ self.bt_carkit_lib.sms_receive_multiple(self.pri_dut, self.sec_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_sms_send_outgoing_texts(self, line):
+ """Test send sms of different sizes."""
+ cmd = "Test send sms of different sizes."
+ try:
+ self.bt_carkit_lib.sms_send_outgoing_texts(self.pri_dut,
+ self.sec_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_sms_during_incomming_call(self, line):
+ """Test incomming call scenario to phone from unknown contact"""
+ cmd = "Test incomming call scenario to phone from unknown contact"
+ try:
+ self.bt_carkit_lib.sms_during_incomming_call(
+ self.pri_dut, self.sec_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_multi_incomming_call(self, line):
+ """Test 2 incomming calls scenario to phone."""
+ cmd = "Test 2 incomming calls scenario to phone."
+ try:
+ self.bt_carkit_lib.multi_incomming_call(self.pri_dut, self.sec_dut,
+ self.ter_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_multi_call_audio_routing(self, line):
+ """Test 2 incomming calls scenario to phone, then test audio routing."""
+ cmd = "Test 2 incomming calls scenario to phone, then test audio" \
+ "routing."
+ try:
+ self.bt_carkit_lib.multi_call_audio_routing(
+ self.pri_dut, self.sec_dut, self.ter_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_multi_call_swap_multiple_times(self, line):
+ """Test 2 incomming calls scenario to phone, then swap the calls
+ multiple times"""
+ cmd = "Test 2 incomming calls scenario to phone, then swap the calls" \
+ "multiple times"
+ try:
+ self.bt_carkit_lib.multi_call_swap_multiple_times(
+ self.pri_dut, self.sec_dut, self.ter_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_multi_call_join_conference_call(self, line):
+ """Test 2 incomming calls scenario to phone then join the calls."""
+ cmd = "Test 2 incomming calls scenario to phone then join the calls."
+ try:
+ self.bt_carkit_lib.multi_call_join_conference_call(
+ self.pri_dut, self.sec_dut, self.ter_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_multi_call_join_conference_call_hangup_conf_call(self, line):
+ """Test 2 incomming calls scenario to phone then join the calls,
+ then terminate the call from the primary dut."""
+ cmd = "Test 2 incomming calls scenario to phone then join the calls, " \
+ "then terminate the call from the primary dut."
+ try:
+ self.bt_carkit_lib.multi_call_join_conference_call_hangup_conf_call(
+ self.pri_dut, self.sec_dut, self.ter_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_outgoing_multi_call_join_conference_call(self, line):
+ """Test 2 outgoing calls scenario from phone then join the calls."""
+ cmd = "Test 2 outgoing calls scenario from phone then join the calls."
+ try:
+ self.bt_carkit_lib.outgoing_multi_call_join_conference_call(
+ self.pri_dut, self.sec_dut, self.ter_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_multi_call_join_conference_call_audio_routes(self, line):
+ """Test 2 incomming calls scenario to phone then join the calls,
+ then test different audio routes."""
+ cmd = "Test 2 incomming calls scenario to phone then join the calls, " \
+ "then test different audio routes."
+ try:
+ self.bt_carkit_lib.multi_call_join_conference_call_audio_routes(
+ self.pri_dut, self.sec_dut, self.ter_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_avrcp_play_pause(self, line):
+ """Test avrcp play/pause commands multiple times from phone"""
+ cmd = "Test avrcp play/pause commands multiple times from phone"
+ try:
+ self.bt_carkit_lib.avrcp_play_pause(self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_avrcp_next_previous_song(self, line):
+ """Test AVRCP go to the next song then the previous song."""
+ cmd = "Test AVRCP go to the next song then the previous song."
+ try:
+ self.bt_carkit_lib.avrcp_next_previous_song(self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_avrcp_next_previous(self, line):
+ """Test AVRCP go to the next song then the press previous after a few
+ seconds."""
+ cmd = "Test AVRCP go to the next song then the press previous after " \
+ "a few seconds."
+ try:
+ self.bt_carkit_lib.avrcp_next_previous(self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_avrcp_next_repetative(self, line):
+ """Test AVRCP go to the next 10 times"""
+ cmd = "Test AVRCP go to the next 10 times"
+ try:
+ self.bt_carkit_lib.avrcp_next_repetative(self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def _process_question(self, question, expected_response):
+ while True:
+ try:
+ result = input(question.format(level)).lower()
+ except Exception as err:
+ print(err)
+
+ def do_e2e_cycle_battery_level(self, line):
+ """Cycle battery level through different values and verify result on carkit"""
+ cmd = "Test that verifies battery level indicator changes with the " \
+ "phone. Phone current level."
+ try:
+ self.bt_carkit_lib.cycle_absolute_volume_control(self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_cycle_absolute_volume_control(self, line):
+ """Cycle media volume level through different values and verify result on carkit"""
+ cmd = "Test aboslute volume on carkit by changed volume levels from phone."
+ try:
+ self.bt_carkit_lib.cycle_absolute_volume_control(self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_test_voice_recognition_from_phone(self, line):
+ """Test Voice Recognition from phone."""
+ cmd = "Test voice recognition from phone."
+ try:
+ self.bt_carkit_lib.test_voice_recognition_from_phone(self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ def do_e2e_test_audio_and_voice_recognition_from_phone(self, line):
+ """Test Voice Recognition from phone and confirm music audio continues."""
+ cmd = "Test Voice Recognition from phone and confirm music audio continues."
+ try:
+ self.bt_carkit_lib.test_audio_and_voice_recognition_from_phone(
+ self.pri_dut)
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ """End carkit test wrappers"""
+ """Begin adb shell test wrappers"""
+
+ def do_set_battery_level(self, line):
+ """Set battery level based on input"""
+ cmd = "Set battery level based on input"
+ try:
+ self.pri_dut.shell.set_battery_level(int(line))
+ except Exception as err:
+ self.log.info(FAILURE.format(cmd, err))
+
+ """End adb shell test wrappers"""
diff --git a/acts/tests/google/bt/pts/instructions/AVDTP_PTS_INSTUCTIONS b/acts/tests/google/bt/pts/instructions/AVDTP_PTS_INSTUCTIONS
new file mode 100644
index 0000000..8ce453a
--- /dev/null
+++ b/acts/tests/google/bt/pts/instructions/AVDTP_PTS_INSTUCTIONS
@@ -0,0 +1,20 @@
+# Copyright (C) 2018 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.
+
+DID
+=================================================================
+
+TC_SRC_INT_SIG_SMG_BV_23_C
+ #Before test run this and set it to false after the test is done.
+ adb shell setprop bluetooth.pts.force_a2dp_abort true
diff --git a/acts/tests/google/bt/pts/instructions/BNEP_PTS_INSTRUCTIONS b/acts/tests/google/bt/pts/instructions/BNEP_PTS_INSTRUCTIONS
new file mode 100644
index 0000000..389b4ef
--- /dev/null
+++ b/acts/tests/google/bt/pts/instructions/BNEP_PTS_INSTRUCTIONS
@@ -0,0 +1,80 @@
+# Copyright (C) 2018 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.
+
+DID
+=================================================================
+
+TC_CTRL_BV_01_C
+ bta_start_pairing_helper
+
+TC_CTRL_BV_02_C
+ bta_start_pairing_helper
+ bta_connect_profiles
+ # Wait 20 seconds
+
+TC_CTRL_BV_03_C
+ bta_start_pairing_helper
+
+TC_CTRL_BV_04_C
+ bta_start_pairing_helper
+
+TC_CTRL_BV_05_C
+ bta_start_pairing_helper
+
+TC_CTRL_BV_06_C
+ bta_start_pairing_helper
+
+TC_CTRL_BV_08_C
+ bta_start_pairing_helper
+
+TC_CTRL_BV_10_C
+ bta_start_pairing_helper
+
+TC_CTRL_BV_19_C
+ bta_start_pairing_helper
+
+TC_RX-TYPE-0_BV_11_C
+ bta_start_pairing_helper
+ [PTS Interaction]
+
+TC_RX-TYPE-0_BV_15_C
+ bta_start_pairing_helper
+ [PTS Interaction]
+
+TC_RX-TYPE-0_BV_16_C
+ bta_start_pairing_helper
+ [PTS Interaction]
+
+TC_RX-TYPE-0_BV_17_C
+ bta_start_pairing_helper
+ [PTS Interaction]
+
+TC_RX-TYPE-0_BV_18_C
+ bta_start_pairing_helper
+ [PTS Interaction]
+
+TC_RX-C_BV_12_C
+ bta_start_pairing_helper
+ [PTS Interaction]
+
+TC_RX-C-S_BV_13_C
+ bta_start_pairing_helper
+ [PTS Interaction]
+
+TC_RX-C-D_BV_14_C
+ bta_start_pairing_helper
+ [PTS Interaction]
+
+TC_TX-TYPE-0_BV_20_C
+ TBD
\ No newline at end of file
diff --git a/acts/tests/google/bt/pts/instructions/GATT_PTS_INSTRUCTIONS b/acts/tests/google/bt/pts/instructions/GATT_PTS_INSTRUCTIONS
index f789c82..49540b4 100644
--- a/acts/tests/google/bt/pts/instructions/GATT_PTS_INSTRUCTIONS
+++ b/acts/tests/google/bt/pts/instructions/GATT_PTS_INSTRUCTIONS
@@ -14,6 +14,14 @@
GATT
=================================================================
+Note: Bug in PTS forces GATT operations to be over BR/EDR. To run tests over
+LE disable BR/EDR in ICS when running tests.
+
+Note: As of PTS version 7.2.1 GATT server tests now performs scans appropriately. Change IXIT value for TSPX_iut_device_name_in_adv_packet_for_random_address to "CMD LINE Test" and make sure you have a connectable advertisement with that local name:
+ ble_adv_data_include_local_name true
+ ble_start_generic_connectable_advertisement
+
+
TC_CL_GAC_BV_01_C
gattc_connect_over_le
gattc_request_mtu 23
@@ -21,6 +29,14 @@
gattc_write_char_by_instance_id [handle] [size]
gattc_disconnect
+#Potential new instructions for SIG testcase rewrite
+TC_CL_GAC_BV_01_C
+ gattc_connect_over_le
+ gattc_request_mtu 23
+ [Read PTS Handle and Size]
+ gattc_write_invalid_char_by_instance_id [handle] [size]
+ gattc_disconnect
+
TC_CL_GAD_BV_01_C
gattc_connect_over_le
gattc_refresh
@@ -37,6 +53,7 @@
gattc_disconnect
TC_CL_GAD_BV_02_C
+ Note: Values of UUIDS sometimes changes.
gattc_connect_over_le
gattc_discover_service_by_uuid 1800
[PTS Interaction] Verify Values
@@ -68,84 +85,101 @@
TC_CL_GAD_BV_03_C
gattc_connect_over_le
+ gattc_refresh
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
+ gattc_refresh
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
+ gattc_refresh
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
+ gattc_refresh
[PTS Interation]
gattc_disconnect
TC_CL_GAD_BV_04_C
gattc_connect_over_le
- gattc_list_all_uuids
+ gattc_refresh
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
- gattc_list_all_uuids
+ gattc_refresh
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
- gattc_list_all_uuids
+ gattc_refresh
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
- gattc_list_all_uuids
+ gattc_refresh
[PTS Interation]
gattc_disconnect
TC_CL_GAD_BV_05_C
gattc_connect_over_le
+ gattc_refesh
gattc_list_all_uuids
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
+ gttc_refesh
gattc_list_all_uuids
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
+ gattc_refesh
gattc_list_all_uuids
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
+ gattc_refesh
+ gattc_list_all_uuids
+ gattc_refesh
+ [PTS Interation]
+ gattc_disconnect
+ gattc_connect_over_le
+ gattc_refesh
gattc_list_all_uuids
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
+ gattc_refesh
gattc_list_all_uuids
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
+ gattc_refesh
gattc_list_all_uuids
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
- gattc_list_all_uuids
- [PTS Interation]
- gattc_disconnect
- gattc_connect_over_le
+ gattc_refesh
gattc_list_all_uuids
[PTS Interation]
gattc_disconnect
TC_CL_GAD_BV_06_C
gattc_connect_over_le
+ gattc_refresh
gattc_list_all_uuids
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
+ gattc_refresh
gattc_list_all_uuids
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
+ gattc_refresh
gattc_list_all_uuids
[PTS Interation]
gattc_disconnect
gattc_connect_over_le
+ gattc_refresh
gattc_list_all_uuids
[PTS Interation]
gattc_disconnect
@@ -209,6 +243,36 @@
[PTS Interaction]
bta_unbond
+TC_CL_GAR_BV_03_C
+ gattc_connect_over_le
+ gattc_read_char_by_uuid [Input UUID]
+ [PTS Interaction] Verify values
+ gattc_read_char_by_uuid 000055f2-0000-0000-0123-456789abcdef
+
+TC_CL_GAR_BI_06_C
+ gattc_connect_over_le
+ gattc_read_char_by_uuid [Input UUID]
+ gattc_disconnect
+ [PTS Interaction] Verify values
+
+TC_CL_GAR_BI_07_C
+ gattc_connect_over_le
+ gattc_read_char_by_uuid [Input UUID]
+ gattc_disconnect
+ [PTS Interaction] Verify values
+
+TC_CL_GAR_BI_10_C
+ gattc_connect_over_le
+ gattc_read_char_by_uuid [Input UUID]
+ gattc_disconnect
+ [PTS Interaction] Verify values
+
+TC_CL_GAR_BI_11_C
+ gattc_connect_over_le
+ gattc_read_char_by_uuid [Input UUID]
+ gattc_disconnect
+ [PTS Interaction] Verify values
+
TC_CL_GAR_BV_04_C
gattc_connect_over_le
gattc_mod_read_char_by_instance_id [handle]
@@ -426,7 +490,7 @@
TC_CL_GAW_BI_08_C
gattc_connect_over_le
- do_gattc_mod_write_char_by_instance_id [handle] 43
+ gattc_mod_write_char_by_instance_id [handle] 43
[PTS Interaction]
gattc_disconnect
@@ -611,6 +675,15 @@
gattc_execute_reliable_write
gattc_disconnect
+TC_CL_GAW_BI_32_C - Alternate method
+ gattc_connect_over_le
+ gattc_write_invalid_char_by_instance_id 0029 1
+ gattc_write_invalid_char_by_instance_id 0029 21
+ gattc_write_invalid_char_by_instance_id 0029 21
+ gattc_write_invalid_char_by_instance_id 0029 21
+ gattc_disconnect
+
+
TC_CL_GAW_BI_33_C
gattc_connect_over_le
gattc_write_char_by_instance_id [handle] [size]
@@ -641,11 +714,12 @@
[PTS Interaction]
gattc_disconnect
-TC_CL_GAN_BV_01_C
- gatts_setup_database TEST_DB_5
- ble_start_generic_connectable_advertisement
- gatts_list_all_uuids
- gatts_notify_characteristic_changed [instance id from previous command of only characteristic.] true
+TC_CL_GAI_BV_01_C
+ gattc_connect_over_le
+ gattc_enable_indication_desc_by_instance_id 0113
+ [PTS Interaction]
+ gattc_disconnect
+
TC_CL_GAS_BV_01_C
gattc_connect_over_le
@@ -718,7 +792,8 @@
Note: Static Address OK
gatts_setup_database TEST_DB_3
ble_start_generic_connectable_advertisement
- gatts_send_response GATT_INVALID_PDU
+ [PTS Interaction] Enter handle of uuid 2a30 after running the command gatts_list_all_uuids
+ gatts_send_response GATT_FAILURE
TC_SR_GAR_BI_03_C
gatts_setup_database TEST_DB_1
@@ -752,8 +827,11 @@
TC_SR_GAR_BI_08_C
Note: Static Address OK
+ bta_start_pairing_helper
gatts_setup_database TEST_DB_1
ble_start_generic_connectable_advertisement
+ [PTS Interaction] Enter pin from PTS to phone
+ gatts_send_response GATT_FAILURE
TC_SR_GAR_BI_09_C
gatts_setup_database TEST_DB_1
@@ -795,6 +873,14 @@
gatts_send_response GATT_SUCCESS 24
gatts_send_response GATT_INVALID_OFFSET
+TC_SR_GAR_BI_13_C (on PTS 7.2.1)
+ gatts_setup_database LARGE_1
+ # Have a logcat going as such: adb logcat | grep GattServer11onCharacteristicReadRequest
+ # Every time the offset is > 23:
+ gatts_send_response GATT_INVALID_OFFSET
+ # Every time the offset is < 23:
+ gatts_send_response GATT_SUCCESS 24
+
TC_SR_GAR_BI_14_C
Note: Static Address OK
gatts_setup_database TEST_DB_3
@@ -817,13 +903,23 @@
gatts_send_response GATT_0C_ERR
gatts_send_response GATT_0C_ERR
-TC_SR_GAR_BV_05_C
+TC_SR_GAR_BV_05_C - Deprecated
Note: Static Address OK
gatts_setup_database TEST_DB_3
ble_start_generic_connectable_advertisement
[PTS Interaction] Enter 002a
gatts_send_response GATT_READ_NOT_PERMITTED
+TC_SR_GAR_BV_05_C - PTS 7.2.1
+ gatts_setup_database TEST_DB_3
+ bta_start_pairing_helper
+ ble_adv_data_include_local_name true
+ ble_start_generic_connectable_advertisement
+ [PTS Interaction] Enter PIN fro PTS to phone
+ gatts_send_response GATT_SUCCESS
+ gatts_send_response GATT_SUCCESS
+ [PTS Interaction] Verify values
+
TC_SR_GAR_BI_18_C
Note: Static Address OK
gatts_setup_database TEST_DB_3
@@ -995,6 +1091,8 @@
gatts_send_response GATT_SUCCESS
gatts_send_response GATT_SUCCESS
gatts_send_response GATT_SUCCESS
+ atts_send_response GATT_SUCCESS
+ atts_send_response GATT_SUCCESS
TC_SR_GAW_BI_02_C
Note: Static Address OK
@@ -1045,7 +1143,7 @@
[PTS Interaction] Enter 002a
gatts_send_response GATT_WRITE_NOT_PERMITTED
-TC_SR_GAW_BI_09_C
+TC_SR_GAW_BI_09_C - Deprecated
Note: Static Address OK
gatts_setup_database PTS_TEST
ble_start_generic_connectable_advertisement
@@ -1055,6 +1153,20 @@
gatts_send_response GATT_SUCCESS
gatts_send_response GATT_INVALID_OFFSET
+TC_SR_GAW_BI_09_C - NEW INSTRUCTIONS
+ Note: This test is a little tricky. Need to monitor logcat as such: adb logcat | grep offset.
+ gatts_setup_database LARGE_DB_3
+ ble_adv_data_include_local_name true
+ ble_start_generic_connectable_advertisement
+ # Whenever offset is <=23
+ gatts_send_response GATT_SUCCESS 24
+ # Whenever offset is > 23
+ # If preparedWrite value is True
+ gatts_send_response GATT_SUCCESS 24
+ gatts_send_response GATT_INVALID_OFFSET
+ # If preparedWrite value is False
+ gatts_send_response GATT_INVALID_OFFSET
+
TC_SR_GAW_BI_11_C
Note: Static Address OK
gatts_setup_database TEST_DB_3
@@ -1086,9 +1198,9 @@
TC_SR_GAW_BV_10_C
Note: Static Address OK
Note: Make sure MTU is set to 23 on PTS
- gatts_setup_database PTS_TEST
+ gatts_setup_database LARGE_DB_3
ble_start_generic_connectable_advertisement
- Whenever PTS prompts: "Discover All Characteristics of Service Request completed successfully" run this cmd:
+ Whenever PTS prompts: "Discover All Characteristics of Service Request completed successfully" in the Output Tool Window run this cmd:
gatts_send_response GATT_SUCCESS 24
Otherwise always respond with:
gatts_send_response GATT_SUCCESS
@@ -1184,7 +1296,7 @@
gatts_send_response GATT_0C_ERR
gatts_send_response GATT_0C_ERR
-TC_SR_GAW_BV_09_C
+TC_SR_GAW_BV_09_C - DEPRECATED
Note: Static Address OK
gatts_setup_database TEST_DB_3
ble_start_generic_connectable_advertisement
@@ -1198,6 +1310,13 @@
gatts_send_response GATT_SUCCESS
gatts_send_response GATT_SUCCESS
+TC_SR_GAW_BV_09_C - New Instructions
+ Note: Static Address OK
+ gatts_setup_database LARGE_DB_1
+ ble_start_generic_connectable_advertisement
+ # Repeat below cmd until success
+ gatts_send_response GATT_SUCCESS
+
TC_SR_GAW_BI_25_C
Note: Static Address OK
gatts_setup_database TEST_DB_3
@@ -1263,6 +1382,7 @@
TC_SR_GAW_BI_33_C
Note: Static Address OK
+ Note: This testcase is tricky as the order randomises a bit each time...
gatts_setup_database LARGE_DB_3
ble_start_generic_connectable_advertisement
gatts_send_response GATT_SUCCESS 24
@@ -1307,18 +1427,22 @@
gatts_notify_characteristic_changed [Handle from PTS] false 10
TC_SR_GAI_BV_01_C
- gattc_connect_over_le
- gattc_write_desc_notification_by_instance_id 00f3 2
+ gatts_setup_database DB_TEST
+ ble_adv_data_include_local_name true
+ ble_start_generic_connectable_advertisement
[PTS Interaction] Verify value and click yes
- gattc_disconnect
+ gatts_notify_characteristic_changed 002a false 10
+ gatts_notify_characteristic_changed 002a true 10
+
TC_SR_GAS_BV_01_C
gatts_setup_database DB_TEST
ble_start_generic_connectable_advertisement
+ [PTS Interaction] Click ok
+ gatts_notify_characteristic_changed 002a false 10
gatts_setup_database TEST_DB_3
- Wait 60 seconds for PTS to process
-TC_SR_GAT_BV_01_C
+TC_SR_GAT_BV_01_C - Deprecated
gatts_setup_database LARGE_DB_3
ble_start_generic_connectable_advertisement
gatts_send_response GATT_SUCCESS
@@ -1326,3 +1450,22 @@
gatts_send_response GATT_SUCCESS
gatts_setup_database TEST_DB_3
Wait 30 seconds for PTS to process
+
+TC_SR_GAT_BV_01_C - Deprecated
+ gatts_setup_database Test_DB_5
+ ble_adv_data_include_local_name true
+ ble_start_generic_connectable_advertisement
+ gatts_notify_characteristic_changed 0013 true
+ gatts_notify_characteristic_changed 002a true
+ gatts_send_response GATT_SUCCESS
+ gatts_send_response GATT_SUCCESS
+ gatts_send_response GATT_SUCCESS
+ gatts_notify_characteristic_changed 0003 true
+ gatts_setup_database TEST_DB_1
+ [PTS WAIT] 30 seconds to timeout to occur
+
+TC_SR_UNS_BI_02_C
+ bta_start_pairing_helper
+ ble_adv_data_include_local_name true
+ ble_start_generic_connectable_advertisement
+ [PTS Interaction] Enter pin from PTS to phone
\ No newline at end of file
diff --git a/acts/tests/google/bt/pts/instructions/L2CAP_PTS_INSTRUCTIONS b/acts/tests/google/bt/pts/instructions/L2CAP_PTS_INSTRUCTIONS
index a865f45..d41872f 100644
--- a/acts/tests/google/bt/pts/instructions/L2CAP_PTS_INSTRUCTIONS
+++ b/acts/tests/google/bt/pts/instructions/L2CAP_PTS_INSTRUCTIONS
@@ -101,6 +101,49 @@
TC_L2CAP_COS_ECH_BV_01_C
bta_enable
+TC_L2CP_COS_CFC_BV_01_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ rfcomm_write 10
+ [PTS Interaction] Verify value
+ rfcomm_stop
+ gattc_disconnect
+ bta_disable
+ bta_enable
+
+TC_L2CP_COS_CFC_BV_02_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ rfcomm_write 10
+ [PTS Interaction] Verify value
+ rfcomm_stop
+ gattc_disconnect
+
+TC_L2CP_COS_CFC_BV_03_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ rfcomm_write 10
+ [PTS Interaction] Verify value
+ rfcomm_stop
+ gattc_disconnect
+
+TC_L2CP_COS_CFC_BV_04_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ [PTS Interaction] Verify value
+ rfcomm_stop
+ gattc_disconnect
+
+TC_L2CP_COS_CFC_BV_05_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ [PTS Interaction] Verify value
+ rfcomm_stop
+ gattc_disconnect
+ bta_disable
+ bta_enable
+
TC_L2CAP_EXF_BV_01_C
bta_set_scan_mode connectable
@@ -119,6 +162,10 @@
TC_L2CAP_CMC_BV_10_C
rfcomm_connect
rfcomm_stop
+ rfcomm_connect
+ rfcomm_stop
+ rfcomm_connect
+ rfcomm_stop
[Wait up to 10-15 seconds]
TC_L2CAP_CMC_BV_11_C
@@ -138,7 +185,7 @@
[PTS Interaction] Yes
TC_L2CAP_CMC_BI_05_C
- rfcomm_accept
+ rfcomm_connect
[Wait up to 10-15 seconds]
TC_L2CAP_CMC_BI_06_C
@@ -161,9 +208,80 @@
gattc_connect_over_le
gattc_disconnect
+TC_L2CAP_LE_CFC_BI_01_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ [PTS Interaction] Verify values
+ gattc_disconnect
+
+TC_L2CAP_LE_CFC_BV_01_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ [PTS Interaction] Verify values
+ gattc_disconnect
+
+TC_L2CAP_LE_CFC_BV_02_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ [PTS Interaction] Verify values
+ gattc_disconnect
+
+TC_L2CAP_LE_CFC_BV_03_C
+ TBD
+
+TC_L2CAP_LE_CFC_BV_04_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 241
+ [PTS Interaction] Verify values
+ gattc_disconnect
+
TC_L2CAP_LE_CFC_BV_05_C
gattc_connect_over_le
gattc_disconnect
bta_disable
bta_enable
gattc_connect_over_le
+
+TC_L2CAP_LE_CFC_BV_06_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ rfcomm_write 10
+ rfcomm_write 10
+ rfcomm_write 10
+ gattc_disconnect
+ bta_disable
+ bta_enable
+
+TC_L2CAP_LE_CFC_BV_07_C
+ TBD
+
+TC_L2CAP_LE_CFC_BV_09_C
+ TBD
+
+TC_L2CAP_LE_CFC_BV_16_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ [PTS Interaction] Verify values
+ gattc_disconnect
+
+TC_L2CAP_LE_CFC_BV_18_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ [PTS Interaction] Verify values
+ gattc_disconnect
+
+TC_L2CAP_LE_CFC_BV_19_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ [PTS Interaction] Verify values
+ gattc_disconnect
+
+TC_L2CAP_LE_CFC_BV_20_C
+ TBD
+
+TC_L2CAP_LE_CFC_BV_21_C
+ gattc_connect_over_le
+ gattc_socket_conn_begin_connect_thread_psm 1 0 1
+ [PTS Interaction] Verify values
+ gattc_disconnect
+
diff --git a/acts/tests/google/bt/system_tests/BtStressTest.py b/acts/tests/google/bt/system_tests/BtStressTest.py
index e756c9c..8e87afc 100644
--- a/acts/tests/google/bt/system_tests/BtStressTest.py
+++ b/acts/tests/google/bt/system_tests/BtStressTest.py
@@ -21,6 +21,8 @@
from acts.base_test import BaseTestClass
from acts.test_decorators import test_tracker_info
from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
+from acts.test_utils.bt.bt_constants import bluetooth_off
+from acts.test_utils.bt.bt_constants import bluetooth_on
from acts.test_utils.bt.bt_test_utils import clear_bonded_devices
from acts.test_utils.bt.bt_test_utils import pair_pri_to_sec
from acts.test_utils.bt.bt_test_utils import reset_bluetooth
@@ -61,10 +63,24 @@
TAGS: Classic, Stress
Priority: 1
"""
+ dut = self.android_devices[0]
for n in range(self.iterations):
self.log.info("Toggling bluetooth iteration {}.".format(n + 1))
- if not reset_bluetooth([self.android_devices[0]]):
- self.log.error("Failure to reset Bluetooth")
+ dut.ed.clear_all_events()
+ try:
+ dut.droid.bluetoothToggleState(False)
+ dut.ed.pop_event(bluetooth_off, self.default_timeout)
+ except Exception as err:
+ dut.log.error(
+ "Failed to toggle off Bluetooth with error: {}".format(
+ err))
+ return False
+ try:
+ dut.droid.bluetoothToggleState(True)
+ dut.ed.pop_event(bluetooth_on, self.default_timeout)
+ except Exception as err:
+ dut.log.error(
+ "Failed to toggle on Bluetooth with error: {}".format(err))
return False
return True
@@ -115,7 +131,7 @@
time.sleep(2)
bonded_devices = ad.droid.bluetoothGetBondedDevices()
if len(bonded_devices) > 0:
- self.log.error("Failed to unbond devices: {}".format(
- bonded_devices))
+ self.log.error(
+ "Failed to unbond devices: {}".format(bonded_devices))
return False
return True
diff --git a/acts/tests/google/wifi/WifiStressTest.py b/acts/tests/google/wifi/WifiStressTest.py
index 0f17c19..d891cc2 100755
--- a/acts/tests/google/wifi/WifiStressTest.py
+++ b/acts/tests/google/wifi/WifiStressTest.py
@@ -149,6 +149,8 @@
wutils.wifi_toggle_state(self.dut, True)
startup_time = time.time() - startTime
self.log.debug("WiFi was enabled on the device in %s s." % startup_time)
+ raise signals.TestPass(details="", extras={"Iterations":"%d" %
+ (count+1)})
@test_tracker_info(uuid="49e3916a-9580-4bf7-a60d-a0f2545dcdde")
def test_stress_connect_traffic_disconnect_5g(self):
@@ -162,19 +164,25 @@
"""
for count in range(self.stress_count):
- net_id = self.dut.droid.wifiAddNetwork(self.wpa_5g)
- asserts.assert_true(net_id != -1, "Add network %r failed" % self.wpa_5g)
- self.scan_and_connect_by_id(self.wpa_5g, net_id)
- # Start IPerf traffic from phone to server.
- # Upload data for 10s.
- args = "-p {} -t {}".format(self.iperf_server.port, 10)
- self.log.info("Running iperf client {}".format(args))
- result, data = self.dut.run_iperf_client(self.iperf_server_address, args)
- if not result:
- self.log.debug("Error occurred in iPerf traffic.")
- self.run_ping(10)
- wutils.wifi_forget_network(self.dut,self.wpa_5g[WifiEnums.SSID_KEY])
- time.sleep(WAIT_BEFORE_CONNECTION)
+ try:
+ net_id = self.dut.droid.wifiAddNetwork(self.wpa_5g)
+ asserts.assert_true(net_id != -1, "Add network %r failed" % self.wpa_5g)
+ self.scan_and_connect_by_id(self.wpa_5g, net_id)
+ # Start IPerf traffic from phone to server.
+ # Upload data for 10s.
+ args = "-p {} -t {}".format(self.iperf_server.port, 10)
+ self.log.info("Running iperf client {}".format(args))
+ result, data = self.dut.run_iperf_client(self.iperf_server_address, args)
+ if not result:
+ self.log.debug("Error occurred in iPerf traffic.")
+ self.run_ping(10)
+ wutils.wifi_forget_network(self.dut,self.wpa_5g[WifiEnums.SSID_KEY])
+ time.sleep(WAIT_BEFORE_CONNECTION)
+ except:
+ raise signals.TestFailure("Network connect-disconnect failed."
+ "Look at logs", extras={"Iterations":"%d" %((count+1))})
+ raise signals.TestPass(details="", extras={"Iterations":"%d" %
+ (count+1)})
@test_tracker_info(uuid="e9827dff-0755-43ec-8b50-1f9756958460")
def test_stress_connect_long_traffic_5g(self):
@@ -186,17 +194,23 @@
3. Verify no WiFi disconnects/data interruption.
"""
- self.scan_and_connect_by_ssid(self.wpa_5g)
- # Start IPerf traffic from server to phone.
- # Download data for 5 hours.
- sec = 5*60*60
- args = "-p {} -t {} -R".format(self.iperf_server.port, sec)
- self.log.info("Running iperf client {}".format(args))
- result, data = self.dut.run_iperf_client(self.iperf_server_address,
- args, timeout=sec+1)
- if not result:
- self.log.debug("Error occurred in iPerf traffic.")
- self.run_ping(sec)
+ try:
+ self.scan_and_connect_by_ssid(self.wpa_5g)
+ # Start IPerf traffic from server to phone.
+ # Download data for 5 hours.
+ #sec = 5*60*60
+ sec = 60
+ args = "-p {} -t {} -R".format(self.iperf_server.port, sec)
+ self.log.info("Running iperf client {}".format(args))
+ result, data = self.dut.run_iperf_client(self.iperf_server_address,
+ args, timeout=sec+1)
+ if not result:
+ self.log.debug("Error occurred in iPerf traffic.")
+ self.run_ping(sec)
+ except:
+ raise signals.TestFailure("Network long-connect failed."
+ "Look at logs", extras={"Seconds":"UNKNOWN"})
+ raise signals.TestPass(details="", extras={"Seconds":"%d" %sec})
@test_tracker_info(uuid="d367c83e-5b00-4028-9ed8-f7b875997d13")
def test_stress_wifi_failover(self):
@@ -238,7 +252,10 @@
self.log.info("Network Config = %s" % network_config)
if len(network_config):
raise signals.TestFailure("All the network configurations were not "
- "removed. Configured networks = %s" % network_config)
+ "removed. Configured networks = %s" % network_config,
+ extras={"Iterations":"%d" %((count+1)*4)})
+ raise signals.TestPass(details="", extras={"Iterations":"%d" %
+ ((count+1)*4)})
@test_tracker_info(uuid="2c19e8d1-ac16-4d7e-b309-795144e6b956")
def test_stress_softAP_startup_and_stop_5g(self):
@@ -273,8 +290,10 @@
time.sleep(TIMEOUT)
cur_wifi_state = self.dut.droid.wifiCheckState()
if initial_wifi_state != cur_wifi_state:
- raise signals.TestFailure("Wifi state was %d before softAP and %d now!" %
- (initial_wifi_state, cur_wifi_state))
+ raise signals.TestFailure("Wifi state was %d before softAP and %d now!" %
+ (initial_wifi_state, cur_wifi_state),
+ extras={"Iterations":"%d" %(count+1)})
+ raise signals.TestPass(details="", extras={"Iterations":"%d" %(count+1)})
@test_tracker_info(uuid="eb22e26b-95d1-4580-8c76-85dfe6a42a0f")
def test_stress_wifi_roaming(self):
@@ -286,10 +305,15 @@
for count in range(int(self.stress_count/2)):
self.log.info("Roaming iteration %d, from %s to %s", count,
AP1_network, AP2_network)
- wutils.trigger_roaming_and_validate(self.dut, self.attenuators,
- "AP1_off_AP2_on", AP2_network)
- self.log.info("Roaming iteration %d, from %s to %s", count,
- AP2_network, AP1_network)
- wutils.trigger_roaming_and_validate(self.dut, self.attenuators,
- "AP1_on_AP2_off", AP1_network)
+ try:
+ wutils.trigger_roaming_and_validate(self.dut, self.attenuators,
+ "AP1_off_AP2_on", AP2_network)
+ self.log.info("Roaming iteration %d, from %s to %s", count,
+ AP2_network, AP1_network)
+ wutils.trigger_roaming_and_validate(self.dut, self.attenuators,
+ "AP1_on_AP2_off", AP1_network)
+ except:
+ raise signals.TestFailure("Roaming failed. Look at logs",
+ extras={"Iterations":"%d" %((count+1)*2)})
+ raise signals.TestPass(details="", extras={"Iterations":"%d" %((count+1)*2)})