1#!/usr/bin/env python3
2#
3# Copyright (C) 2018 The Android Open Source Project
4#
5# Licensed under the Apache License, Version 2.0 (the "License"); you may not
6# use this file except in compliance with the License. You may obtain a copy of
7# the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14# License for the specific language governing permissions and limitations under
15# the License.
16
17import time
18import os
19
20from acts.keys import Config
21from acts.utils import rand_ascii_str
22from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
23from acts.test_utils.bt.bt_constants import logcat_strings
24from acts.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
25from acts.test_utils.tel.tel_defines import AUDIO_ROUTE_BLUETOOTH
26from acts.test_utils.tel.tel_defines import AUDIO_ROUTE_EARPIECE
27from acts.test_utils.tel.tel_defines import AUDIO_ROUTE_SPEAKER
28from acts.test_utils.tel.tel_defines import WAIT_TIME_IN_CALL
29from acts.test_utils.tel.tel_test_utils import get_phone_number
30from acts.test_utils.tel.tel_test_utils import hangup_call
31from acts.test_utils.tel.tel_test_utils import initiate_call
32from acts.test_utils.tel.tel_test_utils import num_active_calls
33from acts.test_utils.tel.tel_test_utils import sms_send_receive_verify
34from acts.test_utils.tel.tel_test_utils import wait_and_answer_call
35from acts.test_utils.tel.tel_voice_utils import get_audio_route
36from acts.test_utils.tel.tel_voice_utils import set_audio_route
37from acts.test_utils.tel.tel_voice_utils import swap_calls
38from acts.test_utils.tel.tel_test_utils import toggle_airplane_mode
39from acts.test_utils.tel.tel_test_utils import call_setup_teardown
40from acts.utils import exe_cmd
41from acts.utils import get_current_epoch_time
42
43KEYCODE_VOLUME_UP = "input keyevent 24"
44KEYCODE_VOLUME_DOWN = "input keyevent 25"
45KEYCODE_EVENT_PLAY_PAUSE = "input keyevent 85"
46KEYCODE_MEDIA_STOP = "input keyevent 86"
47KEYCODE_EVENT_NEXT = "input keyevent 87"
48KEYCODE_EVENT_PREVIOUS = "input keyevent 88"
49KEYCODE_MEDIA_REWIND = "input keyevent 89"
50KEYCODE_MEDIA_FAST_FORWARD = "input keyevent 90"
51KEYCODE_MUTE = "input keyevent 91"
52
53default_timeout = 10
54
55
56class E2eBtCarkitLib():
57
58    android_devices = []
59    short_timeout = 3
60    active_call_id = None
61    hold_call_id = None
62    log = None
63    mac_address = None
64
65    def __init__(self, log, target_mac_address=None):
66        self.log = log
67        self.target_mac_address = target_mac_address
68
69    def connect_hsp_helper(self, ad):
70        end_time = time.time() + default_timeout + 10
71        connected_hsp_devices = len(ad.droid.bluetoothHspGetConnectedDevices())
72        while connected_hsp_devices != 1 and time.time() < end_time:
73            try:
74                ad.droid.bluetoothHspConnect(self.target_mac_address)
75                time.sleep(3)
76                if len(ad.droid.bluetoothHspGetConnectedDevices() == 1):
77                    break
78            except Exception:
79                self.log.debug("Failed to connect hsp trying again...")
80            try:
81                ad.droid.bluetoothConnectBonded(self.target_mac_address)
82            except Exception:
83                self.log.info("Failed to connect to bonded device...")
84            connected_hsp_devices = len(
85                ad.droid.bluetoothHspGetConnectedDevices())
86        if connected_hsp_devices != 1:
87            self.log.error("Failed to reconnect to HSP service...")
88            return False
89        self.log.info("Connected to HSP service...")
90        return True
91
92    def setup_multi_call(self, caller0, caller1, callee):
93        outgoing_num = get_phone_number(self.log, callee)
94        if not initiate_call(self.log, caller0, outgoing_num):
95            self.log.error("Failed to initiate call")
96            return False
97        if not wait_and_answer_call(self.log, callee):
98            self.log.error("Failed to answer call.")
99            return False
100        time.sleep(self.short_timeout)
101        if not initiate_call(self.log, caller1, outgoing_num):
102            self.log.error("Failed to initiate call")
103            return False
104        if not wait_and_answer_call(self.log, callee):
105            self.log.error("Failed to answer call.")
106            return False
107        return True
108
109    def process_tests(self, tests):
110        for test in tests:
111            try:
112                test()
113            except Exception as err:
114                self.log.error(err)
115
116    def run_suite_hfp_tests(self):
117        tests = [
118            self.outgoing_call_private_number,
119            self.outgoing_call_unknown_contact,
120            self.incomming_call_private_number,
121            self.incomming_call_unknown_contact,
122            self.outgoing_call_multiple_iterations,
123            self.outgoing_call_hsp_disabled_then_enabled_during_call,
124            self.call_audio_routes,
125            self.sms_during_incomming_call,
126            self.multi_incomming_call,
127            self.multi_call_audio_routing,
128            self.multi_call_swap_multiple_times,
129            self.outgoing_call_a2dp_play_before_and_after,
130        ]
131        _process_tests(tests)
132
133    def run_suite_hfp_conf_tests(self):
134        tests = [
135            self.multi_call_join_conference_call,
136            self.multi_call_join_conference_call_hangup_conf_call,
137            self.outgoing_multi_call_join_conference_call,
138            self.multi_call_join_conference_call_audio_routes,
139        ]
140        _process_tests(tests)
141
142    def run_suite_map_tests(self):
143        tests = [
144            self.sms_receive_different_sizes,
145            self.sms_receive_multiple,
146            self.sms_send_outgoing_texts,
147        ]
148        _process_tests(tests)
149
150    def run_suite_avrcp_tests(self):
151        tests = [
152            self.avrcp_play_pause,
153            self.avrcp_next_previous_song,
154            self.avrcp_next_previous,
155            self.avrcp_next_repetative,
156        ]
157        _process_tests(tests)
158
159    def disconnect_reconnect_multiple_iterations(self, pri_dut):
160        iteration_count = 5
161        self.log.info(
162            "Test disconnect-reconnect scenario from phone {} times.".format(
163                iteration_count))
164        self.log.info(
165            "This test will prompt for user interaction after each reconnect.")
166        input("Press enter to execute this testcase...")
167        #Assumes only one devices connected
168        grace_timeout = 4  #disconnect and reconnect timeout
169        for n in range(iteration_count):
170            self.log.info("Test iteration {}.".format(n + 1))
171            self.log.info("Disconnecting device {}...".format(
172                self.target_mac_address))
173            pri_dut.droid.bluetoothDisconnectConnected(self.target_mac_address)
174            # May have to do a longer sleep for carkits.... need to test
175            time.sleep(grace_timeout)
176            self.log.info("Connecting device {}...".format(
177                self.target_mac_address))
178            pri_dut.droid.bluetoothConnectBonded(self.target_mac_address)
179            if not self.connect_hsp_helper(pri_dut):
180                return False
181            start_time = time.time()
182            connected_devices = pri_dut.droid.bluetoothGetConnectedDevices()
183            self.log.info(
184                "Waiting up to 10 seconds for device to reconnect...")
185            while time.time() < start_time + 10 and len(connected_devices) != 1:
186                connected_devices = pri_dut.droid.bluetoothGetConnectedDevices(
187                )
188                time.sleep(1)
189            if len(connected_devices) != 1:
190                self.log.error(
191                    "Failed to reconnect at iteration {}... continuing".format(
192                        n))
193                return False
194            input("Continue to next iteration?")
195        return True
196
197    def disconnect_a2dp_only_then_reconnect(self, pri_dut):
198        self.log.info(
199            "Test disconnect-reconnect a2dp only scenario from phone.")
200        input("Press enter to execute this testcase...")
201        if not pri_dut.droid.bluetoothA2dpDisconnect(self.target_mac_address):
202            self.log.error("Failed to disconnect A2DP service...")
203            return False
204        time.sleep(self.short_timeout)
205        result = input("Confirm A2DP disconnected? (Y/n) ")
206        if result == "n":
207            self.log.error(
208                "Tester confirmed that A2DP did not disconnect. Failing test.")
209            return False
210        if len(pri_dut.droid.bluetoothA2dpGetConnectedDevices()) != 0:
211            self.log.error("Failed to disconnect from A2DP service")
212            return False
213        pri_dut.droid.bluetoothA2dpConnect(self.target_mac_address)
214        time.sleep(self.short_timeout)
215        if len(pri_dut.droid.bluetoothA2dpGetConnectedDevices()) != 1:
216            self.log.error("Failed to reconnect to A2DP service...")
217            return False
218        return True
219
220    def disconnect_hsp_only_then_reconnect(self, pri_dut):
221        self.log.info(
222            "Test disconnect-reconnect hsp only scenario from phone.")
223        input("Press enter to execute this testcase...")
224        if not pri_dut.droid.bluetoothHspDisconnect(self.target_mac_address):
225            self.log.error("Failed to disconnect HSP service...")
226            return False
227        time.sleep(self.short_timeout)
228        result = input("Confirm HFP disconnected? (Y/n) ")
229        pri_dut.droid.bluetoothHspConnect(self.target_mac_address)
230        time.sleep(self.short_timeout)
231        if len(pri_dut.droid.bluetoothHspGetConnectedDevices()) != 1:
232            self.log.error("Failed to connect from HSP service")
233            return False
234        return True
235
236    def disconnect_both_hsp_and_a2dp_then_reconnect(self, pri_dut):
237        self.log.info(
238            "Test disconnect-reconnect hsp and a2dp scenario from phone.")
239        input("Press enter to execute this testcase...")
240        if not pri_dut.droid.bluetoothA2dpDisconnect(self.target_mac_address):
241            self.log.error("Failed to disconnect A2DP service...")
242            return False
243        if not pri_dut.droid.bluetoothHspDisconnect(self.target_mac_address):
244            self.log.error("Failed to disconnect HSP service...")
245            return False
246        time.sleep(self.short_timeout)
247        if len(pri_dut.droid.bluetoothA2dpGetConnectedDevices()) != 0:
248            self.log.error("Failed to disconnect from A2DP service")
249            return False
250        if len(pri_dut.droid.bluetoothHspGetConnectedDevices()) != 0:
251            self.log.error("Failed to disconnect from HSP service")
252            return False
253        result = input("Confirm HFP and A2DP disconnected? (Y/n) ")
254        pri_dut.droid.bluetoothConnectBonded(self.target_mac_address)
255        time.sleep(self.short_timeout)
256        if len(pri_dut.droid.bluetoothA2dpGetConnectedDevices()) != 1:
257            self.log.error("Failed to reconnect to A2DP service...")
258            return False
259        if not self.connect_hsp_helper(pri_dut):
260            return False
261        return True
262
263    def outgoing_call_private_number(self, pri_dut, ter_dut):
264        self.log.info(
265            "Test outgoing call scenario from phone to private number")
266        input("Press enter to execute this testcase...")
267        outgoing_num = "*67" + get_phone_number(self.log, ter_dut)
268        if not initiate_call(self.log, pri_dut, outgoing_num):
269            self.log.error("Failed to initiate call")
270            return False
271        if not wait_and_answer_call(self.log, ter_dut):
272            self.log.error("Failed to answer call.")
273            return False
274        time.sleep(self.short_timeout)
275        input("Press enter to hangup call...")
276        if not hangup_call(self.log, pri_dut):
277            self.log.error("Failed to hangup call")
278            return False
279        return True
280
281    def outgoing_call_a2dp_play_before_and_after(self, pri_dut, sec_dut):
282        self.log.info(
283            "Test outgoing call scenario while playing music. Music should resume after call."
284        )
285        pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
286        input(
287            "Press enter to execute this testcase when music is in a play state..."
288        )
289        outgoing_num = get_phone_number(self.log, sec_dut)
290        if not initiate_call(self.log, pri_dut, outgoing_num):
291            self.log.error("Failed to initiate call")
292            return False
293        if not wait_and_answer_call(self.log, sec_dut):
294            self.log.error("Failed to answer call.")
295            return False
296        time.sleep(self.short_timeout)
297        input("Press enter to hangup call...")
298        if not hangup_call(self.log, pri_dut):
299            self.log.error("Failed to hangup call")
300            return False
301        input("Press enter when music continues to play.")
302        self.log.info("Pausing Music...")
303        pri_dut.adb.shell(KEYCODE_EVENT_PLAY_PAUSE)
304        return True
305
306    def outgoing_call_unknown_contact(self, pri_dut, ter_dut):
307        self.log.info(
308            "Test outgoing call scenario from phone to unknow contact")
309        input("Press enter to execute this testcase...")
310        outgoing_num = get_phone_number(self.log, ter_dut)
311        if not initiate_call(self.log, pri_dut, outgoing_num):
312            self.log.error("Failed to initiate call")
313            return False
314        if not wait_and_answer_call(self.log, ter_dut):
315            self.log.error("Failed to answer call.")
316            return False
317        time.sleep(self.short_timeout)
318        input("Press enter to hangup call...")
319        if not hangup_call(self.log, pri_dut):
320            self.log.error("Failed to hangup call")
321            return False
322        return True
323
324    def incomming_call_private_number(self, pri_dut, ter_dut):
325        self.log.info(
326            "Test incomming call scenario to phone from private number")
327        input("Press enter to execute this testcase...")
328        outgoing_num = "*67" + get_phone_number(self.log, pri_dut)
329        if not initiate_call(self.log, ter_dut, outgoing_num):
330            self.log.error("Failed to initiate call")
331            return False
332        if not wait_and_answer_call(self.log, pri_dut):
333            self.log.error("Failed to answer call.")
334            return False
335        time.sleep(self.short_timeout)
336        input("Press enter to hangup call...")
337        if not hangup_call(self.log, ter_dut):
338            self.log.error("Failed to hangup call")
339            return False
340
341        return True
342
343    def incomming_call_unknown_contact(self, pri_dut, ter_dut):
344        self.log.info(
345            "Test incomming call scenario to phone from unknown contact")
346        input("Press enter to execute this testcase...")
347        outgoing_num = get_phone_number(self.log, pri_dut)
348        if not initiate_call(self.log, ter_dut, outgoing_num):
349            self.log.error("Failed to initiate call")
350            return False
351        if not wait_and_answer_call(self.log, pri_dut):
352            self.log.error("Failed to answer call.")
353            return False
354        time.sleep(self.short_timeout)
355        input("Press enter to hangup call...")
356        if not hangup_call(self.log, ter_dut):
357            self.log.error("Failed to hangup call")
358            return False
359        return True
360
361    def outgoing_call_multiple_iterations(self, pri_dut, sec_dut):
362        iteration_count = 3
363        self.log.info(
364            "Test outgoing call scenario from phone {} times from known contact".
365            format(iteration_count))
366        input("Press enter to execute this testcase...")
367        outgoing_num = get_phone_number(self.log, sec_dut)
368        for _ in range(iteration_count):
369            if not initiate_call(self.log, pri_dut, outgoing_num):
370                self.log.error("Failed to initiate call")
371                return False
372            if not wait_and_answer_call(self.log, sec_dut):
373                self.log.error("Failed to answer call.")
374                return False
375            time.sleep(self.short_timeout)
376            if not hangup_call(self.log, pri_dut):
377                self.log.error("Failed to hangup call")
378                return False
379        return True
380
381    def outgoing_call_hsp_disabled_then_enabled_during_call(
382            self, pri_dut, sec_dut):
383        self.log.info(
384            "Test outgoing call hsp disabled then enable during call.")
385        input("Press enter to execute this testcase...")
386        outgoing_num = get_phone_number(self.log, sec_dut)
387        if not pri_dut.droid.bluetoothHspDisconnect(self.target_mac_address):
388            self.log.error("Failed to disconnect HSP service...")
389            return False
390        time.sleep(self.short_timeout)
391        if len(pri_dut.droid.bluetoothHspGetConnectedDevices()) != 0:
392            self.log.error("Failed to disconnect from HSP service")
393            return False
394        if not initiate_call(self.log, pri_dut, outgoing_num):
395            self.log.error("Failed to initiate call")
396            return False
397        time.sleep(default_timeout)
398        pri_dut.droid.bluetoothConnectBonded(self.target_mac_address)
399        time.sleep(self.short_timeout)
400        test_result = True
401        if len(pri_dut.droid.bluetoothHspGetConnectedDevices()) != 1:
402            self.log.error("Failed to reconnect to HSP service...")
403            return
404        if not hangup_call(self.log, pri_dut):
405            self.log.error("Failed to hangup call")
406            return False
407        return test_result
408
409    def call_audio_routes(self, pri_dut, sec_dut):
410        self.log.info("Test various audio routes scenario from phone.")
411        input("Press enter to execute this testcase...")
412        outgoing_num = get_phone_number(self.log, sec_dut)
413        if not initiate_call(self.log, pri_dut, outgoing_num):
414            self.log.error("Failed to initiate call")
415            return False
416        if not wait_and_answer_call(self.log, sec_dut):
417            self.log.error("Failed to answer call.")
418            return False
419        time.sleep(self.short_timeout)
420        call_id = pri_dut.droid.telecomCallGetCallIds()[0]
421        pri_dut.droid.telecomCallPlayDtmfTone(call_id, "9")
422        input("Press enter to switch to speaker...")
423        self.log.info("Switching to speaker.")
424        set_audio_route(self.log, pri_dut, AUDIO_ROUTE_SPEAKER)
425        time.sleep(self.short_timeout)
426        if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_SPEAKER:
427            self.log.error(
428                "Audio Route not set to {}".format(AUDIO_ROUTE_SPEAKER))
429            return False
430        input("Press enter to switch to earpiece...")
431        self.log.info("Switching to earpiece.")
432        set_audio_route(self.log, pri_dut, AUDIO_ROUTE_EARPIECE)
433        time.sleep(self.short_timeout)
434        if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_EARPIECE:
435            self.log.error(
436                "Audio Route not set to {}".format(AUDIO_ROUTE_EARPIECE))
437            return False
438        input("Press enter to switch to Bluetooth...")
439        self.log.info("Switching to Bluetooth...")
440        set_audio_route(self.log, pri_dut, AUDIO_ROUTE_BLUETOOTH)
441        time.sleep(self.short_timeout)
442        if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_BLUETOOTH:
443            self.log.error(
444                "Audio Route not set to {}".format(AUDIO_ROUTE_BLUETOOTH))
445            return False
446        input("Press enter to hangup call...")
447        self.log.info("Hanging up call...")
448        pri_dut.droid.telecomCallStopDtmfTone(call_id)
449        if not hangup_call(self.log, pri_dut):
450            self.log.error("Failed to hangup call")
451            return False
452        return True
453
454    def sms_receive_different_sizes(self, pri_dut, sec_dut):
455        self.log.info("Test recieve sms.")
456        input("Press enter to execute this testcase...")
457        msg = [rand_ascii_str(50), rand_ascii_str(1), rand_ascii_str(500)]
458        if not sms_send_receive_verify(self.log, sec_dut, pri_dut, msg):
459            return False
460        else:
461            self.log.info("Successfully sent sms. Please verify on carkit.")
462        return True
463
464    def sms_receive_multiple(self, pri_dut, sec_dut):
465        text_count = 10
466        self.log.info(
467            "Test sending {} sms messages to phone.".format(text_count))
468        input("Press enter to execute this testcase...")
469        for _ in range(text_count):
470            msg = [rand_ascii_str(50)]
471            if not sms_send_receive_verify(self.log, sec_dut, pri_dut, msg):
472                return False
473            else:
474                self.log.info(
475                    "Successfully sent sms. Please verify on carkit.")
476        return True
477
478    def sms_send_outgoing_texts(self, pri_dut, sec_dut):
479        self.log.info("Test send sms of different sizes.")
480        input("Press enter to execute this testcase...")
481        msg = [rand_ascii_str(50), rand_ascii_str(1), rand_ascii_str(500)]
482        if not sms_send_receive_verify(self.log, pri_dut, sec_dut, msg):
483            return False
484        else:
485            self.log.info("Successfully sent sms. Please verify on carkit.")
486        return True
487
488    def sms_during_incomming_call(self, pri_dut, sec_dut):
489        self.log.info(
490            "Test incomming call scenario to phone from unknown contact")
491        input("Press enter to execute this testcase...")
492        outgoing_num = get_phone_number(self.log, pri_dut)
493        if not initiate_call(self.log, sec_dut, outgoing_num):
494            self.log.error("Failed to initiate call")
495            return False
496        if not wait_and_answer_call(self.log, pri_dut):
497            self.log.error("Failed to answer call.")
498            return False
499        time.sleep(self.short_timeout)
500        msg = [rand_ascii_str(10)]
501        if not sms_send_receive_verify(self.log, sec_dut, pri_dut, msg):
502            return False
503        else:
504            self.log.info("Successfully sent sms. Please verify on carkit.")
505        input("Press enter to hangup call...")
506        if not hangup_call(self.log, sec_dut):
507            self.log.error("Failed to hangup call")
508            return False
509        return True
510
511    def multi_incomming_call(self, pri_dut, sec_dut, ter_dut):
512        self.log.info("Test 2 incomming calls scenario to phone.")
513        input("Press enter to execute this testcase...")
514        if not self.setup_multi_call(sec_dut, ter_dut, pri_dut):
515            return False
516        input("Press enter to hangup call 1...")
517        if not hangup_call(self.log, sec_dut):
518            self.log.error("Failed to hangup call")
519            return False
520        input("Press enter to hangup call 2...")
521        if not hangup_call(self.log, ter_dut):
522            self.log.error("Failed to hangup call")
523            return False
524        return True
525
526    def multi_call_audio_routing(self, pri_dut, sec_dut, ter_dut):
527        self.log.info(
528            "Test 2 incomming calls scenario to phone, then test audio routing."
529        )
530        input("Press enter to execute this testcase...")
531        if not self.setup_multi_call(sec_dut, ter_dut, pri_dut):
532            return False
533        input("Press enter to switch to earpiece...")
534        self.log.info("Switching to earpiece.")
535        set_audio_route(self.log, pri_dut, AUDIO_ROUTE_EARPIECE)
536        time.sleep(self.short_timeout)
537        if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_EARPIECE:
538            self.log.error(
539                "Audio Route not set to {}".format(AUDIO_ROUTE_EARPIECE))
540            return False
541        input("Press enter to switch to Bluetooth...")
542        self.log.info("Switching to Bluetooth...")
543        set_audio_route(self.log, pri_dut, AUDIO_ROUTE_BLUETOOTH)
544        time.sleep(self.short_timeout)
545        if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_BLUETOOTH:
546            self.log.error(
547                "Audio Route not set to {}".format(AUDIO_ROUTE_BLUETOOTH))
548            return False
549        input("Press enter to hangup call 1...")
550        if not hangup_call(self.log, sec_dut):
551            self.log.error("Failed to hangup call")
552            return False
553        input("Press enter to hangup call 2...")
554        if not hangup_call(self.log, ter_dut):
555            self.log.error("Failed to hangup call")
556            return False
557        return True
558
559    def multi_call_swap_multiple_times(self, pri_dut, sec_dut, ter_dut):
560        self.log.info(
561            "Test 2 incomming calls scenario to phone, then test audio routing."
562        )
563        input("Press enter to execute this testcase...")
564        if not self.setup_multi_call(sec_dut, ter_dut, pri_dut):
565            return False
566        input("Press enter to swap active calls...")
567        calls = pri_dut.droid.telecomCallGetCallIds()
568        if not swap_calls(self.log, [pri_dut, sec_dut, ter_dut], calls[0],
569                          calls[1], 5):
570            return False
571        input("Press enter to hangup call 1...")
572        if not hangup_call(self.log, sec_dut):
573            self.log.error("Failed to hangup call")
574            return False
575        input("Press enter to hangup call 2...")
576        if not hangup_call(self.log, ter_dut):
577            self.log.error("Failed to hangup call")
578            return False
579        return True
580
581    def multi_call_join_conference_call(self, pri_dut, sec_dut, ter_dut):
582        self.log.info(
583            "Test 2 incomming calls scenario to phone then join the calls.")
584        input("Press enter to execute this testcase...")
585        if not self.setup_multi_call(sec_dut, ter_dut, pri_dut):
586            return False
587        input("Press enter to join active calls...")
588        calls = pri_dut.droid.telecomCallGetCallIds()
589        pri_dut.droid.telecomCallJoinCallsInConf(calls[0], calls[1])
590        time.sleep(WAIT_TIME_IN_CALL)
591        if num_active_calls(self.log, pri_dut) != 4:
592            self.log.error("Total number of call ids in {} is not 4.".format(
593                pri_dut.serial))
594            return False
595        input("Press enter to hangup call 1...")
596        if not hangup_call(self.log, sec_dut):
597            self.log.error("Failed to hangup call")
598            return False
599        input("Press enter to hangup call 2...")
600        if not hangup_call(self.log, ter_dut):
601            self.log.error("Failed to hangup call")
602            return False
603        return True
604
605    def multi_call_join_conference_call_hangup_conf_call(
606            self, pri_dut, sec_dut, ter_dut):
607        self.log.info(
608            "Test 2 incomming calls scenario to phone then join the calls, then terminate the call from the primary dut."
609        )
610        input("Press enter to execute this testcase...")
611        if not self.setup_multi_call(sec_dut, ter_dut, pri_dut):
612            return False
613        input("Press enter to join active calls...")
614        calls = pri_dut.droid.telecomCallGetCallIds()
615        pri_dut.droid.telecomCallJoinCallsInConf(calls[0], calls[1])
616        time.sleep(WAIT_TIME_IN_CALL)
617        if num_active_calls(self.log, pri_dut) != 4:
618            self.log.error("Total number of call ids in {} is not 4.".format(
619                pri_dut.serial))
620            return False
621        input("Press enter to hangup conf call...")
622        if not hangup_call(self.log, pri_dut):
623            self.log.error("Failed to hangup call")
624            return False
625        return True
626
627    def outgoing_multi_call_join_conference_call(self, pri_dut, sec_dut,
628                                                 ter_dut):
629        self.log.info(
630            "Test 2 outgoing calls scenario from phone then join the calls.")
631        input("Press enter to execute this testcase...")
632        outgoing_num = get_phone_number(self.log, sec_dut)
633        if not initiate_call(self.log, pri_dut, outgoing_num):
634            self.log.error("Failed to initiate call")
635            return False
636        if not wait_and_answer_call(self.log, sec_dut):
637            self.log.error("Failed to answer call.")
638            return False
639        time.sleep(self.short_timeout)
640        outgoing_num = get_phone_number(self.log, ter_dut)
641        if not initiate_call(self.log, pri_dut, outgoing_num):
642            self.log.error("Failed to initiate call")
643            return False
644        if not wait_and_answer_call(self.log, ter_dut):
645            self.log.error("Failed to answer call.")
646            return False
647        input("Press enter to join active calls...")
648        calls = pri_dut.droid.telecomCallGetCallIds()
649        pri_dut.droid.telecomCallJoinCallsInConf(calls[0], calls[1])
650        time.sleep(WAIT_TIME_IN_CALL)
651        if num_active_calls(self.log, pri_dut) != 4:
652            self.log.error("Total number of call ids in {} is not 4.".format(
653                pri_dut.serial))
654            return False
655        input("Press enter to hangup call 1...")
656        if not hangup_call(self.log, sec_dut):
657            self.log.error("Failed to hangup call")
658            return False
659        input("Press enter to hangup call 2...")
660        if not hangup_call(self.log, ter_dut):
661            self.log.error("Failed to hangup call")
662            return False
663        return True
664
665    def multi_call_join_conference_call_audio_routes(self, pri_dut, sec_dut,
666                                                     ter_dut):
667        self.log.info(
668            "Test 2 incomming calls scenario to phone then join the calls, then test different audio routes."
669        )
670        input("Press enter to execute this testcase...")
671        if not self.setup_multi_call(sec_dut, ter_dut, pri_dut):
672            return False
673        input("Press enter to join active calls...")
674        calls = pri_dut.droid.telecomCallGetCallIds()
675        pri_dut.droid.telecomCallJoinCallsInConf(calls[0], calls[1])
676        time.sleep(WAIT_TIME_IN_CALL)
677        if num_active_calls(self.log, pri_dut) != 4:
678            self.log.error("Total number of call ids in {} is not 4.".format(
679                pri_dut.serial))
680            return False
681        input("Press enter to switch to phone speaker...")
682        self.log.info("Switching to earpiece.")
683        set_audio_route(self.log, pri_dut, AUDIO_ROUTE_EARPIECE)
684        time.sleep(self.short_timeout)
685        if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_EARPIECE:
686            self.log.error(
687                "Audio Route not set to {}".format(AUDIO_ROUTE_EARPIECE))
688            return False
689        input("Press enter to switch to Bluetooth...")
690        self.log.info("Switching to Bluetooth...")
691        set_audio_route(self.log, pri_dut, AUDIO_ROUTE_BLUETOOTH)
692        time.sleep(self.short_timeout)
693        if get_audio_route(self.log, pri_dut) != AUDIO_ROUTE_BLUETOOTH:
694            self.log.error(
695                "Audio Route not set to {}".format(AUDIO_ROUTE_BLUETOOTH))
696            return False
697        input("Press enter to hangup conf call...")
698        if not hangup_call(self.log, pri_dut):
699            self.log.error("Failed to hangup call")
700            return False
701        return True
702
703    def avrcp_play_pause(self, pri_dut):
704        play_pause_count = 5
705        self.log.info(
706            "Test AVRCP play/pause {} times.".format(play_pause_count))
707        pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
708        input(
709            "Press enter to execute this testcase when music is in the play state..."
710        )
711        for i in range(play_pause_count):
712            input("Execute iteration {}?".format(i + 1))
713            pri_dut.adb.shell(KEYCODE_EVENT_PLAY_PAUSE)
714        self.log.info("Test should end in a paused state.")
715        return True
716
717    def avrcp_next_previous_song(self, pri_dut):
718        self.log.info("Test AVRCP go to the next song then the previous song.")
719        pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
720        input(
721            "Press enter to execute this testcase when music is in the play state..."
722        )
723        self.log.info("Hitting Next input event...")
724        pri_dut.adb.shell(KEYCODE_EVENT_NEXT)
725        input("Press enter to go to the previous song")
726        pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
727        pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
728        self.log.info("Test should end on original song.")
729        return True
730
731    def avrcp_next_previous(self, pri_dut):
732        self.log.info(
733            "Test AVRCP go to the next song then the press previous after a few seconds."
734        )
735        pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
736        input(
737            "Press enter to execute this testcase when music is in the play state..."
738        )
739        self.log.info("Hitting Next input event...")
740        pri_dut.adb.shell(KEYCODE_EVENT_NEXT)
741        time.sleep(5)
742        self.log.info("Hitting Previous input event...")
743        pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
744        self.log.info("Test should end on \"next\" song.")
745        return True
746
747    def avrcp_next_repetative(self, pri_dut):
748        iterations = 10
749        self.log.info("Test AVRCP go to the next {} times".format(iterations))
750        pri_dut.adb.shell(KEYCODE_EVENT_PREVIOUS)
751        input(
752            "Press enter to execute this testcase when music is in the play state..."
753        )
754        for i in range(iterations):
755            self.log.info(
756                "Hitting Next input event, iteration {}...".format(i + 1))
757            pri_dut.adb.shell(KEYCODE_EVENT_NEXT)
758            # Allow time for the carkit to update.
759            time.sleep(1)
760        return True
761
762    def _cycle_aboslute_volume_control_helper(self, volume_step,
763                                              android_volume_steps, pri_dut):
764        begin_time = get_current_epoch_time()
765        pri_dut.droid.setMediaVolume(volume_step)
766        percentage_to_set = int((volume_step / android_volume_steps) * 100)
767        self.log.info("Setting phone volume to {}%".format(percentage_to_set))
768        volume_info_logcat = pri_dut.search_logcat(
769            logcat_strings['media_playback_vol_changed'], begin_time)
770        if len(volume_info_logcat) > 1:
771            self.log.info("Instant response detected.")
772            carkit_response = volume_info_logcat[-1]['log_message'].split(',')
773            for item in carkit_response:
774                if " volume=" in item:
775                    carkit_vol_response = int((
776                        int(item.split("=")[-1]) / android_volume_steps) * 100)
777                    self.log.info(
778                        "Carkit set volume to {}%".format(carkit_vol_response))
779        result = input(
780            "Did volume change reflect properly on carkit and phone? (Y/n) "
781        ).lower()
782
783    def cycle_absolute_volume_control(self, pri_dut):
784        result = input(
785            "Does carkit support Absolute Volume Control? (Y/n) ").lower()
786        if result is "n":
787            return True
788        android_volume_steps = 25
789        for i in range(android_volume_steps):
790            self._cycle_aboslute_volume_control_helper(i, android_volume_steps,
791                                                       pri_dut)
792        for i in reversed(range(android_volume_steps)):
793            self._cycle_aboslute_volume_control_helper(i, android_volume_steps,
794                                                       pri_dut)
795        return True
796
797    def cycle_battery_level(self, pri_dut):
798        for i in range(11):
799            level = i * 10
800            pri_dut.shell.set_battery_level(level)
801            question = "Phone battery level {}. Has the carkit indicator " \
802                "changed? (Y/n) "
803            result = input(question.format(level)).lower()
804
805    def test_voice_recognition_from_phone(self, pri_dut):
806        result = input(
807            "Does carkit support voice recognition (BVRA)? (Y/n) ").lower()
808        if result is "n":
809            return True
810        input("Press enter to start voice recognition from phone.")
811        self.pri_dut.droid.bluetoothHspStartVoiceRecognition(
812            self.target_mac_address)
813        input("Press enter to stop voice recognition from phone.")
814        self.pri_dut.droid.bluetoothHspStopVoiceRecognition(
815            self.target_mac_address)
816
817    def test_audio_and_voice_recognition_from_phone(self, pri_dut):
818        result = input(
819            "Does carkit support voice recognition (BVRA)? (Y/n) ").lower()
820        if result is "n":
821            return True
822        # Start playing music here
823        input("Press enter to start voice recognition from phone.")
824        self.pri_dut.droid.bluetoothHspStartVoiceRecognition(
825            self.target_mac_address)
826        input("Press enter to stop voice recognition from phone.")
827        self.pri_dut.droid.bluetoothHspStopVoiceRecognition(
828            self.target_mac_address)
829        time.sleep(2)
830        result = input("Did carkit continue music playback after? (Y/n) ")
831