1 #/usr/bin/env python3.4 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 17 import os 18 import threading 19 import time 20 21 from acts.base_test import BaseTestClass 22 from acts.controllers.iperf_client import IPerfClient 23 from acts.test_utils.bt.bt_test_utils import disable_bluetooth 24 from acts.test_utils.bt.bt_test_utils import enable_bluetooth 25 from acts.test_utils.bt.bt_test_utils import setup_multiple_devices_for_bt_test 26 from acts.test_utils.bt.bt_test_utils import take_btsnoop_logs 27 from acts.test_utils.coex.coex_test_utils import configure_and_start_ap 28 from acts.test_utils.coex.coex_test_utils import iperf_result 29 from acts.test_utils.coex.coex_test_utils import get_phone_ip 30 from acts.test_utils.coex.coex_test_utils import wifi_connection_check 31 from acts.test_utils.coex.coex_test_utils import xlsheet 32 from acts.test_utils.wifi.wifi_test_utils import reset_wifi 33 from acts.test_utils.wifi.wifi_test_utils import wifi_connect 34 from acts.test_utils.wifi.wifi_test_utils import wifi_test_device_init 35 from acts.test_utils.wifi.wifi_test_utils import wifi_toggle_state 36 from acts.utils import create_dir 37 from acts.utils import start_standing_subprocess 38 from acts.utils import stop_standing_subprocess 39 40 TEST_CASE_TOKEN = "[Test Case]" 41 RESULT_LINE_TEMPLATE = TEST_CASE_TOKEN + " %s %s" 42 IPERF_SERVER_WAIT_TIME = 5 43 44 45 class CoexBaseTest(BaseTestClass): 46 47 def __init__(self, controllers): 48 BaseTestClass.__init__(self, controllers) 49 self.pri_ad = self.android_devices[0] 50 if len(self.android_devices) == 2: 51 self.sec_ad = self.android_devices[1] 52 elif len(self.android_devices) == 3: 53 self.third_ad = self.android_devices[2] 54 55 def setup_class(self): 56 self.tag = 0 57 self.iperf_result = {} 58 self.thread_list = [] 59 if not setup_multiple_devices_for_bt_test(self.android_devices): 60 self.log.error("Failed to setup devices for bluetooth test") 61 return False 62 req_params = ["network", "iperf"] 63 self.unpack_userparams(req_params) 64 if "RelayDevice" in self.user_params: 65 self.audio_receiver = self.relay_devices[0] 66 else: 67 self.log.warning("Missing Relay config file.") 68 if "music_file" in self.user_params: 69 self.push_music_to_android_device(self.pri_ad) 70 self.path = self.pri_ad.log_path 71 if "AccessPoints" in self.user_params: 72 self.ap = self.access_points[0] 73 configure_and_start_ap(self.ap, self.network) 74 wifi_test_device_init(self.pri_ad) 75 wifi_connect(self.pri_ad, self.network) 76 77 def setup_test(self): 78 self.received = [] 79 for a in self.android_devices: 80 a.ed.clear_all_events() 81 if not wifi_connection_check(self.pri_ad, self.network["SSID"]): 82 self.log.error("Wifi connection does not exist") 83 return False 84 if not enable_bluetooth(self.pri_ad.droid, self.pri_ad.ed): 85 self.log.error("Failed to enable bluetooth") 86 return False 87 88 def teardown_test(self): 89 if not disable_bluetooth(self.pri_ad.droid): 90 self.log.info("Failed to disable bluetooth") 91 return False 92 self.teardown_thread() 93 94 def teardown_class(self): 95 if "AccessPoints" in self.user_params: 96 self.ap.close() 97 reset_wifi(self.pri_ad) 98 wifi_toggle_state(self.pri_ad, False) 99 json_result = self.results.json_str() 100 xlsheet(self.pri_ad, json_result) 101 102 def start_iperf_server_on_shell(self, server_port): 103 """Starts iperf server on android device with specified. 104 105 Args: 106 server_port: Port in which server should be started. 107 """ 108 log_path = os.path.join(self.pri_ad.log_path, "iPerf{}".format( 109 server_port)) 110 iperf_server = "iperf3 -s -p {} -J".format(server_port) 111 log_files = [] 112 create_dir(log_path) 113 out_file_name = "IPerfServer,{},{},{}.log".format( 114 server_port, self.tag, log_files) 115 self.tag = self.tag + 1 116 full_out_path = os.path.join(log_path, out_file_name) 117 cmd = "adb -s {} shell {} > {}".format( 118 self.pri_ad.serial, iperf_server, full_out_path) 119 self.iperf_process.append(start_standing_subprocess(cmd)) 120 log_files.append(full_out_path) 121 time.sleep(IPERF_SERVER_WAIT_TIME) 122 123 def stop_iperf_server_on_shell(self): 124 """Stops all the instances of iperf server on shell.""" 125 try: 126 for process in self.iperf_process: 127 stop_standing_subprocess(process) 128 except Exception: 129 self.log.info("Iperf server already killed") 130 131 def run_iperf_and_get_result(self): 132 """Frames iperf command based on test and starts iperf client on 133 host machine. 134 """ 135 self.flag_list = [] 136 self.iperf_process = [] 137 test_params = self.current_test_name.split("_") 138 139 self.protocol = test_params[-2:-1] 140 self.stream = test_params[-1:] 141 142 if self.protocol[0] == "tcp": 143 self.iperf_args = "-t {} -p {} {} -J".format( 144 self.iperf["duration"], self.iperf["port_1"], 145 self.iperf["tcp_window_size"]) 146 else: 147 self.iperf_args = ("-t {} -p {} -u {} --get-server-output -J" 148 .format(self.iperf["duration"], 149 self.iperf["port_1"], 150 self.iperf["udp_bandwidth"])) 151 152 if self.stream[0] == "ul": 153 self.iperf_args += " -R" 154 155 if self.protocol[0] == "tcp" and self.stream[0] == "bidirectional": 156 self.bidirectional_args = "-t {} -p {} {} -R -J".format( 157 self.iperf["duration"], self.iperf["port_2"], 158 self.iperf["tcp_window_size"]) 159 else: 160 self.bidirectional_args = ("-t {} -p {} -u {} --get-server-output" 161 " -J".format(self.iperf["duration"], 162 self.iperf["port_2"], 163 self.iperf["udp_bandwidth"] 164 )) 165 166 if self.stream[0] == "bidirectional": 167 self.start_iperf_server_on_shell(self.iperf["port_2"]) 168 self.start_iperf_server_on_shell(self.iperf["port_1"]) 169 170 if self.stream[0] == "bidirectional": 171 args = [ 172 lambda: self.run_iperf(self.iperf_args, self.iperf["port_1"]), 173 lambda: self.run_iperf(self.bidirectional_args, 174 self.iperf["port_2"]) 175 ] 176 self.run_thread(args) 177 else: 178 args = [ 179 lambda: self.run_iperf(self.iperf_args, self.iperf["port_1"]) 180 ] 181 self.run_thread(args) 182 183 def run_iperf(self, iperf_args, server_port): 184 """Gets android device ip and start iperf client from host machine to 185 that ip and parses the iperf result. 186 187 Args: 188 iperf_args: Iperf parameters to run traffic. 189 server_port: Iperf port to start client. 190 """ 191 ip = get_phone_ip(self.pri_ad) 192 iperf_client = IPerfClient(server_port, ip, self.pri_ad.log_path) 193 result = iperf_client.start(iperf_args) 194 try: 195 sent, received = iperf_result(result, self.stream) 196 self.received.append(str(round(received, 2)) + "Mb/s") 197 self.log.info("Sent: {} Mb/s, Received: {} Mb/s".format( 198 sent, received)) 199 self.flag_list.append(True) 200 201 except TypeError: 202 self.log.error("Iperf failed/stopped.") 203 self.flag_list.append(False) 204 self.received.append("Iperf Failed") 205 206 self.iperf_result[self.current_test_name] = self.received 207 208 def on_fail(self, record, test_name, begin_time): 209 self.log.info( 210 "Test {} failed, Fetching Btsnoop logs and bugreport".format( 211 test_name)) 212 take_btsnoop_logs(self.android_devices, self, test_name) 213 self._take_bug_report(test_name, begin_time) 214 record.extras = self.received 215 216 def _on_fail(self, record): 217 """Proxy function to guarantee the base implementation of on_fail is 218 called. 219 220 Args: 221 record: The records.TestResultRecord object for the failed test 222 case. 223 """ 224 if record.details: 225 self.log.error(record.details) 226 self.log.info(RESULT_LINE_TEMPLATE, record.test_name, record.result) 227 self.on_fail(record, record.test_name, record.log_begin_time) 228 229 def _on_pass(self, record): 230 """Proxy function to guarantee the base implementation of on_pass is 231 called. 232 233 Args: 234 record: The records.TestResultRecord object for the passed test 235 case. 236 """ 237 msg = record.details 238 if msg: 239 self.log.info(msg) 240 self.log.info(RESULT_LINE_TEMPLATE, record.test_name, record.result) 241 record.extras = self.received 242 243 def run_thread(self, kwargs): 244 """Convenience function to start thread. 245 246 Args: 247 kwargs: Function object to start in thread. 248 """ 249 for function in kwargs: 250 self.thread = threading.Thread(target=function) 251 self.thread_list.append(self.thread) 252 self.thread.start() 253 254 def teardown_result(self): 255 """Convenience function to join thread and fetch iperf result.""" 256 for thread_id in self.thread_list: 257 if thread_id.is_alive(): 258 thread_id.join() 259 self.stop_iperf_server_on_shell() 260 if False in self.flag_list: 261 return False 262 return True 263 264 def teardown_thread(self): 265 """Convenience function to join thread.""" 266 for thread_id in self.thread_list: 267 if thread_id.is_alive(): 268 thread_id.join() 269 self.stop_iperf_server_on_shell() 270 271 def push_music_to_android_device(self, ad): 272 """Add music to Android device as specified by the test config 273 274 Args: 275 ad: Android device 276 277 Returns: 278 True on success, False on failure 279 """ 280 self.log.info("Pushing music to the Android device") 281 music_path_str = "music_file" 282 android_music_path = "/sdcard/Music/" 283 if music_path_str not in self.user_params: 284 self.log.error("Need music for audio testcases") 285 return False 286 music_path = self.user_params[music_path_str] 287 if type(music_path) is list: 288 self.log.info("Media ready to push as is.") 289 if type(music_path) is list: 290 for item in music_path: 291 self.music_file_to_play = item 292 ad.adb.push("{} {}".format(item, android_music_path)) 293 return True 294 295 def avrcp_actions(self): 296 """Performs avrcp controls like volume up, volume down, skip next and 297 skip previous. 298 299 Returns: True if successful, otherwise False. 300 """ 301 #TODO: Validate the success state of functionalities performed. 302 self.audio_receiver.volume_up() 303 time.sleep(2) 304 self.audio_receiver.volume_down() 305 time.sleep(2) 306 self.audio_receiver.skip_next() 307 time.sleep(2) 308 self.audio_receiver.skip_previous() 309 time.sleep(2) 310 return True 311 312 def change_volume(self): 313 """Changes volume with HFP call. 314 315 Returns: True if successful, otherwise False. 316 """ 317 self.audio_receiver.volume_up() 318 time.sleep(2) 319 self.audio_receiver.volume_down() 320 time.sleep(2) 321 return True 322
