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_contrib.test_utils.bt.bt_constants import bt_scan_mode_types 20from acts_contrib.test_utils.bt.bt_constants import gatt_server_responses 21import acts_contrib.test_utils.bt.gatt_test_database as gatt_test_database 22from acts_contrib.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 def do_bta_hfp_client_connect(self, line): 1027 self.pri_dut.droid.bluetoothHfpClientConnect(self.mac_addr) 1028 1029 def do_bta_hfp_client_disconnect(self, line): 1030 self.pri_dut.droid.bluetoothHfpClientDisconnect(self.mac_addr) 1031 1032 def do_bta_hfp_client_connect_audio(self, line): 1033 self.pri_dut.droid.bluetoothHfpClientConnectAudio(self.mac_addr) 1034 1035 def do_bta_hfp_client_disconnect_audio(self, line): 1036 self.pri_dut.droid.bluetoothHfpClientDisconnectAudio(self.mac_addr) 1037 1038 """End HFP/HSP wrapper""" 1039 """Begin HID wrappers""" 1040 1041 def do_hid_get_report(self, line): 1042 """Get HID Report""" 1043 cmd = "Get HID Report" 1044 try: 1045 self.pri_dut.droid.bluetoothHidGetReport(self.mac_addr, 1, 1, 1024) 1046 except Exception as err: 1047 self.log.info(FAILURE.format(cmd, err)) 1048 1049 def do_hid_set_report(self, line): 1050 """Get HID Report""" 1051 cmd = "Get HID Report" 1052 try: 1053 self.pri_dut.droid.bluetoothHidSetReport(self.mac_addr, 1, "Test") 1054 except Exception as err: 1055 self.log.info(FAILURE.format(cmd, err)) 1056 1057 def do_hid_virtual_unplug(self, line): 1058 """Get HID Report""" 1059 cmd = "Get HID Report" 1060 try: 1061 self.pri_dut.droid.bluetoothHidVirtualUnplug(self.mac_addr) 1062 except Exception as err: 1063 self.log.info(FAILURE.format(cmd, err)) 1064 1065 def do_hid_send_report(self, line): 1066 """Get HID Report""" 1067 cmd = "Get HID Report" 1068 try: 1069 self.pri_dut.droid.bluetoothHidSendData(self.mac_addr, "42") 1070 except Exception as err: 1071 self.log.info(FAILURE.format(cmd, err)) 1072 1073 def do_hid_set_protocol_mode(self, line): 1074 """HID set protocol mode (0 == report, 1 == boot, 225 == unsupported)""" 1075 cmd = "Set protocol mode (0 == report, 1 == boot, 225 == unsupported)" 1076 try: 1077 self.pri_dut.droid.bluetoothHidSetProtocolMode(self.mac_addr, int(line)) 1078 except Exception as err: 1079 self.log.info(FAILURE.format(cmd, err)) 1080 1081 """End HID wrappers""" 1082 """Begin carkit test wrappers""" 1083 1084 def do_test_suite_generic_bt_tests(self, line): 1085 """Run generic Bluetooth connection test suite""" 1086 generic_bt_tests = [ 1087 tuple((self.bt_carkit_lib.disconnect_reconnect_multiple_iterations, 1088 [self.pri_dut])), 1089 tuple((self.bt_carkit_lib.disconnect_a2dp_only_then_reconnect, 1090 [self.pri_dut])), 1091 tuple((self.bt_carkit_lib.disconnect_hsp_only_then_reconnect, 1092 [self.pri_dut])), 1093 tuple(( 1094 self.bt_carkit_lib.disconnect_both_hsp_and_a2dp_then_reconnect, 1095 [self.pri_dut])), 1096 ] 1097 try: 1098 for func, param in generic_bt_tests: 1099 try: 1100 func(param) 1101 except Exception: 1102 self.log.info("Test {} failed.".format(func)) 1103 except Exception as err: 1104 self.log.info(FAILURE.format(cmd, err)) 1105 1106 def do_e2e_connect_hsp_helper(self, line): 1107 """Connect to HSP/HFP with additional tries and help""" 1108 cmd = "Connect to hsp with some help" 1109 try: 1110 self.bt_carkit_lib.connect_hsp_helper(self.pri_dut) 1111 except Exception as err: 1112 self.log.info(FAILURE.format(cmd, err)) 1113 1114 def do_e2e_hfp_setup_multi_incomming_calls(self, line): 1115 """Setup two incomming calls""" 1116 cmd = "Setup multiple incomming calls" 1117 try: 1118 self.bt_carkit_lib.setup_multi_call(self.sec_dut, self.ter_dut, 1119 self.pri_dut) 1120 except Exception as err: 1121 self.log.info(FAILURE.format(cmd, err)) 1122 1123 def do_e2e_disconnect_reconnect_multiple_iterations(self, line): 1124 """Quick disconnect/reconnect stress test""" 1125 try: 1126 self.bt_carkit_lib.disconnect_reconnect_multiple_iterations( 1127 self.pri_dut) 1128 except Exception as err: 1129 self.log.info(FAILURE.format(cmd, err)) 1130 1131 def do_e2e_disconnect_a2dp_only_then_reconnect(self, line): 1132 """Test disconnect-reconnect a2dp only scenario from phone.""" 1133 cmd = "Test disconnect-reconnect a2dp only scenario from phone." 1134 try: 1135 self.bt_carkit_lib.disconnect_a2dp_only_then_reconnect( 1136 self.pri_dut) 1137 except Exception as err: 1138 self.log.info(FAILURE.format(cmd, err)) 1139 1140 def do_e2e_disconnect_hsp_only_then_reconnect(self, line): 1141 """Test disconnect-reconnect hsp only scenario from phone.""" 1142 cmd = "Test disconnect-reconnect hsp only scenario from phone." 1143 try: 1144 self.bt_carkit_lib.disconnect_hsp_only_then_reconnect(self.pri_dut) 1145 except Exception as err: 1146 self.log.info(FAILURE.format(cmd, err)) 1147 1148 def do_e2e_disconnect_both_hsp_and_a2dp_then_reconnect(self, line): 1149 """Test disconnect-reconnect hsp and a2dp scenario from phone.""" 1150 cmd = "Test disconnect-reconnect hsp and a2dp scenario from phone." 1151 try: 1152 self.bt_carkit_lib.disconnect_both_hsp_and_a2dp_then_reconnect( 1153 self.pri_dut) 1154 except Exception as err: 1155 self.log.info(FAILURE.format(cmd, err)) 1156 1157 def do_e2e_outgoing_call_private_number(self, line): 1158 """Test outgoing call scenario from phone to private number""" 1159 cmd = "Test outgoing call scenario from phone to private number" 1160 try: 1161 self.bt_carkit_lib.outgoing_call_private_number( 1162 self.pri_dut, self.ter_dut) 1163 except Exception as err: 1164 self.log.info(FAILURE.format(cmd, err)) 1165 1166 def do_e2e_outgoing_call_a2dp_play_before_and_after(self, line): 1167 """Test outgoing call scenario while playing music. Music should resume 1168 after call.""" 1169 cmd = "Test outgoing call scenario while playing music. Music should " \ 1170 "resume after call." 1171 try: 1172 self.bt_carkit_lib.outgoing_call_a2dp_play_before_and_after( 1173 self.pri_dut, self.sec_dut) 1174 except Exception as err: 1175 self.log.info(FAILURE.format(cmd, err)) 1176 1177 def do_e2e_outgoing_call_unknown_contact(self, line): 1178 """Test outgoing call scenario from phone to unknow contact""" 1179 cmd = "Test outgoing call scenario from phone to unknow contact" 1180 try: 1181 self.bt_carkit_lib.outgoing_call_unknown_contact( 1182 self.pri_dut, self.ter_dut) 1183 except Exception as err: 1184 self.log.info(FAILURE.format(cmd, err)) 1185 1186 def do_e2e_incomming_call_private_number(self, line): 1187 """Test incomming call scenario to phone from unknown contact""" 1188 cmd = "Test incomming call scenario to phone from unknown contact" 1189 try: 1190 self.bt_carkit_lib.incomming_call_private_number( 1191 self.pri_dut, self.ter_dut) 1192 except Exception as err: 1193 self.log.info(FAILURE.format(cmd, err)) 1194 1195 def do_e2e_outgoing_call_multiple_iterations(self, line): 1196 """Test outgoing call quickly 3 times""" 1197 cmd = "Test outgoing call quickly 3 times" 1198 try: 1199 self.bt_carkit_lib.outgoing_call_multiple_iterations( 1200 self.pri_dut, self.sec_dut) 1201 except Exception as err: 1202 self.log.info(FAILURE.format(cmd, err)) 1203 1204 def do_e2e_outgoing_call_hsp_disabled_then_enabled_during_call(self, line): 1205 """Test outgoing call hsp disabled then enable during call.""" 1206 cmd = "Test outgoing call hsp disabled then enable during call." 1207 try: 1208 self.bt_carkit_lib.outgoing_call_hsp_disabled_then_enabled_during_call( 1209 self.pri_dut, self.sec_dut) 1210 except Exception as err: 1211 self.log.info(FAILURE.format(cmd, err)) 1212 1213 def do_e2e_call_audio_routes(self, line): 1214 """Test various audio routes scenario from phone.""" 1215 cmd = "Test various audio routes scenario from phone." 1216 try: 1217 self.bt_carkit_lib.call_audio_routes(self.pri_dut, self.sec_dut) 1218 except Exception as err: 1219 self.log.info(FAILURE.format(cmd, err)) 1220 1221 def do_e2e_sms_receive_different_sizes(self, line): 1222 """Test recieve sms of different sizes.""" 1223 cmd = "Test recieve sms of different sizes." 1224 try: 1225 self.bt_carkit_lib.sms_receive_different_sizes( 1226 self.pri_dut, self.sec_dut) 1227 except Exception as err: 1228 self.log.info(FAILURE.format(cmd, err)) 1229 1230 def do_e2e_sms_receive_multiple(self, line): 1231 """Test recieve sms of different sizes.""" 1232 cmd = "Test recieve sms of different sizes." 1233 try: 1234 self.bt_carkit_lib.sms_receive_multiple(self.pri_dut, self.sec_dut) 1235 except Exception as err: 1236 self.log.info(FAILURE.format(cmd, err)) 1237 1238 def do_e2e_sms_send_outgoing_texts(self, line): 1239 """Test send sms of different sizes.""" 1240 cmd = "Test send sms of different sizes." 1241 try: 1242 self.bt_carkit_lib.sms_send_outgoing_texts(self.pri_dut, 1243 self.sec_dut) 1244 except Exception as err: 1245 self.log.info(FAILURE.format(cmd, err)) 1246 1247 def do_e2e_sms_during_incomming_call(self, line): 1248 """Test incomming call scenario to phone from unknown contact""" 1249 cmd = "Test incomming call scenario to phone from unknown contact" 1250 try: 1251 self.bt_carkit_lib.sms_during_incomming_call( 1252 self.pri_dut, self.sec_dut) 1253 except Exception as err: 1254 self.log.info(FAILURE.format(cmd, err)) 1255 1256 def do_e2e_multi_incomming_call(self, line): 1257 """Test 2 incomming calls scenario to phone.""" 1258 cmd = "Test 2 incomming calls scenario to phone." 1259 try: 1260 self.bt_carkit_lib.multi_incomming_call(self.pri_dut, self.sec_dut, 1261 self.ter_dut) 1262 except Exception as err: 1263 self.log.info(FAILURE.format(cmd, err)) 1264 1265 def do_e2e_multi_call_audio_routing(self, line): 1266 """Test 2 incomming calls scenario to phone, then test audio routing.""" 1267 cmd = "Test 2 incomming calls scenario to phone, then test audio" \ 1268 "routing." 1269 try: 1270 self.bt_carkit_lib.multi_call_audio_routing( 1271 self.pri_dut, self.sec_dut, self.ter_dut) 1272 except Exception as err: 1273 self.log.info(FAILURE.format(cmd, err)) 1274 1275 def do_e2e_multi_call_swap_multiple_times(self, line): 1276 """Test 2 incomming calls scenario to phone, then swap the calls 1277 multiple times""" 1278 cmd = "Test 2 incomming calls scenario to phone, then swap the calls" \ 1279 "multiple times" 1280 try: 1281 self.bt_carkit_lib.multi_call_swap_multiple_times( 1282 self.pri_dut, self.sec_dut, self.ter_dut) 1283 except Exception as err: 1284 self.log.info(FAILURE.format(cmd, err)) 1285 1286 def do_e2e_multi_call_join_conference_call(self, line): 1287 """Test 2 incomming calls scenario to phone then join the calls.""" 1288 cmd = "Test 2 incomming calls scenario to phone then join the calls." 1289 try: 1290 self.bt_carkit_lib.multi_call_join_conference_call( 1291 self.pri_dut, self.sec_dut, self.ter_dut) 1292 except Exception as err: 1293 self.log.info(FAILURE.format(cmd, err)) 1294 1295 def do_e2e_multi_call_join_conference_call_hangup_conf_call(self, line): 1296 """Test 2 incomming calls scenario to phone then join the calls, 1297 then terminate the call from the primary dut.""" 1298 cmd = "Test 2 incomming calls scenario to phone then join the calls, " \ 1299 "then terminate the call from the primary dut." 1300 try: 1301 self.bt_carkit_lib.multi_call_join_conference_call_hangup_conf_call( 1302 self.pri_dut, self.sec_dut, self.ter_dut) 1303 except Exception as err: 1304 self.log.info(FAILURE.format(cmd, err)) 1305 1306 def do_e2e_outgoing_multi_call_join_conference_call(self, line): 1307 """Test 2 outgoing calls scenario from phone then join the calls.""" 1308 cmd = "Test 2 outgoing calls scenario from phone then join the calls." 1309 try: 1310 self.bt_carkit_lib.outgoing_multi_call_join_conference_call( 1311 self.pri_dut, self.sec_dut, self.ter_dut) 1312 except Exception as err: 1313 self.log.info(FAILURE.format(cmd, err)) 1314 1315 def do_e2e_multi_call_join_conference_call_audio_routes(self, line): 1316 """Test 2 incomming calls scenario to phone then join the calls, 1317 then test different audio routes.""" 1318 cmd = "Test 2 incomming calls scenario to phone then join the calls, " \ 1319 "then test different audio routes." 1320 try: 1321 self.bt_carkit_lib.multi_call_join_conference_call_audio_routes( 1322 self.pri_dut, self.sec_dut, self.ter_dut) 1323 except Exception as err: 1324 self.log.info(FAILURE.format(cmd, err)) 1325 1326 def do_e2e_avrcp_play_pause(self, line): 1327 """Test avrcp play/pause commands multiple times from phone""" 1328 cmd = "Test avrcp play/pause commands multiple times from phone" 1329 try: 1330 self.bt_carkit_lib.avrcp_play_pause(self.pri_dut) 1331 except Exception as err: 1332 self.log.info(FAILURE.format(cmd, err)) 1333 1334 def do_e2e_avrcp_next_previous_song(self, line): 1335 """Test AVRCP go to the next song then the previous song.""" 1336 cmd = "Test AVRCP go to the next song then the previous song." 1337 try: 1338 self.bt_carkit_lib.avrcp_next_previous_song(self.pri_dut) 1339 except Exception as err: 1340 self.log.info(FAILURE.format(cmd, err)) 1341 1342 def do_e2e_avrcp_next_previous(self, line): 1343 """Test AVRCP go to the next song then the press previous after a few 1344 seconds.""" 1345 cmd = "Test AVRCP go to the next song then the press previous after " \ 1346 "a few seconds." 1347 try: 1348 self.bt_carkit_lib.avrcp_next_previous(self.pri_dut) 1349 except Exception as err: 1350 self.log.info(FAILURE.format(cmd, err)) 1351 1352 def do_e2e_avrcp_next_repetative(self, line): 1353 """Test AVRCP go to the next 10 times""" 1354 cmd = "Test AVRCP go to the next 10 times" 1355 try: 1356 self.bt_carkit_lib.avrcp_next_repetative(self.pri_dut) 1357 except Exception as err: 1358 self.log.info(FAILURE.format(cmd, err)) 1359 1360 def _process_question(self, question, expected_response): 1361 while True: 1362 try: 1363 result = input(question.format(level)).lower() 1364 except Exception as err: 1365 print(err) 1366 1367 def do_e2e_cycle_battery_level(self, line): 1368 """Cycle battery level through different values and verify result on carkit""" 1369 cmd = "Test that verifies battery level indicator changes with the " \ 1370 "phone. Phone current level." 1371 try: 1372 self.bt_carkit_lib.cycle_battery_level(self.pri_dut) 1373 except Exception as err: 1374 self.log.info(FAILURE.format(cmd, err)) 1375 1376 def do_e2e_cycle_absolute_volume_control(self, line): 1377 """Cycle media volume level through different values and verify result on carkit""" 1378 cmd = "Test aboslute volume on carkit by changed volume levels from phone." 1379 try: 1380 self.bt_carkit_lib.cycle_absolute_volume_control(self.pri_dut) 1381 except Exception as err: 1382 self.log.info(FAILURE.format(cmd, err)) 1383 1384 def do_e2e_test_voice_recognition_from_phone(self, line): 1385 """Test Voice Recognition from phone.""" 1386 cmd = "Test voice recognition from phone." 1387 try: 1388 self.bt_carkit_lib.test_voice_recognition_from_phone(self.pri_dut) 1389 except Exception as err: 1390 self.log.info(FAILURE.format(cmd, err)) 1391 1392 def do_e2e_test_audio_and_voice_recognition_from_phone(self, line): 1393 """Test Voice Recognition from phone and confirm music audio continues.""" 1394 cmd = "Test Voice Recognition from phone and confirm music audio continues." 1395 try: 1396 self.bt_carkit_lib.test_audio_and_voice_recognition_from_phone( 1397 self.pri_dut) 1398 except Exception as err: 1399 self.log.info(FAILURE.format(cmd, err)) 1400 1401 """End carkit test wrappers""" 1402 """Begin adb shell test wrappers""" 1403 1404 def do_set_battery_level(self, line): 1405 """Set battery level based on input""" 1406 cmd = "Set battery level based on input" 1407 try: 1408 self.pri_dut.shell.set_battery_level(int(line)) 1409 except Exception as err: 1410 self.log.info(FAILURE.format(cmd, err)) 1411 1412 """End adb shell test wrappers""" 1413