Add E2E tests for Bluetooth carkit testing
Bug: 73728298
Test: Manual and UPF
Change-Id: I19c56c8b5e006d2d50c914e2c89fd29d1964c016
(cherry picked from commit 1102acf41d454400c3a49ab533d197da762f4ffb)
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_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"""