| #/usr/bin/env python3.4 |
| # |
| # Copyright (C) 2016 The Android Open Source Project |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); you may not |
| # use this file except in compliance with the License. You may obtain a copy of |
| # the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT |
| # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the |
| # License for the specific language governing permissions and limitations under |
| # the License. |
| """ |
| Python script for wrappers to various libraries. |
| """ |
| from acts.test_utils.bt.bt_constants import bt_scan_mode_types |
| from acts.test_utils.bt.bt_constants import gatt_server_responses |
| import acts.test_utils.bt.gatt_test_database as gatt_test_database |
| from acts.test_utils.bt.bt_carkit_lib import E2eBtCarkitLib |
| |
| import threading |
| import time |
| import cmd |
| """Various Global Strings""" |
| CMD_LOG = "CMD {} result: {}" |
| FAILURE = "CMD {} threw exception: {}" |
| |
| |
| class CmdInput(cmd.Cmd): |
| android_devices = [] |
| """Simple command processor for Bluetooth PTS Testing""" |
| |
| def connect_hsp_helper(self, ad): |
| """A helper function for making HSP connections""" |
| end_time = time.time() + 20 |
| connected_hsp_devices = len(ad.droid.bluetoothHspGetConnectedDevices()) |
| while connected_hsp_devices != 1 and time.time() < end_time: |
| try: |
| ad.droid.bluetoothHspConnect(self.mac_addr) |
| time.sleep(3) |
| if len(ad.droid.bluetoothHspGetConnectedDevices() == 1): |
| break |
| except Exception: |
| self.log.debug("Failed to connect hsp trying again...") |
| try: |
| ad.droid.bluetoothConnectBonded(self.mac_addr) |
| except Exception: |
| self.log.info("Failed to connect to bonded device...") |
| connected_hsp_devices = len( |
| ad.droid.bluetoothHspGetConnectedDevices()) |
| if connected_hsp_devices != 1: |
| self.log.error("Failed to reconnect to HSP service...") |
| return False |
| self.log.info("Connected to HSP service...") |
| return True |
| |
| def setup_vars(self, android_devices, mac_addr, log): |
| self.pri_dut = android_devices[0] |
| if len(android_devices) > 1: |
| self.sec_dut = android_devices[1] |
| if len(android_devices) > 2: |
| self.ter_dut = android_devices[2] |
| if len(android_devices) > 3: |
| self.qua_dut = android_devices[3] |
| self.mac_addr = mac_addr |
| self.log = log |
| self.pri_dut.bta.set_target_mac_addr(mac_addr) |
| self.pri_dut.gattc.set_target_mac_addr(mac_addr) |
| self.pri_dut.rfcomm.set_target_mac_addr(mac_addr) |
| self.bt_carkit_lib = E2eBtCarkitLib(self.log, mac_addr) |
| |
| def emptyline(self): |
| pass |
| |
| def do_EOF(self, line): |
| "End Script" |
| return True |
| |
| def do_reset_mac_address(self, line): |
| """Reset target mac address for libraries based on the current connected device""" |
| try: |
| device = self.pri_dut.droid.bluetoothGetConnectedDevices()[0] |
| #self.setup_vars(self.android_devices, device['address'], self.log) |
| self.log.info("New device is {}".format(device)) |
| except Exception as err: |
| self.log.info("Failed to setup new vars with {}".format(err)) |
| |
| """Begin GATT Client wrappers""" |
| |
| def do_gattc_connect_over_le(self, line): |
| """Perform GATT connection over LE""" |
| cmd = "Gatt connect over LE" |
| try: |
| autoconnect = False |
| if line: |
| autoconnect = bool(line) |
| self.pri_dut.gattc.connect_over_le(autoconnect) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_connect_over_bredr(self, line): |
| """Perform GATT connection over BREDR""" |
| cmd = "Gatt connect over BR/EDR" |
| try: |
| self.pri_dut.gattc.connect_over_bredr() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_disconnect(self, line): |
| """Perform GATT disconnect""" |
| cmd = "Gatt Disconnect" |
| try: |
| self.pri_dut.gattc.disconnect() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_read_char_by_uuid(self, line): |
| """GATT client read Characteristic by UUID.""" |
| cmd = "GATT client read Characteristic by UUID." |
| try: |
| self.pri_dut.gattc.read_char_by_uuid(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_request_mtu(self, line): |
| """Request MTU Change of input value""" |
| cmd = "Request MTU Value" |
| try: |
| self.pri_dut.gattc.request_mtu(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_list_all_uuids(self, line): |
| """From the GATT Client, discover services and list all services, |
| chars and descriptors |
| """ |
| cmd = "Discovery Services and list all UUIDS" |
| try: |
| self.pri_dut.gattc.list_all_uuids() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_discover_services(self, line): |
| """GATT Client discover services of GATT Server""" |
| cmd = "Discovery Services of GATT Server" |
| try: |
| self.pri_dut.gattc.discover_services() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_refresh(self, line): |
| """Perform Gatt Client Refresh""" |
| cmd = "GATT Client Refresh" |
| try: |
| self.pri_dut.gattc.refresh() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_read_char_by_instance_id(self, line): |
| """From the GATT Client, discover services and list all services, |
| chars and descriptors |
| """ |
| cmd = "GATT Client Read By Instance ID" |
| try: |
| self.pri_dut.gattc.read_char_by_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_write_char_by_instance_id(self, line): |
| """GATT Client Write to Characteristic by instance ID""" |
| cmd = "GATT Client write to Characteristic by instance ID" |
| try: |
| self.pri_dut.gattc.write_char_by_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_write_char_by_instance_id_value(self, line): |
| """GATT Client Write to Characteristic by instance ID""" |
| cmd = "GATT Client write to Characteristic by instance ID" |
| try: |
| self.pri_dut.gattc.write_char_by_instance_id_value(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_mod_write_char_by_instance_id(self, line): |
| """GATT Client Write to Char that doesn't have write permission""" |
| cmd = "GATT Client Write to Char that doesn't have write permission" |
| try: |
| self.pri_dut.gattc.mod_write_char_by_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_write_invalid_char_by_instance_id(self, line): |
| """GATT Client Write to Char that doesn't exists""" |
| try: |
| self.pri_dut.gattc.write_invalid_char_by_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_mod_read_char_by_instance_id(self, line): |
| """GATT Client Read Char that doesn't have write permission""" |
| cmd = "GATT Client Read Char that doesn't have write permission" |
| try: |
| self.pri_dut.gattc.mod_read_char_by_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_read_invalid_char_by_instance_id(self, line): |
| """GATT Client Read Char that doesn't exists""" |
| cmd = "GATT Client Read Char that doesn't exists" |
| try: |
| self.pri_dut.gattc.read_invalid_char_by_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_mod_write_desc_by_instance_id(self, line): |
| """GATT Client Write to Desc that doesn't have write permission""" |
| cmd = "GATT Client Write to Desc that doesn't have write permission" |
| try: |
| self.pri_dut.gattc.mod_write_desc_by_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_write_invalid_desc_by_instance_id(self, line): |
| """GATT Client Write to Desc that doesn't exists""" |
| cmd = "GATT Client Write to Desc that doesn't exists" |
| try: |
| self.pri_dut.gattc.write_invalid_desc_by_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_mod_read_desc_by_instance_id(self, line): |
| """GATT Client Read Desc that doesn't have write permission""" |
| cmd = "GATT Client Read Desc that doesn't have write permission" |
| try: |
| self.pri_dut.gattc.mod_read_desc_by_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_read_invalid_desc_by_instance_id(self, line): |
| """GATT Client Read Desc that doesn't exists""" |
| cmd = "GATT Client Read Desc that doesn't exists" |
| try: |
| self.pri_dut.gattc.read_invalid_desc_by_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_mod_read_char_by_uuid_and_instance_id(self, line): |
| """GATT Client Read Char that doesn't have write permission""" |
| cmd = "GATT Client Read Char that doesn't have write permission" |
| try: |
| self.pri_dut.gattc.mod_read_char_by_uuid_and_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_read_invalid_char_by_uuid(self, line): |
| """GATT Client Read Char that doesn't exist""" |
| cmd = "GATT Client Read Char that doesn't exist" |
| try: |
| self.pri_dut.gattc.read_invalid_char_by_uuid(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_write_desc_by_instance_id(self, line): |
| """GATT Client Write to Descriptor by instance ID""" |
| cmd = "GATT Client Write to Descriptor by instance ID" |
| try: |
| self.pri_dut.gattc.write_desc_by_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_enable_notification_desc_by_instance_id(self, line): |
| """GATT Client Enable Notification on Descriptor by instance ID""" |
| cmd = "GATT Client Enable Notification on Descriptor by instance ID" |
| try: |
| self.pri_dut.gattc.enable_notification_desc_by_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_char_enable_all_notifications(self, line): |
| """GATT Client enable all notifications""" |
| cmd = "GATT Client enable all notifications" |
| try: |
| self.pri_dut.gattc.char_enable_all_notifications() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_read_char_by_invalid_instance_id(self, line): |
| """GATT Client read char by non-existant instance id""" |
| cmd = "GATT Client read char by non-existant instance id" |
| try: |
| self.pri_dut.gattc.read_char_by_invalid_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_begin_reliable_write(self, line): |
| """Begin a reliable write on the Bluetooth Gatt Client""" |
| cmd = "GATT Client Begin Reliable Write" |
| try: |
| self.pri_dut.gattc.begin_reliable_write() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_abort_reliable_write(self, line): |
| """Abort a reliable write on the Bluetooth Gatt Client""" |
| cmd = "GATT Client Abort Reliable Write" |
| try: |
| self.pri_dut.gattc.abort_reliable_write() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_execute_reliable_write(self, line): |
| """Execute a reliable write on the Bluetooth Gatt Client""" |
| cmd = "GATT Client Execute Reliable Write" |
| try: |
| self.pri_dut.gattc.execute_reliable_write() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_read_all_char(self, line): |
| """GATT Client read all Characteristic values""" |
| cmd = "GATT Client read all Characteristic values" |
| try: |
| self.pri_dut.gattc.read_all_char() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_write_all_char(self, line): |
| """Write to every Characteristic on the GATT server""" |
| cmd = "GATT Client Write All Characteristics" |
| try: |
| self.pri_dut.gattc.write_all_char(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_write_all_desc(self, line): |
| """ Write to every Descriptor on the GATT server """ |
| cmd = "GATT Client Write All Descriptors" |
| try: |
| self.pri_dut.gattc.write_all_desc(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_write_desc_notification_by_instance_id(self, line): |
| """Write 0x00 or 0x02 to notification descriptor [instance_id]""" |
| cmd = "Write 0x00 0x02 to notification descriptor" |
| try: |
| self.pri_dut.gattc.write_desc_notification_by_instance_id(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_discover_service_by_uuid(self, line): |
| """Discover service by uuid""" |
| cmd = "Discover service by uuid" |
| try: |
| self.pri_dut.gattc.discover_service_by_uuid(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gattc_read_all_desc(self, line): |
| """Read all Descriptor values""" |
| cmd = "Read all Descriptor values" |
| try: |
| self.pri_dut.gattc.read_all_desc() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| """End GATT Client wrappers""" |
| """Begin GATT Server wrappers""" |
| |
| def do_gatts_close_bluetooth_gatt_servers(self, line): |
| """Close Bluetooth Gatt Servers""" |
| cmd = "Close Bluetooth Gatt Servers" |
| try: |
| self.pri_dut.gatts.close_bluetooth_gatt_servers() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def complete_gatts_setup_database(self, text, line, begidx, endidx): |
| if not text: |
| completions = list( |
| gatt_test_database.GATT_SERVER_DB_MAPPING.keys())[:] |
| else: |
| completions = [ |
| s for s in gatt_test_database.GATT_SERVER_DB_MAPPING.keys() |
| if s.startswith(text) |
| ] |
| return completions |
| |
| def complete_gatts_send_response(self, text, line, begidx, endidx): |
| """GATT Server database name completion""" |
| if not text: |
| completions = list(gatt_server_responses.keys())[:] |
| else: |
| completions = [ |
| s for s in gatt_server_responses.keys() if s.startswith(text) |
| ] |
| return completions |
| |
| def complete_gatts_send_continuous_response(self, text, line, begidx, |
| endidx): |
| """GATT Server database name completion""" |
| if not text: |
| completions = list(gatt_server_responses.keys())[:] |
| else: |
| completions = [ |
| s for s in gatt_server_responses.keys() if s.startswith(text) |
| ] |
| return completions |
| |
| def complete_gatts_send_continuous_response_data(self, text, line, begidx, |
| endidx): |
| """GATT Server database name completion""" |
| if not text: |
| completions = list(gatt_server_responses.keys())[:] |
| else: |
| completions = [ |
| s for s in gatt_server_responses.keys() if s.startswith(text) |
| ] |
| return completions |
| |
| def do_gatts_list_all_uuids(self, line): |
| """From the GATT Client, discover services and list all services, |
| chars and descriptors |
| """ |
| cmd = "Discovery Services and list all UUIDS" |
| try: |
| self.pri_dut.gatts.list_all_uuids() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gatts_send_response(self, line): |
| """Send a response to the GATT Client""" |
| cmd = "GATT server send response" |
| try: |
| self.pri_dut.gatts.send_response(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gatts_notify_characteristic_changed(self, line): |
| """Notify char changed by instance id [instance_id] [true|false]""" |
| cmd = "Notify characteristic changed by instance id" |
| try: |
| info = line.split() |
| instance_id = info[0] |
| confirm_str = info[1] |
| confirm = False |
| if confirm_str.lower() == 'true': |
| confirm = True |
| self.pri_dut.gatts.notify_characteristic_changed( |
| instance_id, confirm) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gatts_setup_database(self, line): |
| cmd = "Setup GATT Server database: {}".format(line) |
| try: |
| self.pri_dut.gatts.setup_gatts_db( |
| gatt_test_database.GATT_SERVER_DB_MAPPING.get(line)) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gatts_characteristic_set_value_by_instance_id(self, line): |
| """Set Characteristic value by instance id""" |
| cmd = "Change value of a characteristic by instance id" |
| try: |
| info = line.split() |
| instance_id = info[0] |
| size = int(info[1]) |
| value = [] |
| for i in range(size): |
| value.append(i % 256) |
| self.pri_dut.gatts.gatt_server_characteristic_set_value_by_instance_id( |
| instance_id, value) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_notify_characteristic_changed(self, line): |
| """Notify characteristic changed [instance_id] [confirm]""" |
| cmd = "Notify characteristic changed" |
| try: |
| info = line.split() |
| instance_id = info[0] |
| confirm = bool(info[1]) |
| self.pri_dut.gatts.notify_characteristic_changed( |
| instance_id, confirm) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gatts_open(self, line): |
| """Open a GATT Server instance""" |
| cmd = "Open an empty GATT Server" |
| try: |
| self.pri_dut.gatts.open() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gatts_clear_services(self, line): |
| """Clear BluetoothGattServices from BluetoothGattServer""" |
| cmd = "Clear BluetoothGattServices from BluetoothGattServer" |
| try: |
| self.pri_dut.gatts.gatt_server_clear_services() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gatts_send_continuous_response(self, line): |
| """Send continous response with random data""" |
| cmd = "Send continous response with random data" |
| try: |
| self.pri_dut.gatts.send_continuous_response(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_gatts_send_continuous_response_data(self, line): |
| """Send continous response including requested data""" |
| cmd = "Send continous response including requested data" |
| try: |
| self.pri_dut.gatts.send_continuous_response_data(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| """End GATT Server wrappers""" |
| """Begin Ble wrappers""" |
| |
| def complete_ble_adv_data_include_local_name(self, text, line, begidx, |
| endidx): |
| options = ['true', 'false'] |
| if not text: |
| completions = list(options)[:] |
| else: |
| completions = [s for s in options if s.startswith(text)] |
| return completions |
| |
| def complete_ble_adv_data_include_tx_power_level(self, text, line, begidx, |
| endidx): |
| options = ['true', 'false'] |
| if not text: |
| completions = list(options)[:] |
| else: |
| completions = [s for s in options if s.startswith(text)] |
| return completions |
| |
| def complete_ble_stop_advertisement(self, text, line, begidx, endidx): |
| str_adv_list = list(map(str, self.pri_dut.ble.ADVERTISEMENT_LIST)) |
| if not text: |
| completions = str_adv_list[:] |
| else: |
| completions = [s for s in str_adv_list if s.startswith(text)] |
| return completions |
| |
| def do_ble_start_generic_connectable_advertisement(self, line): |
| """Start a connectable LE advertisement""" |
| cmd = "Start a connectable LE advertisement" |
| try: |
| self.pri_dut.ble.start_generic_connectable_advertisement(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def _start_max_advertisements(self, ad): |
| self.pri_dut.ble.start_max_advertisements(ad) |
| |
| def do_ble_start_generic_connectable_beacon_swarm(self, line): |
| """Start a connectable LE advertisement""" |
| cmd = "Start as many advertisements as possible on all devices" |
| try: |
| threads = [] |
| for ad in self.android_devices: |
| thread = threading.Thread( |
| target=self._start_max_advertisements, args=([ad])) |
| threads.append(thread) |
| thread.start() |
| for t in threads: |
| t.join() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_ble_start_connectable_advertisement_set(self, line): |
| """Start a connectable advertisement set""" |
| try: |
| self.pri_dut.ble.start_connectable_advertisement_set(line) |
| except Exception as err: |
| self.log.error("Failed to start advertisement: {}".format(err)) |
| |
| def do_ble_stop_all_advertisement_set(self, line): |
| """Stop all advertisement sets""" |
| try: |
| self.pri_dut.ble.stop_all_advertisement_set(line) |
| except Exception as err: |
| self.log.error("Failed to stop advertisement: {}".format(err)) |
| |
| def do_ble_adv_add_service_uuid_list(self, line): |
| """Add service UUID to the LE advertisement inputs: |
| [uuid1 uuid2 ... uuidN]""" |
| cmd = "Add a valid service UUID to the advertisement." |
| try: |
| self.pri_dut.ble.adv_add_service_uuid_list(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_ble_adv_data_include_local_name(self, line): |
| """Include local name in the advertisement. inputs: [true|false]""" |
| cmd = "Include local name in the advertisement." |
| try: |
| self.pri_dut.ble.adv_data_include_local_name(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_ble_adv_data_include_tx_power_level(self, line): |
| """Include tx power level in the advertisement. inputs: [true|false]""" |
| cmd = "Include local name in the advertisement." |
| try: |
| self.pri_dut.ble.adv_data_include_tx_power_level(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_ble_adv_data_add_manufacturer_data(self, line): |
| """Include manufacturer id and data to the advertisment: |
| [id data1 data2 ... dataN]""" |
| cmd = "Include manufacturer id and data to the advertisment." |
| try: |
| self.pri_dut.ble.adv_data_add_manufacturer_data(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_ble_start_generic_nonconnectable_advertisement(self, line): |
| """Start a nonconnectable LE advertisement""" |
| cmd = "Start a nonconnectable LE advertisement" |
| try: |
| self.pri_dut.ble.start_generic_nonconnectable_advertisement(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_ble_list_active_advertisement_ids(self, line): |
| """List all active BLE advertisements""" |
| self.log.info("IDs: {}".format(self.pri_dut.ble.advertisement_list)) |
| |
| def do_ble_stop_all_advertisements(self, line): |
| """Stop all LE advertisements""" |
| cmd = "Stop all LE advertisements" |
| try: |
| self.pri_dut.ble.stop_all_advertisements(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_ble_stop_advertisement(self, line): |
| """Stop an LE advertisement""" |
| cmd = "Stop a connectable LE advertisement" |
| try: |
| self.pri_dut.ble.ble_stop_advertisement(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| """End Ble wrappers""" |
| """Begin Bta wrappers""" |
| |
| def complete_bta_start_pairing_helper(self, text, line, begidx, endidx): |
| options = ['true', 'false'] |
| if not text: |
| completions = list(options)[:] |
| else: |
| completions = [s for s in options if s.startswith(text)] |
| return completions |
| |
| def complete_bta_set_scan_mode(self, text, line, begidx, endidx): |
| completions = list(bt_scan_mode_types.keys())[:] |
| if not text: |
| completions = completions[:] |
| else: |
| completions = [s for s in completions if s.startswith(text)] |
| return completions |
| |
| def do_bta_set_scan_mode(self, line): |
| """Set the Scan mode of the Bluetooth Adapter""" |
| cmd = "Set the Scan mode of the Bluetooth Adapter" |
| try: |
| self.pri_dut.bta.set_scan_mode(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_set_device_name(self, line): |
| """Set Bluetooth Adapter Name""" |
| cmd = "Set Bluetooth Adapter Name" |
| try: |
| self.pri_dut.bta.set_device_name(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_enable(self, line): |
| """Enable Bluetooth Adapter""" |
| cmd = "Enable Bluetooth Adapter" |
| try: |
| self.pri_dut.bta.enable() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_disable(self, line): |
| """Disable Bluetooth Adapter""" |
| cmd = "Disable Bluetooth Adapter" |
| try: |
| self.pri_dut.bta.disable() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_init_bond(self, line): |
| """Initiate bond to PTS device""" |
| cmd = "Initiate Bond" |
| try: |
| self.pri_dut.bta.init_bond() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_start_discovery(self, line): |
| """Start BR/EDR Discovery""" |
| cmd = "Start BR/EDR Discovery" |
| try: |
| self.pri_dut.bta.start_discovery() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_stop_discovery(self, line): |
| """Stop BR/EDR Discovery""" |
| cmd = "Stop BR/EDR Discovery" |
| try: |
| self.pri_dut.bta.stop_discovery() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_get_discovered_devices(self, line): |
| """Get Discovered Br/EDR Devices""" |
| cmd = "Get Discovered Br/EDR Devices\n" |
| try: |
| self.pri_dut.bta.get_discovered_devices() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_bond(self, line): |
| """Bond to PTS device""" |
| cmd = "Bond to the PTS dongle directly" |
| try: |
| self.pri_dut.bta.bond() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_disconnect(self, line): |
| """BTA disconnect""" |
| cmd = "BTA disconnect" |
| try: |
| self.pri_dut.bta.disconnect() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_unbond(self, line): |
| """Unbond from PTS device""" |
| cmd = "Unbond from the PTS dongle" |
| try: |
| self.pri_dut.bta.unbond() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_start_pairing_helper(self, line): |
| """Start or stop Bluetooth Pairing Helper""" |
| cmd = "Start or stop BT Pairing helper" |
| try: |
| self.pri_dut.bta.start_pairing_helper(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_push_pairing_pin(self, line): |
| """Push pairing pin to the Android Device""" |
| cmd = "Push the pin to the Android Device" |
| try: |
| self.pri_dut.bta.push_pairing_pin(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_get_pairing_pin(self, line): |
| """Get pairing PIN""" |
| cmd = "Get Pin Info" |
| try: |
| self.pri_dut.bta.get_pairing_pin() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_fetch_uuids_with_sdp(self, line): |
| """BTA fetch UUIDS with SDP""" |
| cmd = "Fetch UUIDS with SDP" |
| try: |
| self.pri_dut.bta.fetch_uuids_with_sdp() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_connect_profiles(self, line): |
| """Connect available profiles""" |
| cmd = "Connect all profiles possible" |
| try: |
| self.pri_dut.bta.connect_profiles() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_tts_speak(self, line): |
| cmd = "Open audio channel by speaking characters" |
| try: |
| self.pri_dut.bta.tts_speak() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| """End Bta wrappers""" |
| """Begin Rfcomm wrappers""" |
| |
| def do_rfcomm_connect(self, line): |
| """Perform an RFCOMM connect""" |
| try: |
| self.pri_dut.rfcomm.connect(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_rfcomm_open_rfcomm_socket(self, line): |
| """Open rfcomm socket""" |
| cmd = "Open RFCOMM socket" |
| try: |
| self.pri_dut.rfcomm.open_rfcomm_socket() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_rfcomm_open_l2cap_socket(self, line): |
| """Open L2CAP socket""" |
| cmd = "Open L2CAP socket" |
| try: |
| self.pri_dut.rfcomm.open_l2cap_socket() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_rfcomm_write(self, line): |
| cmd = "Write String data over an RFCOMM connection" |
| try: |
| self.pri_dut.rfcomm.write(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_rfcomm_write_binary(self, line): |
| cmd = "Write String data over an RFCOMM connection" |
| try: |
| self.pri_dut.rfcomm.write_binary(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_rfcomm_end_connect(self, line): |
| cmd = "End RFCOMM connection" |
| try: |
| self.pri_dut.rfcomm.end_connect() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_rfcomm_accept(self, line): |
| cmd = "Accept RFCOMM connection" |
| try: |
| self.pri_dut.rfcomm.accept(line) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_rfcomm_stop(self, line): |
| cmd = "STOP RFCOMM Connection" |
| try: |
| self.pri_dut.rfcomm.stop() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_rfcomm_open_l2cap_socket(self, line): |
| """Open L2CAP socket""" |
| cmd = "Open L2CAP socket" |
| try: |
| self.pri_dut.rfcomm.open_l2cap_socket() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| """End Rfcomm wrappers""" |
| """Begin Config wrappers""" |
| |
| def do_config_reset(self, line): |
| """Reset Bluetooth Config file""" |
| cmd = "Reset Bluetooth Config file" |
| try: |
| self.pri_dut.config.reset() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_config_set_nonbond(self, line): |
| """Set NonBondable Mode""" |
| cmd = "Set NonBondable Mode" |
| try: |
| self.pri_dut.config.set_nonbond() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_config_set_disable_mitm(self, line): |
| """Set Disable MITM""" |
| cmd = "Set Disable MITM" |
| try: |
| self.pri_dut.config.set_disable_mitm() |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| """End Config wrappers""" |
| """Begin HFP/HSP wrapper""" |
| |
| def do_bta_hfp_start_voice_recognition(self, line): |
| self.pri_dut.droid.bluetoothHspStartVoiceRecognition(self.mac_addr) |
| |
| def do_bta_hfp_stop_voice_recognition(self, line): |
| self.pri_dut.droid.bluetoothHspStopVoiceRecognition(self.mac_addr) |
| |
| def do_test_clcc_response(self, line): |
| # Experimental |
| clcc_index = 1 |
| clcc_direction = 1 |
| clcc_status = 4 |
| clcc_mode = 0 |
| clcc_mpty = False |
| clcc_number = "18888888888" |
| clcc_type = 0 |
| self.pri_dut.droid.bluetoothHspClccResponse( |
| clcc_index, clcc_direction, clcc_status, clcc_mode, clcc_mpty, |
| clcc_number, clcc_type) |
| |
| def do_bta_hsp_force_sco_audio_on(self, line): |
| """HFP/HSP Force SCO Audio ON""" |
| cmd = "HFP/HSP Force SCO Audio ON" |
| try: |
| if not self.pri_dut.droid.bluetoothHspForceScoAudio(True): |
| self.log.info( |
| FAILURE.format(cmd, |
| "bluetoothHspForceScoAudio returned false")) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_hsp_force_sco_audio_off(self, line): |
| """HFP/HSP Force SCO Audio OFF""" |
| cmd = "HFP/HSP Force SCO Audio OFF" |
| try: |
| if not self.pri_dut.droid.bluetoothHspForceScoAudio(False): |
| self.log.info( |
| FAILURE.format(cmd, |
| "bluetoothHspForceScoAudio returned false")) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_hsp_connect_audio(self, line): |
| """HFP/HSP connect audio""" |
| cmd = "HFP/HSP connect audio" |
| try: |
| if not self.pri_dut.droid.bluetoothHspConnectAudio(self.mac_addr): |
| self.log.info( |
| FAILURE.format( |
| cmd, "bluetoothHspConnectAudio returned false for " + |
| self.mac_addr)) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_hsp_disconnect_audio(self, line): |
| """HFP/HSP disconnect audio""" |
| cmd = "HFP/HSP disconnect audio" |
| try: |
| if not self.pri_dut.droid.bluetoothHspDisconnectAudio( |
| self.mac_addr): |
| self.log.info( |
| FAILURE.format( |
| cmd, |
| "bluetoothHspDisconnectAudio returned false for " + |
| self.mac_addr)) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_hsp_connect_slc(self, line): |
| """HFP/HSP connect SLC with additional tries and help""" |
| cmd = "Connect to hsp with some help" |
| try: |
| if not self.connect_hsp_helper(self.pri_dut): |
| self.log.error("Failed to connect to HSP") |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_bta_hsp_disconnect_slc(self, line): |
| """HFP/HSP disconnect SLC""" |
| cmd = "HFP/HSP disconnect SLC" |
| try: |
| if not self.pri_dut.droid.bluetoothHspDisconnect(self.mac_addr): |
| self.log.info( |
| FAILURE.format( |
| cmd, "bluetoothHspDisconnect returned false for " + |
| self.mac_addr)) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| """End HFP/HSP wrapper""" |
| """Begin HID wrappers""" |
| |
| def do_hid_get_report(self, line): |
| """Get HID Report""" |
| cmd = "Get HID Report" |
| try: |
| self.pri_dut.droid.bluetoothHidGetReport(self.mac_addr, 1, 1, 1024) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_hid_set_report(self, line): |
| """Get HID Report""" |
| cmd = "Get HID Report" |
| try: |
| self.pri_dut.droid.bluetoothHidSetReport(self.mac_addr, 1, "Test") |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_hid_virtual_unplug(self, line): |
| """Get HID Report""" |
| cmd = "Get HID Report" |
| try: |
| self.pri_dut.droid.bluetoothHidVirtualUnplug(self.mac_addr) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_hid_send_report(self, line): |
| """Get HID Report""" |
| cmd = "Get HID Report" |
| try: |
| self.pri_dut.droid.bluetoothHidSendData(device_id, "42") |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| """End HID wrappers""" |
| """Begin carkit test wrappers""" |
| |
| def do_test_suite_generic_bt_tests(self, line): |
| """Run generic Bluetooth connection test suite""" |
| generic_bt_tests = [ |
| tuple((self.bt_carkit_lib.disconnect_reconnect_multiple_iterations, |
| [self.pri_dut])), |
| tuple((self.bt_carkit_lib.disconnect_a2dp_only_then_reconnect, |
| [self.pri_dut])), |
| tuple((self.bt_carkit_lib.disconnect_hsp_only_then_reconnect, |
| [self.pri_dut])), |
| tuple(( |
| self.bt_carkit_lib.disconnect_both_hsp_and_a2dp_then_reconnect, |
| [self.pri_dut])), |
| ] |
| try: |
| for func, param in generic_bt_tests: |
| try: |
| func(param) |
| except Exception: |
| self.log.info("Test {} failed.".format(func)) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_connect_hsp_helper(self, line): |
| """Connect to HSP/HFP with additional tries and help""" |
| cmd = "Connect to hsp with some help" |
| try: |
| self.bt_carkit_lib.connect_hsp_helper(self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_hfp_setup_multi_incomming_calls(self, line): |
| """Setup two incomming calls""" |
| cmd = "Setup multiple incomming calls" |
| try: |
| self.bt_carkit_lib.setup_multi_call(self.sec_dut, self.ter_dut, |
| self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_disconnect_reconnect_multiple_iterations(self, line): |
| """Quick disconnect/reconnect stress test""" |
| try: |
| self.bt_carkit_lib.disconnect_reconnect_multiple_iterations( |
| self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_disconnect_a2dp_only_then_reconnect(self, line): |
| """Test disconnect-reconnect a2dp only scenario from phone.""" |
| cmd = "Test disconnect-reconnect a2dp only scenario from phone." |
| try: |
| self.bt_carkit_lib.disconnect_a2dp_only_then_reconnect( |
| self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_disconnect_hsp_only_then_reconnect(self, line): |
| """Test disconnect-reconnect hsp only scenario from phone.""" |
| cmd = "Test disconnect-reconnect hsp only scenario from phone." |
| try: |
| self.bt_carkit_lib.disconnect_hsp_only_then_reconnect(self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_disconnect_both_hsp_and_a2dp_then_reconnect(self, line): |
| """Test disconnect-reconnect hsp and a2dp scenario from phone.""" |
| cmd = "Test disconnect-reconnect hsp and a2dp scenario from phone." |
| try: |
| self.bt_carkit_lib.disconnect_both_hsp_and_a2dp_then_reconnect( |
| self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_outgoing_call_private_number(self, line): |
| """Test outgoing call scenario from phone to private number""" |
| cmd = "Test outgoing call scenario from phone to private number" |
| try: |
| self.bt_carkit_lib.outgoing_call_private_number( |
| self.pri_dut, self.ter_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_outgoing_call_a2dp_play_before_and_after(self, line): |
| """Test outgoing call scenario while playing music. Music should resume |
| after call.""" |
| cmd = "Test outgoing call scenario while playing music. Music should " \ |
| "resume after call." |
| try: |
| self.bt_carkit_lib.outgoing_call_a2dp_play_before_and_after( |
| self.pri_dut, self.sec_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_outgoing_call_unknown_contact(self, line): |
| """Test outgoing call scenario from phone to unknow contact""" |
| cmd = "Test outgoing call scenario from phone to unknow contact" |
| try: |
| self.bt_carkit_lib.outgoing_call_unknown_contact( |
| self.pri_dut, self.ter_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_incomming_call_private_number(self, line): |
| """Test incomming call scenario to phone from unknown contact""" |
| cmd = "Test incomming call scenario to phone from unknown contact" |
| try: |
| self.bt_carkit_lib.incomming_call_private_number( |
| self.pri_dut, self.ter_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_outgoing_call_multiple_iterations(self, line): |
| """Test outgoing call quickly 3 times""" |
| cmd = "Test outgoing call quickly 3 times" |
| try: |
| self.bt_carkit_lib.outgoing_call_multiple_iterations( |
| self.pri_dut, self.sec_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_outgoing_call_hsp_disabled_then_enabled_during_call(self, line): |
| """Test outgoing call hsp disabled then enable during call.""" |
| cmd = "Test outgoing call hsp disabled then enable during call." |
| try: |
| self.bt_carkit_lib.outgoing_call_hsp_disabled_then_enabled_during_call( |
| self.pri_dut, self.sec_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_call_audio_routes(self, line): |
| """Test various audio routes scenario from phone.""" |
| cmd = "Test various audio routes scenario from phone." |
| try: |
| self.bt_carkit_lib.call_audio_routes(self.pri_dut, self.sec_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_sms_receive_different_sizes(self, line): |
| """Test recieve sms of different sizes.""" |
| cmd = "Test recieve sms of different sizes." |
| try: |
| self.bt_carkit_lib.sms_receive_different_sizes( |
| self.pri_dut, self.sec_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_sms_receive_multiple(self, line): |
| """Test recieve sms of different sizes.""" |
| cmd = "Test recieve sms of different sizes." |
| try: |
| self.bt_carkit_lib.sms_receive_multiple(self.pri_dut, self.sec_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_sms_send_outgoing_texts(self, line): |
| """Test send sms of different sizes.""" |
| cmd = "Test send sms of different sizes." |
| try: |
| self.bt_carkit_lib.sms_send_outgoing_texts(self.pri_dut, |
| self.sec_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_sms_during_incomming_call(self, line): |
| """Test incomming call scenario to phone from unknown contact""" |
| cmd = "Test incomming call scenario to phone from unknown contact" |
| try: |
| self.bt_carkit_lib.sms_during_incomming_call( |
| self.pri_dut, self.sec_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_multi_incomming_call(self, line): |
| """Test 2 incomming calls scenario to phone.""" |
| cmd = "Test 2 incomming calls scenario to phone." |
| try: |
| self.bt_carkit_lib.multi_incomming_call(self.pri_dut, self.sec_dut, |
| self.ter_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_multi_call_audio_routing(self, line): |
| """Test 2 incomming calls scenario to phone, then test audio routing.""" |
| cmd = "Test 2 incomming calls scenario to phone, then test audio" \ |
| "routing." |
| try: |
| self.bt_carkit_lib.multi_call_audio_routing( |
| self.pri_dut, self.sec_dut, self.ter_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_multi_call_swap_multiple_times(self, line): |
| """Test 2 incomming calls scenario to phone, then swap the calls |
| multiple times""" |
| cmd = "Test 2 incomming calls scenario to phone, then swap the calls" \ |
| "multiple times" |
| try: |
| self.bt_carkit_lib.multi_call_swap_multiple_times( |
| self.pri_dut, self.sec_dut, self.ter_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_multi_call_join_conference_call(self, line): |
| """Test 2 incomming calls scenario to phone then join the calls.""" |
| cmd = "Test 2 incomming calls scenario to phone then join the calls." |
| try: |
| self.bt_carkit_lib.multi_call_join_conference_call( |
| self.pri_dut, self.sec_dut, self.ter_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_multi_call_join_conference_call_hangup_conf_call(self, line): |
| """Test 2 incomming calls scenario to phone then join the calls, |
| then terminate the call from the primary dut.""" |
| cmd = "Test 2 incomming calls scenario to phone then join the calls, " \ |
| "then terminate the call from the primary dut." |
| try: |
| self.bt_carkit_lib.multi_call_join_conference_call_hangup_conf_call( |
| self.pri_dut, self.sec_dut, self.ter_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_outgoing_multi_call_join_conference_call(self, line): |
| """Test 2 outgoing calls scenario from phone then join the calls.""" |
| cmd = "Test 2 outgoing calls scenario from phone then join the calls." |
| try: |
| self.bt_carkit_lib.outgoing_multi_call_join_conference_call( |
| self.pri_dut, self.sec_dut, self.ter_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_multi_call_join_conference_call_audio_routes(self, line): |
| """Test 2 incomming calls scenario to phone then join the calls, |
| then test different audio routes.""" |
| cmd = "Test 2 incomming calls scenario to phone then join the calls, " \ |
| "then test different audio routes." |
| try: |
| self.bt_carkit_lib.multi_call_join_conference_call_audio_routes( |
| self.pri_dut, self.sec_dut, self.ter_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_avrcp_play_pause(self, line): |
| """Test avrcp play/pause commands multiple times from phone""" |
| cmd = "Test avrcp play/pause commands multiple times from phone" |
| try: |
| self.bt_carkit_lib.avrcp_play_pause(self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_avrcp_next_previous_song(self, line): |
| """Test AVRCP go to the next song then the previous song.""" |
| cmd = "Test AVRCP go to the next song then the previous song." |
| try: |
| self.bt_carkit_lib.avrcp_next_previous_song(self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_avrcp_next_previous(self, line): |
| """Test AVRCP go to the next song then the press previous after a few |
| seconds.""" |
| cmd = "Test AVRCP go to the next song then the press previous after " \ |
| "a few seconds." |
| try: |
| self.bt_carkit_lib.avrcp_next_previous(self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_avrcp_next_repetative(self, line): |
| """Test AVRCP go to the next 10 times""" |
| cmd = "Test AVRCP go to the next 10 times" |
| try: |
| self.bt_carkit_lib.avrcp_next_repetative(self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def _process_question(self, question, expected_response): |
| while True: |
| try: |
| result = input(question.format(level)).lower() |
| except Exception as err: |
| print(err) |
| |
| def do_e2e_cycle_battery_level(self, line): |
| """Cycle battery level through different values and verify result on carkit""" |
| cmd = "Test that verifies battery level indicator changes with the " \ |
| "phone. Phone current level." |
| try: |
| self.bt_carkit_lib.cycle_absolute_volume_control(self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_cycle_absolute_volume_control(self, line): |
| """Cycle media volume level through different values and verify result on carkit""" |
| cmd = "Test aboslute volume on carkit by changed volume levels from phone." |
| try: |
| self.bt_carkit_lib.cycle_absolute_volume_control(self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_test_voice_recognition_from_phone(self, line): |
| """Test Voice Recognition from phone.""" |
| cmd = "Test voice recognition from phone." |
| try: |
| self.bt_carkit_lib.test_voice_recognition_from_phone(self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| def do_e2e_test_audio_and_voice_recognition_from_phone(self, line): |
| """Test Voice Recognition from phone and confirm music audio continues.""" |
| cmd = "Test Voice Recognition from phone and confirm music audio continues." |
| try: |
| self.bt_carkit_lib.test_audio_and_voice_recognition_from_phone( |
| self.pri_dut) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| """End carkit test wrappers""" |
| """Begin adb shell test wrappers""" |
| |
| def do_set_battery_level(self, line): |
| """Set battery level based on input""" |
| cmd = "Set battery level based on input" |
| try: |
| self.pri_dut.shell.set_battery_level(int(line)) |
| except Exception as err: |
| self.log.info(FAILURE.format(cmd, err)) |
| |
| """End adb shell test wrappers""" |