Merge changes Ib593cdbf,Iacad78dd,Iaa69c5ee,I8d86aa76,I19c56c8b, ... into pi-dev
* changes:
Add new tools for GATT and L2CAP.
Add new instructions for AVDTP and BNEP
Change access of libraries to Android bt libs
Add special attributes for Bluetooth libs
Add E2E tests for Bluetooth carkit testing
Change imports for ble_lib and add new function
Add new library for common adb shell commands.
Fix GattToolTest
Change gattc_lib to use common nomenclature
Return scan_callback when looking for generic addr
Change rfcomm_lib to use common nomenclature
Change bta_lib to use common nomenclature
Add additional instructions to GATT.
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/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
+