1#!/usr/bin/env python3 2# 3# Copyright (C) 2018 The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); you may not 6# use this file except in compliance with the License. You may obtain a copy of 7# the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 14# License for the specific language governing permissions and limitations under 15# the License. 16 17import json 18import os 19import threading 20import time 21 22from acts.base_test import BaseTestClass 23from acts.controllers import android_device 24from acts.controllers import relay_device_controller 25from acts_contrib.test_utils.bt.bt_test_utils import disable_bluetooth 26from acts_contrib.test_utils.bt.bt_test_utils import enable_bluetooth 27from acts_contrib.test_utils.bt.bt_test_utils import setup_multiple_devices_for_bt_test 28from acts_contrib.test_utils.bt.bt_test_utils import take_btsnoop_logs 29from acts_contrib.test_utils.coex.coex_test_utils import A2dpDumpsysParser 30from acts_contrib.test_utils.coex.coex_test_utils import ( 31 collect_bluetooth_manager_dumpsys_logs) 32from acts_contrib.test_utils.coex.coex_test_utils import configure_and_start_ap 33from acts_contrib.test_utils.coex.coex_test_utils import iperf_result 34from acts_contrib.test_utils.coex.coex_test_utils import parse_fping_results 35from acts_contrib.test_utils.coex.coex_test_utils import wifi_connection_check 36from acts_contrib.test_utils.wifi import wifi_retail_ap as retail_ap 37from acts_contrib.test_utils.wifi.wifi_performance_test_utils import get_iperf_arg_string 38from acts_contrib.test_utils.wifi.wifi_power_test_utils import get_phone_ip 39from acts_contrib.test_utils.wifi.wifi_test_utils import reset_wifi 40from acts_contrib.test_utils.wifi.wifi_test_utils import wifi_connect 41from acts_contrib.test_utils.wifi.wifi_test_utils import wifi_test_device_init 42from acts_contrib.test_utils.wifi.wifi_test_utils import wifi_toggle_state 43 44AVRCP_WAIT_TIME = 3 45 46 47class CoexBaseTest(BaseTestClass): 48 49 class IperfVariables: 50 51 def __init__(self, current_test_name): 52 self.iperf_started = False 53 self.bidirectional_client_path = None 54 self.bidirectional_server_path = None 55 self.iperf_server_path = None 56 self.iperf_client_path = None 57 self.throughput = [] 58 self.protocol = current_test_name.split("_")[-2] 59 self.stream = current_test_name.split("_")[-1] 60 self.is_bidirectional = False 61 if self.stream == 'bidirectional': 62 self.is_bidirectional = True 63 64 def setup_class(self): 65 super().setup_class() 66 self.pri_ad = self.android_devices[0] 67 if len(self.android_devices) == 2: 68 self.sec_ad = self.android_devices[1] 69 elif len(self.android_devices) == 3: 70 self.third_ad = self.android_devices[2] 71 self.ssh_config = None 72 73 self.counter = 0 74 self.thread_list = [] 75 if not setup_multiple_devices_for_bt_test(self.android_devices): 76 self.log.error('Failed to setup devices for bluetooth test') 77 return False 78 req_params = ['network', 'iperf'] 79 opt_params = [ 80 'AccessPoint', 'RetailAccessPoints', 'RelayDevice', 81 'required_devices' 82 ] 83 self.unpack_userparams(req_params, opt_params) 84 85 self.iperf_server = self.iperf_servers[0] 86 self.iperf_client = self.iperf_clients[0] 87 88 if hasattr(self, 'RelayDevice'): 89 self.audio_receiver = self.relay_devices[0] 90 self.audio_receiver.power_on() 91 self.headset_mac_address = self.audio_receiver.mac_address 92 else: 93 self.log.warning('Missing Relay config file.') 94 95 if hasattr(self, 'AccessPoint'): 96 self.ap = self.access_points[0] 97 configure_and_start_ap(self.ap, self.network) 98 elif hasattr(self, 'RetailAccessPoints'): 99 self.retail_access_points = retail_ap.create( 100 self.RetailAccessPoints) 101 self.retail_access_point = self.retail_access_points[0] 102 band = self.retail_access_point.band_lookup_by_channel( 103 self.network['channel']) 104 self.retail_access_point.set_channel(band, self.network['channel']) 105 else: 106 self.log.warning('config file have no access point information') 107 108 wifi_test_device_init(self.pri_ad) 109 wifi_connect(self.pri_ad, self.network, num_of_tries=5) 110 111 def setup_test(self): 112 self.tag = 0 113 self.result = {} 114 self.dev_list = {} 115 self.iperf_variables = self.IperfVariables(self.current_test_name) 116 self.a2dp_dumpsys = A2dpDumpsysParser() 117 self.log_path = os.path.join(self.pri_ad.log_path, 118 self.current_test_name) 119 os.makedirs(self.log_path, exist_ok=True) 120 self.json_file = os.path.join(self.log_path, 'test_results.json') 121 for a in self.android_devices: 122 a.ed.clear_all_events() 123 if not wifi_connection_check(self.pri_ad, self.network['SSID']): 124 self.log.error('Wifi connection does not exist') 125 return False 126 if not enable_bluetooth(self.pri_ad.droid, self.pri_ad.ed): 127 self.log.error('Failed to enable bluetooth') 128 return False 129 if hasattr(self, 'required_devices'): 130 if ('discovery' in self.current_test_name or 131 'ble' in self.current_test_name): 132 self.create_android_relay_object() 133 else: 134 self.log.warning('required_devices is not given in config file') 135 136 def teardown_test(self): 137 self.parsing_results() 138 with open(self.json_file, 'a') as results_file: 139 json.dump(self.result, results_file, indent=4, sort_keys=True) 140 if not disable_bluetooth(self.pri_ad.droid): 141 self.log.info('Failed to disable bluetooth') 142 return False 143 self.destroy_android_and_relay_object() 144 145 def teardown_class(self): 146 if hasattr(self, 'AccessPoint'): 147 self.ap.close() 148 self.reset_wifi_and_store_results() 149 150 def reset_wifi_and_store_results(self): 151 """Resets wifi and store test results.""" 152 reset_wifi(self.pri_ad) 153 wifi_toggle_state(self.pri_ad, False) 154 155 def create_android_relay_object(self): 156 """Creates android device object and relay device object if required 157 devices has android device and relay device.""" 158 if 'AndroidDevice' in self.required_devices: 159 self.inquiry_devices = android_device.create( 160 self.required_devices['AndroidDevice']) 161 self.dev_list['AndroidDevice'] = self.inquiry_devices 162 if 'RelayDevice' in self.required_devices: 163 self.relay = relay_device_controller.create( 164 self.required_devices['RelayDevice']) 165 self.dev_list['RelayDevice'] = self.relay 166 167 def destroy_android_and_relay_object(self): 168 """Destroys android device object and relay device object if required 169 devices has android device and relay device.""" 170 if hasattr(self, 'required_devices'): 171 if ('discovery' in self.current_test_name or 172 'ble' in self.current_test_name): 173 if hasattr(self, 'inquiry_devices'): 174 for device in range(len(self.inquiry_devices)): 175 inquiry_device = self.inquiry_devices[device] 176 if not disable_bluetooth(inquiry_device.droid): 177 self.log.info('Failed to disable bluetooth') 178 android_device.destroy(self.inquiry_devices) 179 if hasattr(self, 'relay'): 180 relay_device_controller.destroy(self.relay) 181 182 def parsing_results(self): 183 """Result parser for fping results and a2dp packet drops.""" 184 if 'fping' in self.current_test_name: 185 output_path = '{}{}{}'.format(self.pri_ad.log_path, '/Fping/', 186 'fping_%s.txt' % self.counter) 187 self.result['fping_loss%'] = parse_fping_results( 188 self.fping_params['fping_drop_tolerance'], output_path) 189 self.counter = +1 190 if 'a2dp_streaming' in self.current_test_name: 191 file_path = collect_bluetooth_manager_dumpsys_logs( 192 self.pri_ad, self.current_test_name) 193 self.result['a2dp_packet_drop'] = ( 194 self.a2dp_dumpsys.parse(file_path)) 195 if self.result['a2dp_packet_drop'] == 0: 196 self.result['a2dp_packet_drop'] = None 197 198 def run_iperf_and_get_result(self): 199 """Frames iperf command based on test and starts iperf client on 200 host machine. 201 202 Returns: 203 throughput: Throughput of the run. 204 """ 205 self.iperf_server.start(tag=self.tag) 206 if self.iperf_variables.stream == 'ul': 207 iperf_args = get_iperf_arg_string( 208 duration=self.iperf['duration'], 209 reverse_direction=1, 210 traffic_type=self.iperf_variables.protocol 211 ) 212 elif self.iperf_variables.stream == 'dl': 213 iperf_args = get_iperf_arg_string( 214 duration=self.iperf['duration'], 215 reverse_direction=0, 216 traffic_type=self.iperf_variables.protocol 217 ) 218 ip = get_phone_ip(self.pri_ad) 219 self.tag = self.tag + 1 220 self.iperf_variables.iperf_client_path = ( 221 self.iperf_client.start(ip, iperf_args, self.tag)) 222 223 self.iperf_server.stop() 224 if (self.iperf_variables.protocol == 'udp' and 225 self.iperf_variables.stream == 'ul'): 226 throughput = iperf_result( 227 self.log, self.iperf_variables.protocol, 228 self.iperf_variables.iperf_server_path) 229 else: 230 throughput = iperf_result(self.log, 231 self.iperf_variables.protocol, 232 self.iperf_variables.iperf_client_path) 233 234 if not throughput: 235 self.log.error('Iperf failed/stopped') 236 self.iperf_variables.throughput.append(0) 237 else: 238 self.iperf_variables.throughput.append( 239 str(round(throughput, 2)) + "Mb/s") 240 self.log.info("Throughput: {} Mb/s".format(throughput)) 241 self.result["throughput"] = self.iperf_variables.throughput 242 return throughput 243 244 def on_fail(self, test_name, begin_time): 245 """A function that is executed upon a test case failure. 246 247 Args: 248 test_name: Name of the test that triggered this function. 249 begin_time: Logline format timestamp taken when the test started. 250 """ 251 self.log.info('Test {} failed, Fetching Btsnoop logs and bugreport'. 252 format(test_name)) 253 take_btsnoop_logs(self.android_devices, self, test_name) 254 self._take_bug_report(test_name, begin_time) 255 256 def run_thread(self, kwargs): 257 """Convenience function to start thread. 258 259 Args: 260 kwargs: Function object to start in thread. 261 """ 262 for function in kwargs: 263 self.thread = threading.Thread(target=function) 264 self.thread_list.append(self.thread) 265 self.thread.start() 266 267 def teardown_thread(self): 268 """Convenience function to join thread.""" 269 for thread_id in self.thread_list: 270 if thread_id.is_alive(): 271 thread_id.join() 272 273 def get_call_volume(self): 274 """Function to get call volume when bluetooth headset connected. 275 276 Returns: 277 Call volume. 278 """ 279 return self.pri_ad.adb.shell( 280 'settings list system|grep volume_bluetooth_sco_bt_sco_hs') 281 282 def change_volume(self): 283 """Changes volume with HFP call. 284 285 Returns: True if successful, otherwise False. 286 """ 287 if 'Volume_up' and 'Volume_down' in ( 288 self.relay_devices[0].relays.keys()): 289 current_volume = self.get_call_volume() 290 self.audio_receiver.press_volume_down() 291 time.sleep(AVRCP_WAIT_TIME) # wait till volume_changes 292 if current_volume == self.get_call_volume(): 293 self.log.error('Decrease volume failed') 294 return False 295 time.sleep(AVRCP_WAIT_TIME) 296 current_volume = self.get_call_volume() 297 self.audio_receiver.press_volume_up() 298 time.sleep(AVRCP_WAIT_TIME) # wait till volume_changes 299 if current_volume == self.get_call_volume(): 300 self.log.error('Increase volume failed') 301 return False 302 else: 303 self.log.warning( 304 'No volume control pins specified in relay config.') 305 return True 306