• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
18
19from acts.keys import Config
20from acts.utils import rand_ascii_str
21from acts_contrib.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
22from acts_contrib.test_utils.bt.bt_constants import logcat_strings
23from acts_contrib.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
24from acts_contrib.test_utils.tel.tel_defines import AUDIO_ROUTE_BLUETOOTH
25from acts_contrib.test_utils.tel.tel_defines import AUDIO_ROUTE_EARPIECE
26from acts_contrib.test_utils.tel.tel_defines import AUDIO_ROUTE_SPEAKER
27from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_IN_CALL
28from acts_contrib.test_utils.tel.tel_message_utils import sms_send_receive_verify
29from acts_contrib.test_utils.tel.tel_test_utils import get_phone_number
30from acts_contrib.test_utils.tel.tel_test_utils import num_active_calls
31from acts_contrib.test_utils.tel.tel_voice_utils import get_audio_route
32from acts_contrib.test_utils.tel.tel_voice_utils import hangup_call
33from acts_contrib.test_utils.tel.tel_voice_utils import initiate_call
34from acts_contrib.test_utils.tel.tel_voice_utils import set_audio_route
35from acts_contrib.test_utils.tel.tel_voice_utils import swap_calls
36from acts_contrib.test_utils.tel.tel_voice_utils import wait_and_answer_call
37from acts_contrib.test_utils.tel.tel_test_utils import toggle_airplane_mode
38from acts_contrib.test_utils.tel.tel_voice_utils import call_setup_teardown
39from acts.utils import exe_cmd
40from acts.utils import get_current_epoch_time
41
42KEYCODE_VOLUME_UP = "input keyevent 24"
43KEYCODE_VOLUME_DOWN = "input keyevent 25"
44KEYCODE_EVENT_PLAY_PAUSE = "input keyevent 85"
45KEYCODE_MEDIA_STOP = "input keyevent 86"
46KEYCODE_EVENT_NEXT = "input keyevent 87"
47KEYCODE_EVENT_PREVIOUS = "input keyevent 88"
48KEYCODE_MEDIA_REWIND = "input keyevent 89"
49KEYCODE_MEDIA_FAST_FORWARD = "input keyevent 90"
50KEYCODE_MUTE = "input keyevent 91"
51
52default_timeout = 10
53
54
55class E2eBtCarkitLib():
56
57    android_devices = []
58    short_timeout = 3
59    active_call_id = None
60    hold_call_id = None
61    log = None
62    mac_address = None
63
64    def __init__(self, log, target_mac_address=None):
65        self.log = log
66        self.target_mac_address = target_mac_address
67
68    def connect_hsp_helper(self, ad):
69        end_time = time.time() + default_timeout + 10
70        connected_hsp_devices = len(ad.droid.bluetoothHspGetConnectedDevices())
71        while connected_hsp_devices != 1 and time.time() < end_time:
72            try:
73                ad.droid.bluetoothHspConnect(self.target_mac_address)
74                time.sleep(3)
75                if len(ad.droid.bluetoothHspGetConnectedDevices() == 1):
76                    break
77            except Exception:
78                self.log.debug("Failed to connect hsp trying again...")
79            try:
80                ad.droid.bluetoothConnectBonded(self.target_mac_address)
81            except Exception:
82                self.log.info("Failed to connect to bonded device...")
83            connected_hsp_devices = len(
84                ad.droid.bluetoothHspGetConnectedDevices())
85        if connected_hsp_devices != 1:
86            self.log.error("Failed to reconnect to HSP service...")
87            return False
88        self.log.info("Connected to HSP service...")
89        return True
90
91    def setup_multi_call(self, caller0, caller1, callee):
92        outgoing_num = get_phone_number(self.log, callee)
93        if not initiate_call(self.log, caller0, outgoing_num):
94            self.log.error("Failed to initiate call")
95            return False
96        if not wait_and_answer_call(self.log, callee):
97            self.log.error("Failed to answer call.")
98            return False
99        time.sleep(self.short_timeout)
100        if not initiate_call(self.log, caller1, outgoing_num):
101            self.log.error("Failed to initiate call")
102            return False
103        if not wait_and_answer_call(self.log, callee):
104            self.log.error("Failed to answer call.")
105            return False
106        return True
107
108    def process_tests(self, tests):
109        for test in tests:
110            try:
111                test()
112            except Exception as err:
113                self.log.error(err)
114
115    def run_suite_hfp_tests(self):
116        tests = [
117            self.outgoing_call_private_number,
118            self.outgoing_call_unknown_contact,
119            self.incomming_call_private_number,
120            self.incomming_call_unknown_contact,
121            self.outgoing_call_multiple_iterations,
122            self.outgoing_call_hsp_disabled_then_enabled_during_call,
123            self.call_audio_routes,
124            self.sms_during_incomming_call,
125            self.multi_incomming_call,
126            self.multi_call_audio_routing,
127            self.multi_call_swap_multiple_times,
128            self.outgoing_call_a2dp_play_before_and_after,
129        ]
130        _process_tests(tests)
131
132    def run_suite_hfp_conf_tests(self):
133        tests = [
134            self.multi_call_join_conference_call,
135            self.multi_call_join_conference_call_hangup_conf_call,
136            self.outgoing_multi_call_join_conference_call,
137            self.multi_call_join_conference_call_audio_routes,
138        ]
139        _process_tests(tests)
140
141    def run_suite_map_tests(self):
142        tests = [
143            self.sms_receive_different_sizes,
144            self.sms_receive_multiple,
145            self.sms_send_outgoing_texts,
146        ]
147        _process_tests(tests)
148
149    def run_suite_avrcp_tests(self):
150        tests = [
151            self.avrcp_play_pause,
152            self.avrcp_next_previous_song,
153            self.avrcp_next_previous,
154            self.avrcp_next_repetative,
155        ]
156        _process_tests(tests)
157
158    def disconnect_reconnect_multiple_iterations(self, pri_dut):
159        iteration_count = 5
160        self.log.info(
161            "Test disconnect-reconnect scenario from phone {} times.".format(
162                iteration_count))
163        self.log.info(
164            "This test will prompt for user interaction after each reconnect.")
165        input("Press enter to execute this testcase...")
166        #Assumes only one devices connected
167        grace_timeout = 4  #disconnect and reconnect timeout
168        for n in range(iteration_count):
169            self.log.info("Test iteration {}.".format(n + 1))
170            self.log.info("Disconnecting device {}...".format(
171                self.target_mac_address))
172            pri_dut.droid.bluetoothDisconnectConnected(self.target_mac_address)
173            # May have to do a longer sleep for carkits.... need to test
174            time.sleep(grace_timeout)
175            self.log.info("Connecting device {}...".format(
176                self.target_mac_address))
177            pri_dut.droid.bluetoothConnectBonded(self.target_mac_address)
178            if not self.connect_hsp_helper(pri_dut):
179                return False
180            start_time = time.time()
181            connected_devices = pri_dut.droid.bluetoothGetConnectedDevices()
182            self.log.info(
183                "Waiting up to 10 seconds for device to reconnect...")
184            while time.time() < start_time + 10 and len(
185                    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) *
777                        100)
778                    self.log.info(
779                        "Carkit set volume to {}%".format(carkit_vol_response))
780        result = input(
781            "Did volume change reflect properly on carkit and phone? (Y/n) "
782        ).lower()
783
784    def cycle_absolute_volume_control(self, pri_dut):
785        result = input(
786            "Does carkit support Absolute Volume Control? (Y/n) ").lower()
787        if result == "n":
788            return True
789        android_volume_steps = 25
790        for i in range(android_volume_steps):
791            self._cycle_aboslute_volume_control_helper(i, android_volume_steps,
792                                                       pri_dut)
793        for i in reversed(range(android_volume_steps)):
794            self._cycle_aboslute_volume_control_helper(i, android_volume_steps,
795                                                       pri_dut)
796        return True
797
798    def cycle_battery_level(self, pri_dut):
799        for i in range(11):
800            level = i * 10
801            pri_dut.shell.set_battery_level(level)
802            question = "Phone battery level {}. Has the carkit indicator " \
803                "changed? (Y/n) "
804            result = input(question.format(level)).lower()
805
806    def test_voice_recognition_from_phone(self, pri_dut):
807        result = input(
808            "Does carkit support voice recognition (BVRA)? (Y/n) ").lower()
809        if result == "n":
810            return True
811        input("Press enter to start voice recognition from phone.")
812        self.pri_dut.droid.bluetoothHspStartVoiceRecognition(
813            self.target_mac_address)
814        input("Press enter to stop voice recognition from phone.")
815        self.pri_dut.droid.bluetoothHspStopVoiceRecognition(
816            self.target_mac_address)
817
818    def test_audio_and_voice_recognition_from_phone(self, pri_dut):
819        result = input(
820            "Does carkit support voice recognition (BVRA)? (Y/n) ").lower()
821        if result == "n":
822            return True
823        # Start playing music here
824        input("Press enter to start voice recognition from phone.")
825        self.pri_dut.droid.bluetoothHspStartVoiceRecognition(
826            self.target_mac_address)
827        input("Press enter to stop voice recognition from phone.")
828        self.pri_dut.droid.bluetoothHspStopVoiceRecognition(
829            self.target_mac_address)
830        time.sleep(2)
831        result = input("Did carkit continue music playback after? (Y/n) ")
832