1#!/usr/bin/env python3.4 2# 3# Copyright 2017 - Google 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of 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, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16""" 17 Test Script for Telephony Mobility Stress Test 18""" 19 20import collections 21import random 22import time 23from acts.asserts import explicit_pass 24from acts.asserts import fail 25from acts.test_decorators import test_tracker_info 26from acts.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest 27from acts.test_utils.tel.tel_atten_utils import set_rssi 28from acts.test_utils.tel.tel_defines import CELL_WEAK_RSSI_VALUE 29from acts.test_utils.tel.tel_defines import CELL_STRONG_RSSI_VALUE 30from acts.test_utils.tel.tel_defines import MAX_RSSI_RESERVED_VALUE 31from acts.test_utils.tel.tel_defines import MIN_RSSI_RESERVED_VALUE 32from acts.test_utils.tel.tel_defines import WFC_MODE_CELLULAR_PREFERRED 33from acts.test_utils.tel.tel_defines import WFC_MODE_DISABLED 34from acts.test_utils.tel.tel_defines import WFC_MODE_WIFI_PREFERRED 35from acts.test_utils.tel.tel_defines import WFC_MODE_CELLULAR_PREFERRED 36from acts.test_utils.tel.tel_defines import WIFI_WEAK_RSSI_VALUE 37from acts.test_utils.tel.tel_test_utils import active_file_download_test 38from acts.test_utils.tel.tel_test_utils import call_setup_teardown 39from acts.test_utils.tel.tel_test_utils import ensure_phone_default_state 40from acts.test_utils.tel.tel_test_utils import ensure_phone_subscription 41from acts.test_utils.tel.tel_test_utils import ensure_phones_idle 42from acts.test_utils.tel.tel_test_utils import ensure_wifi_connected 43from acts.test_utils.tel.tel_test_utils import hangup_call 44from acts.test_utils.tel.tel_test_utils import is_voice_attached 45from acts.test_utils.tel.tel_test_utils import run_multithread_func 46from acts.test_utils.tel.tel_test_utils import set_wfc_mode 47from acts.test_utils.tel.tel_test_utils import sms_send_receive_verify 48from acts.test_utils.tel.tel_test_utils import start_qxdm_loggers 49from acts.test_utils.tel.tel_test_utils import mms_send_receive_verify 50from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_3g 51from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_2g 52from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_csfb 53from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_iwlan 54from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_volte 55from acts.test_utils.tel.tel_voice_utils import phone_setup_csfb 56from acts.test_utils.tel.tel_voice_utils import phone_setup_iwlan 57from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_3g 58from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_2g 59from acts.test_utils.tel.tel_voice_utils import phone_setup_volte 60from acts.test_utils.tel.tel_voice_utils import get_current_voice_rat 61 62from acts.logger import epoch_to_log_line_timestamp 63from acts.utils import get_current_epoch_time 64from acts.utils import rand_ascii_str 65 66from TelWifiVoiceTest import TelWifiVoiceTest 67from TelWifiVoiceTest import ATTEN_NAME_FOR_WIFI_2G 68from TelWifiVoiceTest import ATTEN_NAME_FOR_WIFI_5G 69from TelWifiVoiceTest import ATTEN_NAME_FOR_CELL_3G 70from TelWifiVoiceTest import ATTEN_NAME_FOR_CELL_4G 71 72import socket 73from acts.controllers.sl4a_lib.rpc_client import Sl4aProtocolError 74 75IGNORE_EXCEPTIONS = (BrokenPipeError, Sl4aProtocolError) 76EXCEPTION_TOLERANCE = 20 77 78 79class TelLiveMobilityStressTest(TelWifiVoiceTest): 80 def setup_class(self): 81 super().setup_class() 82 #super(TelWifiVoiceTest, self).setup_class() 83 self.user_params["telephony_auto_rerun"] = False 84 self.max_phone_call_duration = int( 85 self.user_params.get("max_phone_call_duration", 600)) 86 self.max_sleep_time = int(self.user_params.get("max_sleep_time", 120)) 87 self.max_run_time = int(self.user_params.get("max_run_time", 7200)) 88 self.max_sms_length = int(self.user_params.get("max_sms_length", 1000)) 89 self.max_mms_length = int(self.user_params.get("max_mms_length", 160)) 90 self.crash_check_interval = int( 91 self.user_params.get("crash_check_interval", 300)) 92 self.signal_change_interval = int( 93 self.user_params.get("signal_change_interval", 10)) 94 self.signal_change_step = int( 95 self.user_params.get("signal_change_step", 5)) 96 self.min_sms_length = int(self.user_params.get("min_sms_length", 1)) 97 self.min_mms_length = int(self.user_params.get("min_mms_length", 1)) 98 self.min_phone_call_duration = int( 99 self.user_params.get("min_phone_call_duration", 10)) 100 self.dut = self.android_devices[0] 101 self.helper = self.android_devices[1] 102 103 return True 104 105 def on_fail(self, test_name, begin_time): 106 pass 107 108 def _setup_volte_wfc_wifi_preferred(self): 109 return self._wfc_phone_setup( 110 False, WFC_MODE_WIFI_PREFERRED, volte_mode=True) 111 112 def _setup_volte_wfc_cell_preferred(self): 113 return self._wfc_phone_setup( 114 False, WFC_MODE_CELLULAR_PREFERRED, volte_mode=True) 115 116 def _setup_csfb_wfc_wifi_preferred(self): 117 return self._wfc_phone_setup( 118 False, WFC_MODE_WIFI_PREFERRED, volte_mode=False) 119 120 def _setup_csfb_wfc_cell_preferred(self): 121 return self._wfc_phone_setup( 122 False, WFC_MODE_CELLULAR_PREFERRED, volte_mode=False) 123 124 def _setup_volte_wfc_disabled(self): 125 return self._wfc_phone_setup(False, WFC_MODE_DISABLED, volte_mode=True) 126 127 def _setup_csfb_wfc_disabled(self): 128 return self._wfc_phone_setup( 129 False, WFC_MODE_DISABLED, volte_mode=False) 130 131 def _send_message(self, ads): 132 selection = random.randrange(0, 2) 133 message_type_map = {0: "SMS", 1: "MMS"} 134 max_length_map = {0: self.max_sms_length, 1: self.max_mms_length} 135 min_length_map = {0: self.min_sms_length, 1: self.min_mms_length} 136 length = random.randrange(min_length_map[selection], 137 max_length_map[selection] + 1) 138 text = rand_ascii_str(length) 139 message_content_map = {0: [text], 1: [("Mms Message", text, None)]} 140 message_func_map = { 141 0: sms_send_receive_verify, 142 1: mms_send_receive_verify 143 } 144 message_type = message_type_map[selection] 145 self.result_info["Total %s" % message_type] += 1 146 begin_time = get_current_epoch_time() 147 start_qxdm_loggers(self.log, self.android_devices) 148 if not message_func_map[selection](self.log, ads[0], ads[1], 149 message_content_map[selection]): 150 self.log.error("%s of length %s from %s to %s fails", message_type, 151 length, ads[0].serial, ads[1].serial) 152 self.result_info["%s failure" % message_type] += 1 153 if message_type == "SMS" or self.result_info["%s failure" % 154 message_type] == 1: 155 self._take_bug_report("%s_%s_failure" % (self.test_name, 156 message_type), 157 begin_time) 158 return False 159 else: 160 self.log.info("%s of length %s from %s to %s succeed", 161 message_type_map[selection], length, ads[0].serial, 162 ads[1].serial) 163 return True 164 165 def _make_phone_call(self, ads): 166 self.result_info["Total Calls"] += 1 167 begin_time = get_current_epoch_time() 168 start_qxdm_loggers(self.log, self.android_devices) 169 if not call_setup_teardown( 170 self.log, 171 ads[0], 172 ads[1], 173 ad_hangup=ads[random.randrange(0, 2)], 174 wait_time_in_call=random.randrange( 175 self.min_phone_call_duration, 176 self.max_phone_call_duration)): 177 self.log.error("Call setup and teardown failed.") 178 self.result_info["Call Failure"] += 1 179 self._take_bug_report("%s_call_failure" % self.test_name, 180 begin_time) 181 return False 182 self.log.info("Call setup and teardown succeed.") 183 return True 184 185 def crash_check_test(self): 186 failure = 0 187 while time.time() < self.finishing_time: 188 self.dut.log.info(dict(self.result_info)) 189 try: 190 begin_time = get_current_epoch_time() 191 time.sleep(self.crash_check_interval) 192 crash_report = self.dut.check_crash_report( 193 "checking_crash", begin_time, True) 194 if crash_report: 195 self.dut.log.error("Find new crash reports %s", 196 crash_report) 197 failure += 1 198 self.result_info["Crashes"] += 1 199 except IGNORE_EXCEPTION as e: 200 self.log.error("Exception error %s", str(e)) 201 self.result_info["Exception Errors"] += 1 202 if self.result_info["Exception Errors"] > EXCEPTION_TOLERANCE: 203 return False 204 except Exception as e: 205 return False 206 self.dut.log.info("Crashes found: %s", failure) 207 if failure: 208 return False 209 else: 210 return True 211 212 def environment_change_4g_wifi(self): 213 #block cell 3G, WIFI 2G 214 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_3G], 0, 215 MIN_RSSI_RESERVED_VALUE) 216 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_WIFI_2G], 0, 217 MIN_RSSI_RESERVED_VALUE) 218 while time.time() < self.finishing_time: 219 #set strong wifi 5G and LTE 220 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_WIFI_5G], 0, 221 MAX_RSSI_RESERVED_VALUE) 222 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_4G], 0, 223 MAX_RSSI_RESERVED_VALUE) 224 #gratually decrease wifi 5g 225 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_WIFI_5G], 226 self.wifi_rssi_with_no_atten, MIN_RSSI_RESERVED_VALUE, 227 self.signal_change_step, self.signal_change_interval) 228 #gratually increase wifi 5g 229 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_WIFI_5G], 230 MIN_RSSI_RESERVED_VALUE, MAX_RSSI_RESERVED_VALUE, 231 self.signal_change_step, self.signal_change_interval) 232 233 #gratually decrease cell 4G 234 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_4G], 235 self.cell_rssi_with_no_atten, CELL_WEAK_RSSI_VALUE, 236 self.signal_change_step, self.signal_change_interval) 237 #gradtually increase cell 4G 238 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_4G], 239 MIN_RSSI_RESERVED_VALUE, MAX_RSSI_RESERVED_VALUE, 240 self.signal_change_step, self.signal_change_interval) 241 return "" 242 243 def environment_change_4g_3g(self): 244 #block wifi 2G and 5G 245 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_WIFI_2G], 0, 246 MIN_RSSI_RESERVED_VALUE) 247 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_WIFI_5G], 0, 248 MIN_RSSI_RESERVED_VALUE) 249 while time.time() < self.finishing_time: 250 #set strong cell 4G and 3G 251 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_4G], 0, 252 MAX_RSSI_RESERVED_VALUE) 253 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_3G], 0, 254 MAX_RSSI_RESERVED_VALUE) 255 #gratually decrease cell 4G 256 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_4G], 257 self.cell_rssi_with_no_atten, MIN_RSSI_RESERVED_VALUE, 258 self.signal_change_step, self.signal_change_interval) 259 #gradtually increase cell 4G 260 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_4G], 261 MIN_RSSI_RESERVED_VALUE, MAX_RSSI_RESERVED_VALUE, 262 self.signal_change_step, self.signal_change_interval) 263 #gratually decrease cell 3G 264 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_3G], 265 self.cell_rssi_with_no_atten, MIN_RSSI_RESERVED_VALUE, 266 self.signal_change_step, self.signal_change_interval) 267 #gradtually increase cell 3G 268 set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_3G], 269 MIN_RSSI_RESERVED_VALUE, MAX_RSSI_RESERVED_VALUE, 270 self.signal_change_step, self.signal_change_interval) 271 272 return "" 273 274 def call_test(self): 275 failure = 0 276 total_count = 0 277 while time.time() < self.finishing_time: 278 try: 279 ads = [self.dut, self.helper] 280 random.shuffle(ads) 281 total_count += 1 282 # Current Voice RAT 283 self.dut.log.info("Current Voice RAT is %s", 284 get_current_voice_rat(self.log, self.dut)) 285 self.helper.log.info("Current Voice RAT is %s", 286 get_current_voice_rat( 287 self.log, self.helper)) 288 if not self._make_phone_call(ads): 289 failure += 1 290 self.dut.droid.goToSleepNow() 291 time.sleep(random.randrange(0, self.max_sleep_time)) 292 except IGNORE_EXCEPTIONS as e: 293 self.log.error("Exception error %s", str(e)) 294 self.result_info["Exception Errors"] += 1 295 if self.result_info["Exception Errors"] > EXCEPTION_TOLERANCE: 296 self.log.error("Too many exception errors %s", 297 IGNORE_EXCEPTIONS) 298 return False 299 except Exception as e: 300 self.log.error(e) 301 return False 302 self.dut.log.info("Call test failure: %s/%s", failure, total_count) 303 if failure: 304 return False 305 else: 306 return True 307 308 def message_test(self): 309 failure = 0 310 total_count = 0 311 while time.time() < self.finishing_time: 312 try: 313 ads = [self.dut, self.helper] 314 random.shuffle(ads) 315 total_count += 1 316 if not self._send_message(ads): 317 failure += 1 318 self.dut.droid.goToSleepNow() 319 time.sleep(random.randrange(0, self.max_sleep_time)) 320 except IGNORE_EXCEPTIONS as e: 321 self.log.error("Exception error %s", str(e)) 322 self.result_info["Exception Errors"] += 1 323 if self.result_info["Exception Errors"] > EXCEPTION_TOLERANCE: 324 self.log.error("Too many exception errors %s", 325 IGNORE_EXCEPTIONS) 326 return False 327 except Exception as e: 328 self.log.error(e) 329 return False 330 self.dut.log.info("Messaging test failure: %s/%s", failure, 331 total_count) 332 if failure / total_count > 0.1: 333 return False 334 else: 335 return True 336 337 def data_test(self): 338 failure = 0 339 total_count = 0 340 #file_names = ["5MB", "10MB", "20MB", "50MB", "200MB", "512MB", "1GB"] 341 #wifi download is very slow in lab, limit the file size upto 200MB 342 file_names = ["5MB", "10MB", "20MB", "50MB", "200MB"] 343 while time.time() < self.finishing_time: 344 total_count += 1 345 begin_time = get_current_epoch_time() 346 start_qxdm_loggers(self.log, self.android_devices) 347 try: 348 self.dut.log.info(dict(self.result_info)) 349 self.result_info["Total file download"] += 1 350 selection = random.randrange(0, len(file_names)) 351 file_name = file_names[selection] 352 if not active_file_download_test(self.log, self.dut, 353 file_name): 354 self.result_info["File download failure"] += 1 355 failure += 1 356 if self.result_info["File download failure"] == 1: 357 self._take_bug_report( 358 "%s_file_download_failure" % self.test_name, 359 begin_time) 360 self.dut.droid.goToSleepNow() 361 time.sleep(random.randrange(0, self.max_sleep_time)) 362 except IGNORE_EXCEPTIONS as e: 363 self.log.error("Exception error %s", str(e)) 364 self.result_info["Exception Errors"] += 1 365 if self.result_info["Exception Errors"] > EXCEPTION_TOLERANCE: 366 self.log.error("Too many exception error %s", 367 IGNORE_EXCEPTIONS) 368 return False 369 except Exception as e: 370 self.log.error(e) 371 return False 372 self.dut.log.info("File download test failure: %s/%s", failure, 373 total_count) 374 if failure / total_count > 0.1: 375 return False 376 else: 377 return True 378 379 def parallel_tests(self, change_env_func, setup_func=None): 380 if setup_func and not setup_func(): 381 self.log.error("Test setup %s failed", setup_func.__name__) 382 return False 383 self.result_info = collections.defaultdict(int) 384 self.finishing_time = time.time() + self.max_run_time 385 results = run_multithread_func(self.log, [(self.call_test, []), ( 386 self.message_test, []), (self.data_test, []), ( 387 self.crash_check_test, []), (change_env_func, [])]) 388 result_message = "%s" % dict(self.result_info) 389 self.log.info(result_message) 390 if all(results): 391 explicit_pass(result_message) 392 else: 393 fail(result_message) 394 395 """ Tests Begin """ 396 397 @test_tracker_info(uuid="6fcba97c-3572-47d7-bcac-9608f1aa5304") 398 @TelephonyBaseTest.tel_test_wrap 399 def test_volte_wfc_wifi_preferred_parallel_stress(self): 400 return self.parallel_tests( 401 self.environment_change_4g_wifi, 402 setup_func=self._setup_volte_wfc_wifi_preferred) 403 404 @test_tracker_info(uuid="df78a9a8-2a14-40bf-a7aa-719502f975be") 405 @TelephonyBaseTest.tel_test_wrap 406 def test_volte_wfc_cell_preferred_parallel_stress(self): 407 return self.parallel_tests( 408 self.environment_change_4g_wifi, 409 setup_func=self._setup_volte_wfc_cell_preferred) 410 411 @test_tracker_info(uuid="4cb47315-c420-44c2-ac47-a8bdca6d0e25") 412 @TelephonyBaseTest.tel_test_wrap 413 def test_csfb_wfc_wifi_preferred_parallel_stress(self): 414 return self.parallel_tests( 415 self.environment_change_4g_wifi, 416 setup_func=self._setup_csfb_wfc_wifi_preferred) 417 418 @test_tracker_info(uuid="92821ef7-542a-4139-b3b0-22278e2b06c4") 419 @TelephonyBaseTest.tel_test_wrap 420 def test_csfb_wfc_cell_preferred_parallel_stress(self): 421 return self.parallel_tests( 422 self.self.environment_change_4g_wifi, 423 setup_func=self._setup_csfb_wfc_cell_preferred) 424 425 @test_tracker_info(uuid="dd23923e-ebbc-461e-950a-0657e845eacf") 426 @TelephonyBaseTest.tel_test_wrap 427 def test_4g_volte_3g_parallel_stress(self): 428 return self.parallel_tests( 429 self.environment_change_4g_3g, 430 setup_func=self._setup_volte_wfc_disabled) 431 432 @test_tracker_info(uuid="faef9384-a5b0-4640-8cfa-f9f34ce6d977") 433 @TelephonyBaseTest.tel_test_wrap 434 def test_4g_csfb_3g_parallel_stress(self): 435 return self.parallel_tests( 436 self.environment_change_4g_3g, 437 setup_func=self._setup_csfb_wfc_disabled) 438 439 """ Tests End """ 440