1#!/usr/bin/env python3
2#
3# Copyright (C) 2016 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"""
17Python script for wrappers to various libraries.
18"""
19from acts.test_utils.bt.bt_constants import bt_scan_mode_types
20from acts.test_utils.bt.bt_constants import gatt_server_responses
21import acts.test_utils.bt.gatt_test_database as gatt_test_database
22from acts.test_utils.bt.bt_carkit_lib import E2eBtCarkitLib
23
24import threading
25import time
26import cmd
27"""Various Global Strings"""
28CMD_LOG = "CMD {} result: {}"
29FAILURE = "CMD {} threw exception: {}"
30
31
32class CmdInput(cmd.Cmd):
33    android_devices = []
34    """Simple command processor for Bluetooth PTS Testing"""
35
36    def connect_hsp_helper(self, ad):
37        """A helper function for making HSP connections"""
38        end_time = time.time() + 20
39        connected_hsp_devices = len(ad.droid.bluetoothHspGetConnectedDevices())
40        while connected_hsp_devices != 1 and time.time() < end_time:
41            try:
42                ad.droid.bluetoothHspConnect(self.mac_addr)
43                time.sleep(3)
44                if len(ad.droid.bluetoothHspGetConnectedDevices() == 1):
45                    break
46            except Exception:
47                self.log.debug("Failed to connect hsp trying again...")
48            try:
49                ad.droid.bluetoothConnectBonded(self.mac_addr)
50            except Exception:
51                self.log.info("Failed to connect to bonded device...")
52            connected_hsp_devices = len(
53                ad.droid.bluetoothHspGetConnectedDevices())
54        if connected_hsp_devices != 1:
55            self.log.error("Failed to reconnect to HSP service...")
56            return False
57        self.log.info("Connected to HSP service...")
58        return True
59
60    def setup_vars(self, android_devices, mac_addr, log):
61        self.pri_dut = android_devices[0]
62        if len(android_devices) > 1:
63            self.sec_dut = android_devices[1]
64        if len(android_devices) > 2:
65            self.ter_dut = android_devices[2]
66        if len(android_devices) > 3:
67            self.qua_dut = android_devices[3]
68        self.mac_addr = mac_addr
69        self.log = log
70        self.pri_dut.bta.set_target_mac_addr(mac_addr)
71        self.pri_dut.gattc.set_target_mac_addr(mac_addr)
72        self.pri_dut.rfcomm.set_target_mac_addr(mac_addr)
73        self.bt_carkit_lib = E2eBtCarkitLib(self.log, mac_addr)
74
75    def emptyline(self):
76        pass
77
78    def do_EOF(self, line):
79        "End Script"
80        return True
81
82    def do_reset_mac_address(self, line):
83        """Reset target mac address for libraries based on the current connected device"""
84        try:
85            device = self.pri_dut.droid.bluetoothGetConnectedDevices()[0]
86            #self.setup_vars(self.android_devices, device['address'], self.log)
87            self.log.info("New device is {}".format(device))
88        except Exception as err:
89            self.log.info("Failed to setup new vars with {}".format(err))
90
91    """Begin GATT Client wrappers"""
92
93    def do_gattc_socket_conn_begin_connect_thread_psm(self, line):
94        cmd = ""
95        try:
96            self.pri_dut.gattc.socket_conn_begin_connect_thread_psm(line)
97        except Exception as err:
98            self.log.info(FAILURE.format(cmd, err))
99
100    def do_gattc_socket_conn_begin_accept_thread_psm(self, line):
101        cmd = ""
102        try:
103            self.pri_dut.gattc.socket_conn_begin_accept_thread_psm(line)
104        except Exception as err:
105            self.log.info(FAILURE.format(cmd, err))
106
107    def do_gattc_request_le_connection_parameters(self, line):
108        cmd = ""
109        try:
110            self.pri_dut.gattc.request_le_connection_parameters()
111        except Exception as err:
112            self.log.info(FAILURE.format(cmd, err))
113
114    def do_gattc_connect_over_le(self, line):
115        """Perform GATT connection over LE"""
116        cmd = "Gatt connect over LE"
117        try:
118            autoconnect = False
119            if line:
120                autoconnect = bool(line)
121            self.pri_dut.gattc.connect_over_le(autoconnect)
122        except Exception as err:
123            self.log.info(FAILURE.format(cmd, err))
124
125    def do_gattc_connect_over_bredr(self, line):
126        """Perform GATT connection over BREDR"""
127        cmd = "Gatt connect over BR/EDR"
128        try:
129            self.pri_dut.gattc.connect_over_bredr()
130        except Exception as err:
131            self.log.info(FAILURE.format(cmd, err))
132
133    def do_gattc_disconnect(self, line):
134        """Perform GATT disconnect"""
135        cmd = "Gatt Disconnect"
136        try:
137            self.pri_dut.gattc.disconnect()
138        except Exception as err:
139            self.log.info(FAILURE.format(cmd, err))
140
141    def do_gattc_read_char_by_uuid(self, line):
142        """GATT client read Characteristic by UUID."""
143        cmd = "GATT client read Characteristic by UUID."
144        try:
145            self.pri_dut.gattc.read_char_by_uuid(line)
146        except Exception as err:
147            self.log.info(FAILURE.format(cmd, err))
148
149    def do_gattc_request_mtu(self, line):
150        """Request MTU Change of input value"""
151        cmd = "Request MTU Value"
152        try:
153            self.pri_dut.gattc.request_mtu(line)
154        except Exception as err:
155            self.log.info(FAILURE.format(cmd, err))
156
157    def do_gattc_list_all_uuids(self, line):
158        """From the GATT Client, discover services and list all services,
159        chars and descriptors
160        """
161        cmd = "Discovery Services and list all UUIDS"
162        try:
163            self.pri_dut.gattc.list_all_uuids()
164        except Exception as err:
165            self.log.info(FAILURE.format(cmd, err))
166
167    def do_gattc_discover_services(self, line):
168        """GATT Client discover services of GATT Server"""
169        cmd = "Discovery Services of GATT Server"
170        try:
171            self.pri_dut.gattc.discover_services()
172        except Exception as err:
173            self.log.info(FAILURE.format(cmd, err))
174
175    def do_gattc_refresh(self, line):
176        """Perform Gatt Client Refresh"""
177        cmd = "GATT Client Refresh"
178        try:
179            self.pri_dut.gattc.refresh()
180        except Exception as err:
181            self.log.info(FAILURE.format(cmd, err))
182
183    def do_gattc_read_char_by_instance_id(self, line):
184        """From the GATT Client, discover services and list all services,
185        chars and descriptors
186        """
187        cmd = "GATT Client Read By Instance ID"
188        try:
189            self.pri_dut.gattc.read_char_by_instance_id(line)
190        except Exception as err:
191            self.log.info(FAILURE.format(cmd, err))
192
193    def do_gattc_write_char_by_instance_id(self, line):
194        """GATT Client Write to Characteristic by instance ID"""
195        cmd = "GATT Client write to Characteristic by instance ID"
196        try:
197            self.pri_dut.gattc.write_char_by_instance_id(line)
198        except Exception as err:
199            self.log.info(FAILURE.format(cmd, err))
200
201    def do_gattc_write_char_by_instance_id_value(self, line):
202        """GATT Client Write to Characteristic by instance ID"""
203        cmd = "GATT Client write to Characteristic by instance ID"
204        try:
205            self.pri_dut.gattc.write_char_by_instance_id_value(line)
206        except Exception as err:
207            self.log.info(FAILURE.format(cmd, err))
208
209    def do_gattc_mod_write_char_by_instance_id(self, line):
210        """GATT Client Write to Char that doesn't have write permission"""
211        cmd = "GATT Client Write to Char that doesn't have write permission"
212        try:
213            self.pri_dut.gattc.mod_write_char_by_instance_id(line)
214        except Exception as err:
215            self.log.info(FAILURE.format(cmd, err))
216
217    def do_gattc_write_invalid_char_by_instance_id(self, line):
218        """GATT Client Write to Char that doesn't exists"""
219        try:
220            self.pri_dut.gattc.write_invalid_char_by_instance_id(line)
221        except Exception as err:
222            self.log.info(FAILURE.format(cmd, err))
223
224    def do_gattc_mod_read_char_by_instance_id(self, line):
225        """GATT Client Read Char that doesn't have write permission"""
226        cmd = "GATT Client Read Char that doesn't have write permission"
227        try:
228            self.pri_dut.gattc.mod_read_char_by_instance_id(line)
229        except Exception as err:
230            self.log.info(FAILURE.format(cmd, err))
231
232    def do_gattc_read_invalid_char_by_instance_id(self, line):
233        """GATT Client Read Char that doesn't exists"""
234        cmd = "GATT Client Read Char that doesn't exists"
235        try:
236            self.pri_dut.gattc.read_invalid_char_by_instance_id(line)
237        except Exception as err:
238            self.log.info(FAILURE.format(cmd, err))
239
240    def do_gattc_mod_write_desc_by_instance_id(self, line):
241        """GATT Client Write to Desc that doesn't have write permission"""
242        cmd = "GATT Client Write to Desc that doesn't have write permission"
243        try:
244            self.pri_dut.gattc.mod_write_desc_by_instance_id(line)
245        except Exception as err:
246            self.log.info(FAILURE.format(cmd, err))
247
248    def do_gattc_write_invalid_desc_by_instance_id(self, line):
249        """GATT Client Write to Desc that doesn't exists"""
250        cmd = "GATT Client Write to Desc that doesn't exists"
251        try:
252            self.pri_dut.gattc.write_invalid_desc_by_instance_id(line)
253        except Exception as err:
254            self.log.info(FAILURE.format(cmd, err))
255
256    def do_gattc_mod_read_desc_by_instance_id(self, line):
257        """GATT Client Read Desc that doesn't have write permission"""
258        cmd = "GATT Client Read Desc that doesn't have write permission"
259        try:
260            self.pri_dut.gattc.mod_read_desc_by_instance_id(line)
261        except Exception as err:
262            self.log.info(FAILURE.format(cmd, err))
263
264    def do_gattc_read_invalid_desc_by_instance_id(self, line):
265        """GATT Client Read Desc that doesn't exists"""
266        cmd = "GATT Client Read Desc that doesn't exists"
267        try:
268            self.pri_dut.gattc.read_invalid_desc_by_instance_id(line)
269        except Exception as err:
270            self.log.info(FAILURE.format(cmd, err))
271
272    def do_gattc_mod_read_char_by_uuid_and_instance_id(self, line):
273        """GATT Client Read Char that doesn't have write permission"""
274        cmd = "GATT Client Read Char that doesn't have write permission"
275        try:
276            self.pri_dut.gattc.mod_read_char_by_uuid_and_instance_id(line)
277        except Exception as err:
278            self.log.info(FAILURE.format(cmd, err))
279
280    def do_gattc_read_invalid_char_by_uuid(self, line):
281        """GATT Client Read Char that doesn't exist"""
282        cmd = "GATT Client Read Char that doesn't exist"
283        try:
284            self.pri_dut.gattc.read_invalid_char_by_uuid(line)
285        except Exception as err:
286            self.log.info(FAILURE.format(cmd, err))
287
288    def do_gattc_write_desc_by_instance_id(self, line):
289        """GATT Client Write to Descriptor by instance ID"""
290        cmd = "GATT Client Write to Descriptor by instance ID"
291        try:
292            self.pri_dut.gattc.write_desc_by_instance_id(line)
293        except Exception as err:
294            self.log.info(FAILURE.format(cmd, err))
295
296    def do_gattc_enable_notification_desc_by_instance_id(self, line):
297        """GATT Client Enable Notification on Descriptor by instance ID"""
298        cmd = "GATT Client Enable Notification on Descriptor by instance ID"
299        try:
300            self.pri_dut.gattc.enable_notification_desc_by_instance_id(line)
301        except Exception as err:
302            self.log.info(FAILURE.format(cmd, err))
303
304    def do_gattc_enable_indication_desc_by_instance_id(self, line):
305        """GATT Client Enable Indication on Descriptor by instance ID"""
306        cmd = "GATT Client Enable Indication on Descriptor by instance ID"
307        try:
308            self.pri_dut.gattc.enable_indication_desc_by_instance_id(line)
309        except Exception as err:
310            self.log.info(FAILURE.format(cmd, err))
311
312    def do_gattc_char_enable_all_notifications(self, line):
313        """GATT Client enable all notifications"""
314        cmd = "GATT Client enable all notifications"
315        try:
316            self.pri_dut.gattc.char_enable_all_notifications()
317        except Exception as err:
318            self.log.info(FAILURE.format(cmd, err))
319
320    def do_gattc_read_char_by_invalid_instance_id(self, line):
321        """GATT Client read char by non-existant instance id"""
322        cmd = "GATT Client read char by non-existant instance id"
323        try:
324            self.pri_dut.gattc.read_char_by_invalid_instance_id(line)
325        except Exception as err:
326            self.log.info(FAILURE.format(cmd, err))
327
328    def do_gattc_begin_reliable_write(self, line):
329        """Begin a reliable write on the Bluetooth Gatt Client"""
330        cmd = "GATT Client Begin Reliable Write"
331        try:
332            self.pri_dut.gattc.begin_reliable_write()
333        except Exception as err:
334            self.log.info(FAILURE.format(cmd, err))
335
336    def do_gattc_abort_reliable_write(self, line):
337        """Abort a reliable write on the Bluetooth Gatt Client"""
338        cmd = "GATT Client Abort Reliable Write"
339        try:
340            self.pri_dut.gattc.abort_reliable_write()
341        except Exception as err:
342            self.log.info(FAILURE.format(cmd, err))
343
344    def do_gattc_execute_reliable_write(self, line):
345        """Execute a reliable write on the Bluetooth Gatt Client"""
346        cmd = "GATT Client Execute Reliable Write"
347        try:
348            self.pri_dut.gattc.execute_reliable_write()
349        except Exception as err:
350            self.log.info(FAILURE.format(cmd, err))
351
352    def do_gattc_read_all_char(self, line):
353        """GATT Client read all Characteristic values"""
354        cmd = "GATT Client read all Characteristic values"
355        try:
356            self.pri_dut.gattc.read_all_char()
357        except Exception as err:
358            self.log.info(FAILURE.format(cmd, err))
359
360    def do_gattc_write_all_char(self, line):
361        """Write to every Characteristic on the GATT server"""
362        cmd = "GATT Client Write All Characteristics"
363        try:
364            self.pri_dut.gattc.write_all_char(line)
365        except Exception as err:
366            self.log.info(FAILURE.format(cmd, err))
367
368    def do_gattc_write_all_desc(self, line):
369        """ Write to every Descriptor on the GATT server """
370        cmd = "GATT Client Write All Descriptors"
371        try:
372            self.pri_dut.gattc.write_all_desc(line)
373        except Exception as err:
374            self.log.info(FAILURE.format(cmd, err))
375
376    def do_gattc_write_desc_notification_by_instance_id(self, line):
377        """Write 0x00 or 0x02 to notification descriptor [instance_id]"""
378        cmd = "Write 0x00 0x02 to notification descriptor"
379        try:
380            self.pri_dut.gattc.write_desc_notification_by_instance_id(line)
381        except Exception as err:
382            self.log.info(FAILURE.format(cmd, err))
383
384    def do_gattc_discover_service_by_uuid(self, line):
385        """Discover service by uuid"""
386        cmd = "Discover service by uuid"
387        try:
388            self.pri_dut.gattc.discover_service_by_uuid(line)
389        except Exception as err:
390            self.log.info(FAILURE.format(cmd, err))
391
392    def do_gattc_read_all_desc(self, line):
393        """Read all Descriptor values"""
394        cmd = "Read all Descriptor values"
395        try:
396            self.pri_dut.gattc.read_all_desc()
397        except Exception as err:
398            self.log.info(FAILURE.format(cmd, err))
399
400    """End GATT Client wrappers"""
401    """Begin GATT Server wrappers"""
402
403    def do_gatts_close_bluetooth_gatt_servers(self, line):
404        """Close Bluetooth Gatt Servers"""
405        cmd = "Close Bluetooth Gatt Servers"
406        try:
407            self.pri_dut.gatts.close_bluetooth_gatt_servers()
408        except Exception as err:
409            self.log.info(FAILURE.format(cmd, err))
410
411    def complete_gatts_setup_database(self, text, line, begidx, endidx):
412        if not text:
413            completions = list(
414                gatt_test_database.GATT_SERVER_DB_MAPPING.keys())[:]
415        else:
416            completions = [
417                s for s in gatt_test_database.GATT_SERVER_DB_MAPPING.keys()
418                if s.startswith(text)
419            ]
420        return completions
421
422    def complete_gatts_send_response(self, text, line, begidx, endidx):
423        """GATT Server database name completion"""
424        if not text:
425            completions = list(gatt_server_responses.keys())[:]
426        else:
427            completions = [
428                s for s in gatt_server_responses.keys() if s.startswith(text)
429            ]
430        return completions
431
432    def complete_gatts_send_continuous_response(self, text, line, begidx,
433                                                endidx):
434        """GATT Server database name completion"""
435        if not text:
436            completions = list(gatt_server_responses.keys())[:]
437        else:
438            completions = [
439                s for s in gatt_server_responses.keys() if s.startswith(text)
440            ]
441        return completions
442
443    def complete_gatts_send_continuous_response_data(self, text, line, begidx,
444                                                     endidx):
445        """GATT Server database name completion"""
446        if not text:
447            completions = list(gatt_server_responses.keys())[:]
448        else:
449            completions = [
450                s for s in gatt_server_responses.keys() if s.startswith(text)
451            ]
452        return completions
453
454    def do_gatts_list_all_uuids(self, line):
455        """From the GATT Client, discover services and list all services,
456        chars and descriptors
457        """
458        cmd = "Discovery Services and list all UUIDS"
459        try:
460            self.pri_dut.gatts.list_all_uuids()
461        except Exception as err:
462            self.log.info(FAILURE.format(cmd, err))
463
464    def do_gatts_send_response(self, line):
465        """Send a response to the GATT Client"""
466        cmd = "GATT server send response"
467        try:
468            self.pri_dut.gatts.send_response(line)
469        except Exception as err:
470            self.log.info(FAILURE.format(cmd, err))
471
472    def do_gatts_notify_characteristic_changed(self, line):
473        """Notify char changed by instance id [instance_id] [true|false]"""
474        cmd = "Notify characteristic changed by instance id"
475        try:
476            info = line.split()
477            instance_id = info[0]
478            confirm_str = info[1]
479            confirm = False
480            if confirm_str.lower() == 'true':
481                confirm = True
482            self.pri_dut.gatts.notify_characteristic_changed(
483                instance_id, confirm)
484        except Exception as err:
485            self.log.info(FAILURE.format(cmd, err))
486
487    def do_gatts_setup_database(self, line):
488        cmd = "Setup GATT Server database: {}".format(line)
489        try:
490            self.pri_dut.gatts.setup_gatts_db(
491                gatt_test_database.GATT_SERVER_DB_MAPPING.get(line))
492        except Exception as err:
493            self.log.info(FAILURE.format(cmd, err))
494
495    def do_gatts_characteristic_set_value_by_instance_id(self, line):
496        """Set Characteristic value by instance id"""
497        cmd = "Change value of a characteristic by instance id"
498        try:
499            info = line.split()
500            instance_id = info[0]
501            size = int(info[1])
502            value = []
503            for i in range(size):
504                value.append(i % 256)
505            self.pri_dut.gatts.characteristic_set_value_by_instance_id(
506                instance_id, value)
507        except Exception as err:
508            self.log.info(FAILURE.format(cmd, err))
509
510    def do_notify_characteristic_changed(self, line):
511        """Notify characteristic changed [instance_id] [confirm]"""
512        cmd = "Notify characteristic changed"
513        try:
514            info = line.split()
515            instance_id = info[0]
516            confirm = bool(info[1])
517            self.pri_dut.gatts.notify_characteristic_changed(
518                instance_id, confirm)
519        except Exception as err:
520            self.log.info(FAILURE.format(cmd, err))
521
522    def do_gatts_open(self, line):
523        """Open a GATT Server instance"""
524        cmd = "Open an empty GATT Server"
525        try:
526            self.pri_dut.gatts.open()
527        except Exception as err:
528            self.log.info(FAILURE.format(cmd, err))
529
530    def do_gatts_clear_services(self, line):
531        """Clear BluetoothGattServices from BluetoothGattServer"""
532        cmd = "Clear BluetoothGattServices from BluetoothGattServer"
533        try:
534            self.pri_dut.gatts.gatt_server_clear_services()
535        except Exception as err:
536            self.log.info(FAILURE.format(cmd, err))
537
538    def do_gatts_send_continuous_response(self, line):
539        """Send continous response with random data"""
540        cmd = "Send continous response with random data"
541        try:
542            self.pri_dut.gatts.send_continuous_response(line)
543        except Exception as err:
544            self.log.info(FAILURE.format(cmd, err))
545
546    def do_gatts_send_continuous_response_data(self, line):
547        """Send continous response including requested data"""
548        cmd = "Send continous response including requested data"
549        try:
550            self.pri_dut.gatts.send_continuous_response_data(line)
551        except Exception as err:
552            self.log.info(FAILURE.format(cmd, err))
553
554    """End GATT Server wrappers"""
555    """Begin Ble wrappers"""
556
557    def complete_ble_adv_data_include_local_name(self, text, line, begidx,
558                                                 endidx):
559        options = ['true', 'false']
560        if not text:
561            completions = list(options)[:]
562        else:
563            completions = [s for s in options if s.startswith(text)]
564        return completions
565
566    def complete_ble_adv_data_include_tx_power_level(self, text, line, begidx,
567                                                     endidx):
568        options = ['true', 'false']
569        if not text:
570            completions = list(options)[:]
571        else:
572            completions = [s for s in options if s.startswith(text)]
573        return completions
574
575    def complete_ble_stop_advertisement(self, text, line, begidx, endidx):
576        str_adv_list = list(map(str, self.pri_dut.ble.ADVERTISEMENT_LIST))
577        if not text:
578            completions = str_adv_list[:]
579        else:
580            completions = [s for s in str_adv_list if s.startswith(text)]
581        return completions
582
583    def do_ble_start_generic_connectable_advertisement(self, line):
584        """Start a connectable LE advertisement"""
585        cmd = "Start a connectable LE advertisement"
586        try:
587            self.pri_dut.ble.start_generic_connectable_advertisement(line)
588        except Exception as err:
589            self.log.info(FAILURE.format(cmd, err))
590
591    def _start_max_advertisements(self, ad):
592        self.pri_dut.ble.start_max_advertisements(ad)
593
594    def do_ble_start_generic_connectable_beacon_swarm(self, line):
595        """Start a connectable LE advertisement"""
596        cmd = "Start as many advertisements as possible on all devices"
597        try:
598            threads = []
599            for ad in self.android_devices:
600                thread = threading.Thread(
601                    target=self._start_max_advertisements, args=([ad]))
602                threads.append(thread)
603                thread.start()
604            for t in threads:
605                t.join()
606        except Exception as err:
607            self.log.info(FAILURE.format(cmd, err))
608
609    def do_ble_start_connectable_advertisement_set(self, line):
610        """Start a connectable advertisement set"""
611        try:
612            self.pri_dut.ble.start_connectable_advertisement_set(line)
613        except Exception as err:
614            self.log.error("Failed to start advertisement: {}".format(err))
615
616    def do_ble_stop_all_advertisement_set(self, line):
617        """Stop all advertisement sets"""
618        try:
619            self.pri_dut.ble.stop_all_advertisement_set(line)
620        except Exception as err:
621            self.log.error("Failed to stop advertisement: {}".format(err))
622
623    def do_ble_adv_add_service_uuid_list(self, line):
624        """Add service UUID to the LE advertisement inputs:
625         [uuid1 uuid2 ... uuidN]"""
626        cmd = "Add a valid service UUID to the advertisement."
627        try:
628            self.pri_dut.ble.adv_add_service_uuid_list(line)
629        except Exception as err:
630            self.log.info(FAILURE.format(cmd, err))
631
632    def do_ble_adv_data_include_local_name(self, line):
633        """Include local name in the advertisement. inputs: [true|false]"""
634        cmd = "Include local name in the advertisement."
635        try:
636            self.pri_dut.ble.adv_data_include_local_name(line)
637        except Exception as err:
638            self.log.info(FAILURE.format(cmd, err))
639
640    def do_ble_adv_data_include_tx_power_level(self, line):
641        """Include tx power level in the advertisement. inputs: [true|false]"""
642        cmd = "Include local name in the advertisement."
643        try:
644            self.pri_dut.ble.adv_data_include_tx_power_level(line)
645        except Exception as err:
646            self.log.info(FAILURE.format(cmd, err))
647
648    def do_ble_adv_data_add_manufacturer_data(self, line):
649        """Include manufacturer id and data to the advertisment:
650        [id data1 data2 ... dataN]"""
651        cmd = "Include manufacturer id and data to the advertisment."
652        try:
653            self.pri_dut.ble.adv_data_add_manufacturer_data(line)
654        except Exception as err:
655            self.log.info(FAILURE.format(cmd, err))
656
657    def do_ble_start_generic_nonconnectable_advertisement(self, line):
658        """Start a nonconnectable LE advertisement"""
659        cmd = "Start a nonconnectable LE advertisement"
660        try:
661            self.pri_dut.ble.start_generic_nonconnectable_advertisement(line)
662        except Exception as err:
663            self.log.info(FAILURE.format(cmd, err))
664
665    def do_ble_list_active_advertisement_ids(self, line):
666        """List all active BLE advertisements"""
667        self.log.info("IDs: {}".format(self.pri_dut.ble.advertisement_list))
668
669    def do_ble_stop_all_advertisements(self, line):
670        """Stop all LE advertisements"""
671        cmd = "Stop all LE advertisements"
672        try:
673            self.pri_dut.ble.stop_all_advertisements(line)
674        except Exception as err:
675            self.log.info(FAILURE.format(cmd, err))
676
677    def do_ble_stop_advertisement(self, line):
678        """Stop an LE advertisement"""
679        cmd = "Stop a connectable LE advertisement"
680        try:
681            self.pri_dut.ble.ble_stop_advertisement(line)
682        except Exception as err:
683            self.log.info(FAILURE.format(cmd, err))
684
685    """End Ble wrappers"""
686    """Begin Bta wrappers"""
687
688    def complete_bta_start_pairing_helper(self, text, line, begidx, endidx):
689        options = ['true', 'false']
690        if not text:
691            completions = list(options)[:]
692        else:
693            completions = [s for s in options if s.startswith(text)]
694        return completions
695
696    def complete_bta_set_scan_mode(self, text, line, begidx, endidx):
697        completions = list(bt_scan_mode_types.keys())[:]
698        if not text:
699            completions = completions[:]
700        else:
701            completions = [s for s in completions if s.startswith(text)]
702        return completions
703
704    def do_bta_set_scan_mode(self, line):
705        """Set the Scan mode of the Bluetooth Adapter"""
706        cmd = "Set the Scan mode of the Bluetooth Adapter"
707        try:
708            self.pri_dut.bta.set_scan_mode(line)
709        except Exception as err:
710            self.log.info(FAILURE.format(cmd, err))
711
712    def do_bta_set_device_name(self, line):
713        """Set Bluetooth Adapter Name"""
714        cmd = "Set Bluetooth Adapter Name"
715        try:
716            self.pri_dut.bta.set_device_name(line)
717        except Exception as err:
718            self.log.info(FAILURE.format(cmd, err))
719
720    def do_bta_enable(self, line):
721        """Enable Bluetooth Adapter"""
722        cmd = "Enable Bluetooth Adapter"
723        try:
724            self.pri_dut.bta.enable()
725        except Exception as err:
726            self.log.info(FAILURE.format(cmd, err))
727
728    def do_bta_disable(self, line):
729        """Disable Bluetooth Adapter"""
730        cmd = "Disable Bluetooth Adapter"
731        try:
732            self.pri_dut.bta.disable()
733        except Exception as err:
734            self.log.info(FAILURE.format(cmd, err))
735
736    def do_bta_init_bond(self, line):
737        """Initiate bond to PTS device"""
738        cmd = "Initiate Bond"
739        try:
740            self.pri_dut.bta.init_bond()
741        except Exception as err:
742            self.log.info(FAILURE.format(cmd, err))
743
744    def do_bta_start_discovery(self, line):
745        """Start BR/EDR Discovery"""
746        cmd = "Start BR/EDR Discovery"
747        try:
748            self.pri_dut.bta.start_discovery()
749        except Exception as err:
750            self.log.info(FAILURE.format(cmd, err))
751
752    def do_bta_stop_discovery(self, line):
753        """Stop BR/EDR Discovery"""
754        cmd = "Stop BR/EDR Discovery"
755        try:
756            self.pri_dut.bta.stop_discovery()
757        except Exception as err:
758            self.log.info(FAILURE.format(cmd, err))
759
760    def do_bta_get_discovered_devices(self, line):
761        """Get Discovered Br/EDR Devices"""
762        cmd = "Get Discovered Br/EDR Devices\n"
763        try:
764            self.pri_dut.bta.get_discovered_devices()
765        except Exception as err:
766            self.log.info(FAILURE.format(cmd, err))
767
768    def do_bta_bond(self, line):
769        """Bond to PTS device"""
770        cmd = "Bond to the PTS dongle directly"
771        try:
772            self.pri_dut.bta.bond()
773        except Exception as err:
774            self.log.info(FAILURE.format(cmd, err))
775
776    def do_bta_disconnect(self, line):
777        """BTA disconnect"""
778        cmd = "BTA disconnect"
779        try:
780            self.pri_dut.bta.disconnect()
781        except Exception as err:
782            self.log.info(FAILURE.format(cmd, err))
783
784    def do_bta_unbond(self, line):
785        """Unbond from PTS device"""
786        cmd = "Unbond from the PTS dongle"
787        try:
788            self.pri_dut.bta.unbond()
789        except Exception as err:
790            self.log.info(FAILURE.format(cmd, err))
791
792    def do_bta_start_pairing_helper(self, line):
793        """Start or stop Bluetooth Pairing Helper"""
794        cmd = "Start or stop BT Pairing helper"
795        try:
796            self.pri_dut.bta.start_pairing_helper(line)
797        except Exception as err:
798            self.log.info(FAILURE.format(cmd, err))
799
800    def do_bta_push_pairing_pin(self, line):
801        """Push pairing pin to the Android Device"""
802        cmd = "Push the pin to the Android Device"
803        try:
804            self.pri_dut.bta.push_pairing_pin(line)
805        except Exception as err:
806            self.log.info(FAILURE.format(cmd, err))
807
808    def do_bta_get_pairing_pin(self, line):
809        """Get pairing PIN"""
810        cmd = "Get Pin Info"
811        try:
812            self.pri_dut.bta.get_pairing_pin()
813        except Exception as err:
814            self.log.info(FAILURE.format(cmd, err))
815
816    def do_bta_fetch_uuids_with_sdp(self, line):
817        """BTA fetch UUIDS with SDP"""
818        cmd = "Fetch UUIDS with SDP"
819        try:
820            self.pri_dut.bta.fetch_uuids_with_sdp()
821        except Exception as err:
822            self.log.info(FAILURE.format(cmd, err))
823
824    def do_bta_connect_profiles(self, line):
825        """Connect available profiles"""
826        cmd = "Connect all profiles possible"
827        try:
828            self.pri_dut.bta.connect_profiles()
829        except Exception as err:
830            self.log.info(FAILURE.format(cmd, err))
831
832    def do_bta_tts_speak(self, line):
833        cmd = "Open audio channel by speaking characters"
834        try:
835            self.pri_dut.bta.tts_speak()
836        except Exception as err:
837            self.log.info(FAILURE.format(cmd, err))
838
839    """End Bta wrappers"""
840    """Begin Rfcomm wrappers"""
841
842    def do_rfcomm_connect(self, line):
843        """Perform an RFCOMM connect"""
844        try:
845            self.pri_dut.rfcomm.connect(line)
846        except Exception as err:
847            self.log.info(FAILURE.format(cmd, err))
848
849    def do_rfcomm_open_rfcomm_socket(self, line):
850        """Open rfcomm socket"""
851        cmd = "Open RFCOMM socket"
852        try:
853            self.pri_dut.rfcomm.open_rfcomm_socket()
854        except Exception as err:
855            self.log.info(FAILURE.format(cmd, err))
856
857    def do_rfcomm_open_l2cap_socket(self, line):
858        """Open L2CAP socket"""
859        cmd = "Open L2CAP socket"
860        try:
861            self.pri_dut.rfcomm.open_l2cap_socket()
862        except Exception as err:
863            self.log.info(FAILURE.format(cmd, err))
864
865    def do_rfcomm_write(self, line):
866        cmd = "Write String data over an RFCOMM connection"
867        try:
868            self.pri_dut.rfcomm.write(line)
869        except Exception as err:
870            self.log.info(FAILURE.format(cmd, err))
871
872    def do_rfcomm_write_binary(self, line):
873        cmd = "Write String data over an RFCOMM connection"
874        try:
875            self.pri_dut.rfcomm.write_binary(line)
876        except Exception as err:
877            self.log.info(FAILURE.format(cmd, err))
878
879    def do_rfcomm_end_connect(self, line):
880        cmd = "End RFCOMM connection"
881        try:
882            self.pri_dut.rfcomm.end_connect()
883        except Exception as err:
884            self.log.info(FAILURE.format(cmd, err))
885
886    def do_rfcomm_accept(self, line):
887        cmd = "Accept RFCOMM connection"
888        try:
889            self.pri_dut.rfcomm.accept(line)
890        except Exception as err:
891            self.log.info(FAILURE.format(cmd, err))
892
893    def do_rfcomm_stop(self, line):
894        cmd = "STOP RFCOMM Connection"
895        try:
896            self.pri_dut.rfcomm.stop()
897        except Exception as err:
898            self.log.info(FAILURE.format(cmd, err))
899
900    def do_rfcomm_open_l2cap_socket(self, line):
901        """Open L2CAP socket"""
902        cmd = "Open L2CAP socket"
903        try:
904            self.pri_dut.rfcomm.open_l2cap_socket()
905        except Exception as err:
906            self.log.info(FAILURE.format(cmd, err))
907
908    """End Rfcomm wrappers"""
909    """Begin Config wrappers"""
910
911    def do_config_reset(self, line):
912        """Reset Bluetooth Config file"""
913        cmd = "Reset Bluetooth Config file"
914        try:
915            self.pri_dut.config.reset()
916        except Exception as err:
917            self.log.info(FAILURE.format(cmd, err))
918
919    def do_config_set_nonbond(self, line):
920        """Set NonBondable Mode"""
921        cmd = "Set NonBondable Mode"
922        try:
923            self.pri_dut.config.set_nonbond()
924        except Exception as err:
925            self.log.info(FAILURE.format(cmd, err))
926
927    def do_config_set_disable_mitm(self, line):
928        """Set Disable MITM"""
929        cmd = "Set Disable MITM"
930        try:
931            self.pri_dut.config.set_disable_mitm()
932        except Exception as err:
933            self.log.info(FAILURE.format(cmd, err))
934
935    """End Config wrappers"""
936    """Begin HFP/HSP wrapper"""
937
938    def do_bta_hfp_start_voice_recognition(self, line):
939        self.pri_dut.droid.bluetoothHspStartVoiceRecognition(self.mac_addr)
940
941    def do_bta_hfp_stop_voice_recognition(self, line):
942        self.pri_dut.droid.bluetoothHspStopVoiceRecognition(self.mac_addr)
943
944    def do_test_clcc_response(self, line):
945        # Experimental
946        clcc_index = 1
947        clcc_direction = 1
948        clcc_status = 4
949        clcc_mode = 0
950        clcc_mpty = False
951        clcc_number = "18888888888"
952        clcc_type = 0
953        self.pri_dut.droid.bluetoothHspClccResponse(
954            clcc_index, clcc_direction, clcc_status, clcc_mode, clcc_mpty,
955            clcc_number, clcc_type)
956
957    def do_bta_hsp_force_sco_audio_on(self, line):
958        """HFP/HSP Force SCO Audio ON"""
959        cmd = "HFP/HSP Force SCO Audio ON"
960        try:
961            if not self.pri_dut.droid.bluetoothHspForceScoAudio(True):
962                self.log.info(
963                    FAILURE.format(cmd,
964                                   "bluetoothHspForceScoAudio returned false"))
965        except Exception as err:
966            self.log.info(FAILURE.format(cmd, err))
967
968    def do_bta_hsp_force_sco_audio_off(self, line):
969        """HFP/HSP Force SCO Audio OFF"""
970        cmd = "HFP/HSP Force SCO Audio OFF"
971        try:
972            if not self.pri_dut.droid.bluetoothHspForceScoAudio(False):
973                self.log.info(
974                    FAILURE.format(cmd,
975                                   "bluetoothHspForceScoAudio returned false"))
976        except Exception as err:
977            self.log.info(FAILURE.format(cmd, err))
978
979    def do_bta_hsp_connect_audio(self, line):
980        """HFP/HSP connect audio"""
981        cmd = "HFP/HSP connect audio"
982        try:
983            if not self.pri_dut.droid.bluetoothHspConnectAudio(self.mac_addr):
984                self.log.info(
985                    FAILURE.format(
986                        cmd, "bluetoothHspConnectAudio returned false for " +
987                        self.mac_addr))
988        except Exception as err:
989            self.log.info(FAILURE.format(cmd, err))
990
991    def do_bta_hsp_disconnect_audio(self, line):
992        """HFP/HSP disconnect audio"""
993        cmd = "HFP/HSP disconnect audio"
994        try:
995            if not self.pri_dut.droid.bluetoothHspDisconnectAudio(
996                    self.mac_addr):
997                self.log.info(
998                    FAILURE.format(
999                        cmd,
1000                        "bluetoothHspDisconnectAudio returned false for " +
1001                        self.mac_addr))
1002        except Exception as err:
1003            self.log.info(FAILURE.format(cmd, err))
1004
1005    def do_bta_hsp_connect_slc(self, line):
1006        """HFP/HSP connect SLC with additional tries and help"""
1007        cmd = "Connect to hsp with some help"
1008        try:
1009            if not self.connect_hsp_helper(self.pri_dut):
1010                self.log.error("Failed to connect to HSP")
1011        except Exception as err:
1012            self.log.info(FAILURE.format(cmd, err))
1013
1014    def do_bta_hsp_disconnect_slc(self, line):
1015        """HFP/HSP disconnect SLC"""
1016        cmd = "HFP/HSP disconnect SLC"
1017        try:
1018            if not self.pri_dut.droid.bluetoothHspDisconnect(self.mac_addr):
1019                self.log.info(
1020                    FAILURE.format(
1021                        cmd, "bluetoothHspDisconnect returned false for " +
1022                        self.mac_addr))
1023        except Exception as err:
1024            self.log.info(FAILURE.format(cmd, err))
1025
1026    """End HFP/HSP wrapper"""
1027    """Begin HID wrappers"""
1028
1029    def do_hid_get_report(self, line):
1030        """Get HID Report"""
1031        cmd = "Get HID Report"
1032        try:
1033            self.pri_dut.droid.bluetoothHidGetReport(self.mac_addr, 1, 1, 1024)
1034        except Exception as err:
1035            self.log.info(FAILURE.format(cmd, err))
1036
1037    def do_hid_set_report(self, line):
1038        """Get HID Report"""
1039        cmd = "Get HID Report"
1040        try:
1041            self.pri_dut.droid.bluetoothHidSetReport(self.mac_addr, 1, "Test")
1042        except Exception as err:
1043            self.log.info(FAILURE.format(cmd, err))
1044
1045    def do_hid_virtual_unplug(self, line):
1046        """Get HID Report"""
1047        cmd = "Get HID Report"
1048        try:
1049            self.pri_dut.droid.bluetoothHidVirtualUnplug(self.mac_addr)
1050        except Exception as err:
1051            self.log.info(FAILURE.format(cmd, err))
1052
1053    def do_hid_send_report(self, line):
1054        """Get HID Report"""
1055        cmd = "Get HID Report"
1056        try:
1057            self.pri_dut.droid.bluetoothHidSendData(self.mac_addr, "42")
1058        except Exception as err:
1059            self.log.info(FAILURE.format(cmd, err))
1060
1061    def do_hid_set_protocol_mode(self, line):
1062        """HID set protocol mode (0 == report, 1 == boot, 225 == unsupported)"""
1063        cmd = "Set protocol mode (0 == report, 1 == boot, 225 == unsupported)"
1064        try:
1065            self.pri_dut.droid.bluetoothHidSetProtocolMode(self.mac_addr, int(line))
1066        except Exception as err:
1067            self.log.info(FAILURE.format(cmd, err))
1068
1069    """End HID wrappers"""
1070    """Begin carkit test wrappers"""
1071
1072    def do_test_suite_generic_bt_tests(self, line):
1073        """Run generic Bluetooth connection test suite"""
1074        generic_bt_tests = [
1075            tuple((self.bt_carkit_lib.disconnect_reconnect_multiple_iterations,
1076                   [self.pri_dut])),
1077            tuple((self.bt_carkit_lib.disconnect_a2dp_only_then_reconnect,
1078                   [self.pri_dut])),
1079            tuple((self.bt_carkit_lib.disconnect_hsp_only_then_reconnect,
1080                   [self.pri_dut])),
1081            tuple((
1082                self.bt_carkit_lib.disconnect_both_hsp_and_a2dp_then_reconnect,
1083                [self.pri_dut])),
1084        ]
1085        try:
1086            for func, param in generic_bt_tests:
1087                try:
1088                    func(param)
1089                except Exception:
1090                    self.log.info("Test {} failed.".format(func))
1091        except Exception as err:
1092            self.log.info(FAILURE.format(cmd, err))
1093
1094    def do_e2e_connect_hsp_helper(self, line):
1095        """Connect to HSP/HFP with additional tries and help"""
1096        cmd = "Connect to hsp with some help"
1097        try:
1098            self.bt_carkit_lib.connect_hsp_helper(self.pri_dut)
1099        except Exception as err:
1100            self.log.info(FAILURE.format(cmd, err))
1101
1102    def do_e2e_hfp_setup_multi_incomming_calls(self, line):
1103        """Setup two incomming calls"""
1104        cmd = "Setup multiple incomming calls"
1105        try:
1106            self.bt_carkit_lib.setup_multi_call(self.sec_dut, self.ter_dut,
1107                                                self.pri_dut)
1108        except Exception as err:
1109            self.log.info(FAILURE.format(cmd, err))
1110
1111    def do_e2e_disconnect_reconnect_multiple_iterations(self, line):
1112        """Quick disconnect/reconnect stress test"""
1113        try:
1114            self.bt_carkit_lib.disconnect_reconnect_multiple_iterations(
1115                self.pri_dut)
1116        except Exception as err:
1117            self.log.info(FAILURE.format(cmd, err))
1118
1119    def do_e2e_disconnect_a2dp_only_then_reconnect(self, line):
1120        """Test disconnect-reconnect a2dp only scenario from phone."""
1121        cmd = "Test disconnect-reconnect a2dp only scenario from phone."
1122        try:
1123            self.bt_carkit_lib.disconnect_a2dp_only_then_reconnect(
1124                self.pri_dut)
1125        except Exception as err:
1126            self.log.info(FAILURE.format(cmd, err))
1127
1128    def do_e2e_disconnect_hsp_only_then_reconnect(self, line):
1129        """Test disconnect-reconnect hsp only scenario from phone."""
1130        cmd = "Test disconnect-reconnect hsp only scenario from phone."
1131        try:
1132            self.bt_carkit_lib.disconnect_hsp_only_then_reconnect(self.pri_dut)
1133        except Exception as err:
1134            self.log.info(FAILURE.format(cmd, err))
1135
1136    def do_e2e_disconnect_both_hsp_and_a2dp_then_reconnect(self, line):
1137        """Test disconnect-reconnect hsp and a2dp scenario from phone."""
1138        cmd = "Test disconnect-reconnect hsp and a2dp scenario from phone."
1139        try:
1140            self.bt_carkit_lib.disconnect_both_hsp_and_a2dp_then_reconnect(
1141                self.pri_dut)
1142        except Exception as err:
1143            self.log.info(FAILURE.format(cmd, err))
1144
1145    def do_e2e_outgoing_call_private_number(self, line):
1146        """Test outgoing call scenario from phone to private number"""
1147        cmd = "Test outgoing call scenario from phone to private number"
1148        try:
1149            self.bt_carkit_lib.outgoing_call_private_number(
1150                self.pri_dut, self.ter_dut)
1151        except Exception as err:
1152            self.log.info(FAILURE.format(cmd, err))
1153
1154    def do_e2e_outgoing_call_a2dp_play_before_and_after(self, line):
1155        """Test outgoing call scenario while playing music. Music should resume
1156        after call."""
1157        cmd = "Test outgoing call scenario while playing music. Music should " \
1158              "resume after call."
1159        try:
1160            self.bt_carkit_lib.outgoing_call_a2dp_play_before_and_after(
1161                self.pri_dut, self.sec_dut)
1162        except Exception as err:
1163            self.log.info(FAILURE.format(cmd, err))
1164
1165    def do_e2e_outgoing_call_unknown_contact(self, line):
1166        """Test outgoing call scenario from phone to unknow contact"""
1167        cmd = "Test outgoing call scenario from phone to unknow contact"
1168        try:
1169            self.bt_carkit_lib.outgoing_call_unknown_contact(
1170                self.pri_dut, self.ter_dut)
1171        except Exception as err:
1172            self.log.info(FAILURE.format(cmd, err))
1173
1174    def do_e2e_incomming_call_private_number(self, line):
1175        """Test incomming call scenario to phone from unknown contact"""
1176        cmd = "Test incomming call scenario to phone from unknown contact"
1177        try:
1178            self.bt_carkit_lib.incomming_call_private_number(
1179                self.pri_dut, self.ter_dut)
1180        except Exception as err:
1181            self.log.info(FAILURE.format(cmd, err))
1182
1183    def do_e2e_outgoing_call_multiple_iterations(self, line):
1184        """Test outgoing call quickly 3 times"""
1185        cmd = "Test outgoing call quickly 3 times"
1186        try:
1187            self.bt_carkit_lib.outgoing_call_multiple_iterations(
1188                self.pri_dut, self.sec_dut)
1189        except Exception as err:
1190            self.log.info(FAILURE.format(cmd, err))
1191
1192    def do_e2e_outgoing_call_hsp_disabled_then_enabled_during_call(self, line):
1193        """Test outgoing call hsp disabled then enable during call."""
1194        cmd = "Test outgoing call hsp disabled then enable during call."
1195        try:
1196            self.bt_carkit_lib.outgoing_call_hsp_disabled_then_enabled_during_call(
1197                self.pri_dut, self.sec_dut)
1198        except Exception as err:
1199            self.log.info(FAILURE.format(cmd, err))
1200
1201    def do_e2e_call_audio_routes(self, line):
1202        """Test various audio routes scenario from phone."""
1203        cmd = "Test various audio routes scenario from phone."
1204        try:
1205            self.bt_carkit_lib.call_audio_routes(self.pri_dut, self.sec_dut)
1206        except Exception as err:
1207            self.log.info(FAILURE.format(cmd, err))
1208
1209    def do_e2e_sms_receive_different_sizes(self, line):
1210        """Test recieve sms of different sizes."""
1211        cmd = "Test recieve sms of different sizes."
1212        try:
1213            self.bt_carkit_lib.sms_receive_different_sizes(
1214                self.pri_dut, self.sec_dut)
1215        except Exception as err:
1216            self.log.info(FAILURE.format(cmd, err))
1217
1218    def do_e2e_sms_receive_multiple(self, line):
1219        """Test recieve sms of different sizes."""
1220        cmd = "Test recieve sms of different sizes."
1221        try:
1222            self.bt_carkit_lib.sms_receive_multiple(self.pri_dut, self.sec_dut)
1223        except Exception as err:
1224            self.log.info(FAILURE.format(cmd, err))
1225
1226    def do_e2e_sms_send_outgoing_texts(self, line):
1227        """Test send sms of different sizes."""
1228        cmd = "Test send sms of different sizes."
1229        try:
1230            self.bt_carkit_lib.sms_send_outgoing_texts(self.pri_dut,
1231                                                       self.sec_dut)
1232        except Exception as err:
1233            self.log.info(FAILURE.format(cmd, err))
1234
1235    def do_e2e_sms_during_incomming_call(self, line):
1236        """Test incomming call scenario to phone from unknown contact"""
1237        cmd = "Test incomming call scenario to phone from unknown contact"
1238        try:
1239            self.bt_carkit_lib.sms_during_incomming_call(
1240                self.pri_dut, self.sec_dut)
1241        except Exception as err:
1242            self.log.info(FAILURE.format(cmd, err))
1243
1244    def do_e2e_multi_incomming_call(self, line):
1245        """Test 2 incomming calls scenario to phone."""
1246        cmd = "Test 2 incomming calls scenario to phone."
1247        try:
1248            self.bt_carkit_lib.multi_incomming_call(self.pri_dut, self.sec_dut,
1249                                                    self.ter_dut)
1250        except Exception as err:
1251            self.log.info(FAILURE.format(cmd, err))
1252
1253    def do_e2e_multi_call_audio_routing(self, line):
1254        """Test 2 incomming calls scenario to phone, then test audio routing."""
1255        cmd = "Test 2 incomming calls scenario to phone, then test audio" \
1256            "routing."
1257        try:
1258            self.bt_carkit_lib.multi_call_audio_routing(
1259                self.pri_dut, self.sec_dut, self.ter_dut)
1260        except Exception as err:
1261            self.log.info(FAILURE.format(cmd, err))
1262
1263    def do_e2e_multi_call_swap_multiple_times(self, line):
1264        """Test 2 incomming calls scenario to phone, then swap the calls
1265        multiple times"""
1266        cmd = "Test 2 incomming calls scenario to phone, then swap the calls" \
1267            "multiple times"
1268        try:
1269            self.bt_carkit_lib.multi_call_swap_multiple_times(
1270                self.pri_dut, self.sec_dut, self.ter_dut)
1271        except Exception as err:
1272            self.log.info(FAILURE.format(cmd, err))
1273
1274    def do_e2e_multi_call_join_conference_call(self, line):
1275        """Test 2 incomming calls scenario to phone then join the calls."""
1276        cmd = "Test 2 incomming calls scenario to phone then join the calls."
1277        try:
1278            self.bt_carkit_lib.multi_call_join_conference_call(
1279                self.pri_dut, self.sec_dut, self.ter_dut)
1280        except Exception as err:
1281            self.log.info(FAILURE.format(cmd, err))
1282
1283    def do_e2e_multi_call_join_conference_call_hangup_conf_call(self, line):
1284        """Test 2 incomming calls scenario to phone then join the calls,
1285        then terminate the call from the primary dut."""
1286        cmd = "Test 2 incomming calls scenario to phone then join the calls, " \
1287            "then terminate the call from the primary dut."
1288        try:
1289            self.bt_carkit_lib.multi_call_join_conference_call_hangup_conf_call(
1290                self.pri_dut, self.sec_dut, self.ter_dut)
1291        except Exception as err:
1292            self.log.info(FAILURE.format(cmd, err))
1293
1294    def do_e2e_outgoing_multi_call_join_conference_call(self, line):
1295        """Test 2 outgoing calls scenario from phone then join the calls."""
1296        cmd = "Test 2 outgoing calls scenario from phone then join the calls."
1297        try:
1298            self.bt_carkit_lib.outgoing_multi_call_join_conference_call(
1299                self.pri_dut, self.sec_dut, self.ter_dut)
1300        except Exception as err:
1301            self.log.info(FAILURE.format(cmd, err))
1302
1303    def do_e2e_multi_call_join_conference_call_audio_routes(self, line):
1304        """Test 2 incomming calls scenario to phone then join the calls,
1305        then test different audio routes."""
1306        cmd = "Test 2 incomming calls scenario to phone then join the calls, " \
1307            "then test different audio routes."
1308        try:
1309            self.bt_carkit_lib.multi_call_join_conference_call_audio_routes(
1310                self.pri_dut, self.sec_dut, self.ter_dut)
1311        except Exception as err:
1312            self.log.info(FAILURE.format(cmd, err))
1313
1314    def do_e2e_avrcp_play_pause(self, line):
1315        """Test avrcp play/pause commands multiple times from phone"""
1316        cmd = "Test avrcp play/pause commands multiple times from phone"
1317        try:
1318            self.bt_carkit_lib.avrcp_play_pause(self.pri_dut)
1319        except Exception as err:
1320            self.log.info(FAILURE.format(cmd, err))
1321
1322    def do_e2e_avrcp_next_previous_song(self, line):
1323        """Test AVRCP go to the next song then the previous song."""
1324        cmd = "Test AVRCP go to the next song then the previous song."
1325        try:
1326            self.bt_carkit_lib.avrcp_next_previous_song(self.pri_dut)
1327        except Exception as err:
1328            self.log.info(FAILURE.format(cmd, err))
1329
1330    def do_e2e_avrcp_next_previous(self, line):
1331        """Test AVRCP go to the next song then the press previous after a few
1332        seconds."""
1333        cmd = "Test AVRCP go to the next song then the press previous after " \
1334            "a few seconds."
1335        try:
1336            self.bt_carkit_lib.avrcp_next_previous(self.pri_dut)
1337        except Exception as err:
1338            self.log.info(FAILURE.format(cmd, err))
1339
1340    def do_e2e_avrcp_next_repetative(self, line):
1341        """Test AVRCP go to the next 10 times"""
1342        cmd = "Test AVRCP go to the next 10 times"
1343        try:
1344            self.bt_carkit_lib.avrcp_next_repetative(self.pri_dut)
1345        except Exception as err:
1346            self.log.info(FAILURE.format(cmd, err))
1347
1348    def _process_question(self, question, expected_response):
1349        while True:
1350            try:
1351                result = input(question.format(level)).lower()
1352            except Exception as err:
1353                print(err)
1354
1355    def do_e2e_cycle_battery_level(self, line):
1356        """Cycle battery level through different values and verify result on carkit"""
1357        cmd = "Test that verifies battery level indicator changes with the " \
1358            "phone. Phone current level."
1359        try:
1360            self.bt_carkit_lib.cycle_battery_level(self.pri_dut)
1361        except Exception as err:
1362            self.log.info(FAILURE.format(cmd, err))
1363
1364    def do_e2e_cycle_absolute_volume_control(self, line):
1365        """Cycle media volume level through different values and verify result on carkit"""
1366        cmd = "Test aboslute volume on carkit by changed volume levels from phone."
1367        try:
1368            self.bt_carkit_lib.cycle_absolute_volume_control(self.pri_dut)
1369        except Exception as err:
1370            self.log.info(FAILURE.format(cmd, err))
1371
1372    def do_e2e_test_voice_recognition_from_phone(self, line):
1373        """Test Voice Recognition from phone."""
1374        cmd = "Test voice recognition from phone."
1375        try:
1376            self.bt_carkit_lib.test_voice_recognition_from_phone(self.pri_dut)
1377        except Exception as err:
1378            self.log.info(FAILURE.format(cmd, err))
1379
1380    def do_e2e_test_audio_and_voice_recognition_from_phone(self, line):
1381        """Test Voice Recognition from phone and confirm music audio continues."""
1382        cmd = "Test Voice Recognition from phone and confirm music audio continues."
1383        try:
1384            self.bt_carkit_lib.test_audio_and_voice_recognition_from_phone(
1385                self.pri_dut)
1386        except Exception as err:
1387            self.log.info(FAILURE.format(cmd, err))
1388
1389    """End carkit test wrappers"""
1390    """Begin adb shell test wrappers"""
1391
1392    def do_set_battery_level(self, line):
1393        """Set battery level based on input"""
1394        cmd = "Set battery level based on input"
1395        try:
1396            self.pri_dut.shell.set_battery_level(int(line))
1397        except Exception as err:
1398            self.log.info(FAILURE.format(cmd, err))
1399
1400    """End adb shell test wrappers"""
1401