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 Stress Call Test 18""" 19 20import collections 21import json 22import os 23import random 24import time 25 26from acts import signals 27from acts import utils 28from acts.libs.proc import job 29from acts.test_decorators import test_tracker_info 30from acts.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest 31from acts.test_utils.tel.tel_defines import CAPABILITY_VOLTE 32from acts.test_utils.tel.tel_defines import CAPABILITY_WFC 33from acts.test_utils.tel.tel_defines import GEN_3G 34from acts.test_utils.tel.tel_defines import GEN_4G 35from acts.test_utils.tel.tel_defines import INCALL_UI_DISPLAY_BACKGROUND 36from acts.test_utils.tel.tel_defines import MAX_WAIT_TIME_SMS_RECEIVE 37from acts.test_utils.tel.tel_defines import NETWORK_MODE_WCDMA_ONLY 38from acts.test_utils.tel.tel_defines import NETWORK_MODE_GLOBAL 39from acts.test_utils.tel.tel_defines import NETWORK_MODE_CDMA 40from acts.test_utils.tel.tel_defines import NETWORK_MODE_GSM_ONLY 41from acts.test_utils.tel.tel_defines import NETWORK_MODE_TDSCDMA_GSM_WCDMA 42from acts.test_utils.tel.tel_defines import RAT_LTE 43from acts.test_utils.tel.tel_defines import RAT_UNKNOWN 44from acts.test_utils.tel.tel_defines import WAIT_TIME_AFTER_MODE_CHANGE 45from acts.test_utils.tel.tel_defines import WFC_MODE_CELLULAR_PREFERRED 46from acts.test_utils.tel.tel_defines import WFC_MODE_WIFI_PREFERRED 47from acts.test_utils.tel.tel_defines import WAIT_TIME_CHANGE_MESSAGE_SUB_ID 48from acts.test_utils.tel.tel_defines import WAIT_TIME_CHANGE_VOICE_SUB_ID 49from acts.test_utils.tel.tel_lookup_tables import is_rat_svd_capable 50from acts.test_utils.tel.tel_test_utils import STORY_LINE 51from acts.test_utils.tel.tel_test_utils import active_file_download_test 52from acts.test_utils.tel.tel_test_utils import is_phone_in_call 53from acts.test_utils.tel.tel_test_utils import call_setup_teardown 54from acts.test_utils.tel.tel_test_utils import check_is_wifi_connected 55from acts.test_utils.tel.tel_test_utils import ensure_network_generation_for_subscription 56from acts.test_utils.tel.tel_test_utils import ensure_wifi_connected 57from acts.test_utils.tel.tel_test_utils import force_connectivity_metrics_upload 58from acts.test_utils.tel.tel_test_utils import get_device_epoch_time 59from acts.test_utils.tel.tel_test_utils import get_telephony_signal_strength 60from acts.test_utils.tel.tel_test_utils import hangup_call 61from acts.test_utils.tel.tel_test_utils import hangup_call_by_adb 62from acts.test_utils.tel.tel_test_utils import initiate_call 63from acts.test_utils.tel.tel_test_utils import last_call_drop_reason 64from acts.test_utils.tel.tel_test_utils import run_multithread_func 65from acts.test_utils.tel.tel_test_utils import set_wfc_mode 66from acts.test_utils.tel.tel_test_utils import sms_send_receive_verify 67from acts.test_utils.tel.tel_test_utils import start_qxdm_loggers 68from acts.test_utils.tel.tel_test_utils import start_adb_tcpdump 69from acts.test_utils.tel.tel_test_utils import synchronize_device_time 70from acts.test_utils.tel.tel_test_utils import mms_send_receive_verify 71from acts.test_utils.tel.tel_test_utils import set_preferred_network_mode_pref 72from acts.test_utils.tel.tel_test_utils import verify_internet_connection 73from acts.test_utils.tel.tel_test_utils import verify_internet_connection_by_ping 74from acts.test_utils.tel.tel_test_utils import verify_http_connection 75from acts.test_utils.tel.tel_test_utils import wait_for_call_id_clearing 76from acts.test_utils.tel.tel_test_utils import wait_for_data_connection 77from acts.test_utils.tel.tel_test_utils import wait_for_in_call_active 78from acts.test_utils.tel.tel_test_utils import wifi_toggle_state 79from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_3g 80from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_2g 81from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_csfb 82from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_iwlan 83from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_volte 84from acts.test_utils.tel.tel_voice_utils import phone_setup_csfb 85from acts.test_utils.tel.tel_voice_utils import phone_setup_iwlan 86from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_3g 87from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_2g 88from acts.test_utils.tel.tel_voice_utils import phone_setup_volte 89from acts.test_utils.tel.tel_voice_utils import phone_idle_iwlan 90from acts.test_utils.tel.tel_voice_utils import phone_idle_volte 91from acts.test_utils.tel.tel_voice_utils import get_current_voice_rat 92from acts.test_utils.tel.tel_subscription_utils import get_default_data_sub_id 93from acts.test_utils.tel.tel_subscription_utils import get_outgoing_message_sub_id 94from acts.test_utils.tel.tel_subscription_utils import get_outgoing_voice_sub_id 95from acts.test_utils.tel.tel_subscription_utils import get_incoming_voice_sub_id 96from acts.test_utils.tel.tel_subscription_utils import get_incoming_message_sub_id 97from acts.test_utils.tel.tel_subscription_utils import get_subid_from_slot_index 98from acts.test_utils.tel.tel_subscription_utils import set_subid_for_data 99from acts.test_utils.tel.tel_subscription_utils import set_subid_for_message 100from acts.test_utils.tel.tel_subscription_utils import set_subid_for_outgoing_call 101from acts.utils import get_current_epoch_time 102from acts.utils import rand_ascii_str 103 104EXCEPTION_TOLERANCE = 5 105BINDER_LOGS = ["/sys/kernel/debug/binder"] 106 107 108class TelLiveStressTest(TelephonyBaseTest): 109 def setup_class(self): 110 super(TelLiveStressTest, self).setup_class() 111 self.dut = self.android_devices[0] 112 self.single_phone_test = self.user_params.get("single_phone_test", 113 False) 114 # supported file download methods: chrome, sl4a, curl 115 self.file_download_method = self.user_params.get( 116 "file_download_method", "curl") 117 self.get_binder_logs = self.user_params.get("get_binder_logs", False) 118 if len(self.android_devices) == 1: 119 self.single_phone_test = True 120 if self.single_phone_test: 121 self.android_devices = self.android_devices[:1] 122 self.call_server_number = self.user_params.get( 123 "call_server_number", STORY_LINE) 124 if self.file_download_method == "sl4a": 125 # with single device, do not use sl4a file download 126 # due to stability issue 127 self.file_download_method = "curl" 128 else: 129 self.android_devices = self.android_devices[:2] 130 for ad in self.android_devices: 131 ad.adb.shell("setprop nfc.debug_enable 1") 132 if self.user_params.get("turn_on_tcpdump", True): 133 start_adb_tcpdump(ad, interface="any", mask="all") 134 self.user_params["telephony_auto_rerun"] = 0 135 self.phone_call_iteration = int( 136 self.user_params.get("phone_call_iteration", 500)) 137 self.max_phone_call_duration = int( 138 self.user_params.get("max_phone_call_duration", 600)) 139 self.min_sleep_time = int(self.user_params.get("min_sleep_time", 30)) 140 self.max_sleep_time = int(self.user_params.get("max_sleep_time", 60)) 141 self.max_run_time = int(self.user_params.get("max_run_time", 14400)) 142 self.max_sms_length = int(self.user_params.get("max_sms_length", 1000)) 143 self.max_mms_length = int(self.user_params.get("max_mms_length", 160)) 144 self.min_sms_length = int(self.user_params.get("min_sms_length", 1)) 145 self.min_mms_length = int(self.user_params.get("min_mms_length", 1)) 146 self.min_phone_call_duration = int( 147 self.user_params.get("min_phone_call_duration", 10)) 148 self.crash_check_interval = int( 149 self.user_params.get("crash_check_interval", 300)) 150 self.dut_incall = False 151 self.dsds_esim = self.user_params.get("dsds_esim", False) 152 telephony_info = getattr(self.dut, "telephony", {}) 153 self.dut_capabilities = telephony_info.get("capabilities", []) 154 self.dut_wfc_modes = telephony_info.get("wfc_modes", []) 155 self.gps_log_file = self.user_params.get("gps_log_file", None) 156 return True 157 158 def setup_test(self): 159 super(TelLiveStressTest, self).setup_test() 160 self.result_info = collections.defaultdict(int) 161 self._init_perf_json() 162 self.internet_connection_check_method = verify_internet_connection 163 164 def on_fail(self, test_name, begin_time): 165 pass 166 167 def _setup_wfc(self): 168 for ad in self.android_devices: 169 if not ensure_wifi_connected( 170 self.log, 171 ad, 172 self.wifi_network_ssid, 173 self.wifi_network_pass, 174 retries=3): 175 ad.log.error("Bringing up Wifi connection fails.") 176 return False 177 ad.log.info("Phone WIFI is connected successfully.") 178 if not set_wfc_mode(self.log, ad, WFC_MODE_WIFI_PREFERRED): 179 ad.log.error("Phone failed to enable Wifi-Calling.") 180 return False 181 ad.log.info("Phone is set in Wifi-Calling successfully.") 182 if not phone_idle_iwlan(self.log, ad): 183 ad.log.error("Phone is not in WFC enabled state.") 184 return False 185 ad.log.info("Phone is in WFC enabled state.") 186 return True 187 188 def _setup_wfc_apm(self): 189 for ad in self.android_devices: 190 if not phone_setup_iwlan( 191 self.log, ad, True, WFC_MODE_CELLULAR_PREFERRED, 192 self.wifi_network_ssid, self.wifi_network_pass): 193 ad.log.error("Failed to setup WFC.") 194 return False 195 return True 196 197 def _setup_lte_volte_enabled(self): 198 for ad in self.android_devices: 199 if not phone_setup_volte(self.log, ad): 200 ad.log.error("Phone failed to enable VoLTE.") 201 return False 202 ad.log.info("Phone VOLTE is enabled successfully.") 203 return True 204 205 def _setup_lte_volte_disabled(self): 206 for ad in self.android_devices: 207 if not phone_setup_csfb(self.log, ad): 208 ad.log.error("Phone failed to setup CSFB.") 209 return False 210 ad.log.info("Phone VOLTE is disabled successfully.") 211 return True 212 213 def _setup_3g(self): 214 for ad in self.android_devices: 215 if not phone_setup_voice_3g(self.log, ad): 216 ad.log.error("Phone failed to setup 3g.") 217 return False 218 ad.log.info("Phone RAT 3G is enabled successfully.") 219 return True 220 221 def _setup_2g(self): 222 for ad in self.android_devices: 223 if not phone_setup_voice_2g(self.log, ad): 224 ad.log.error("Phone failed to setup 2g.") 225 return False 226 ad.log.info("RAT 2G is enabled successfully.") 227 return True 228 229 def _send_message(self, max_wait_time=2 * MAX_WAIT_TIME_SMS_RECEIVE): 230 slot_id_rx = None 231 if self.single_phone_test: 232 ads = [self.dut, self.dut] 233 else: 234 ads = self.android_devices[:] 235 random.shuffle(ads) 236 slot_id = random.randint(0,1) 237 if self.dsds_esim: 238 sub_id = get_subid_from_slot_index(self.log, ads[0], slot_id) 239 ads[0].log.info("Message - MO - slot_Id %d", slot_id) 240 set_subid_for_message(ads[0], sub_id) 241 time.sleep(WAIT_TIME_CHANGE_MESSAGE_SUB_ID) 242 slot_id_rx = random.randint(0,1) 243 ads[1].log.info("Message - MT - slot_id %d", slot_id_rx) 244 selection = random.randrange(0, 2) 245 message_type_map = {0: "SMS", 1: "MMS"} 246 max_length_map = {0: self.max_sms_length, 1: self.max_mms_length} 247 min_length_map = {0: self.min_sms_length, 1: self.min_mms_length} 248 length = random.randrange(min_length_map[selection], 249 max_length_map[selection] + 1) 250 message_func_map = { 251 0: sms_send_receive_verify, 252 1: mms_send_receive_verify 253 } 254 rat = self.dut.adb.getprop("gsm.network.type") 255 if "," in rat: 256 if self.dsds_esim: 257 rat = rat.split(',')[slot_id] 258 else: 259 rat = rat.split(',')[0] 260 self.dut.log.info("Network in RAT %s", rat) 261 if self.dut_incall and not is_rat_svd_capable(rat.upper()): 262 self.dut.log.info("In call data not supported, test SMS only") 263 selection = 0 264 message_type = message_type_map[selection] 265 the_number = self.result_info["%s Total" % message_type] + 1 266 begin_time = get_device_epoch_time(self.dut) 267 test_name = "%s_No_%s_%s" % (self.test_name, the_number, message_type) 268 start_qxdm_loggers(self.log, self.android_devices) 269 log_msg = "[Test Case] %s" % test_name 270 self.log.info("%s begin", log_msg) 271 for ad in self.android_devices: 272 if self.user_params.get("turn_on_tcpdump", True): 273 start_adb_tcpdump(ad, interface="any", mask="all") 274 if not getattr(ad, "messaging_droid", None): 275 ad.messaging_droid, ad.messaging_ed = ad.get_droid() 276 ad.messaging_ed.start() 277 else: 278 try: 279 if not ad.messaging_droid.is_live: 280 ad.messaging_droid, ad.messaging_ed = ad.get_droid() 281 ad.messaging_ed.start() 282 else: 283 ad.messaging_ed.clear_all_events() 284 except Exception: 285 ad.log.info("Create new sl4a session for messaging") 286 ad.messaging_droid, ad.messaging_ed = ad.get_droid() 287 ad.messaging_ed.start() 288 ad.messaging_droid.logI("[BEGIN]%s" % log_msg) 289 290 text = "%s:" % test_name 291 text_length = len(text) 292 if length < text_length: 293 text = text[:length] 294 else: 295 text += rand_ascii_str(length - text_length) 296 message_content_map = {0: [text], 1: [(test_name, text, None)]} 297 298 result = message_func_map[selection](self.log, ads[0], ads[1], 299 message_content_map[selection], 300 max_wait_time, 301 slot_id_rx=slot_id_rx) 302 self.log.info("%s end", log_msg) 303 for ad in self.android_devices: 304 ad.messaging_droid.logI("[END]%s" % log_msg) 305 if not result: 306 self.result_info["%s Total" % message_type] += 1 307 if message_type == "SMS": 308 self.log.error("%s fails", log_msg) 309 self.result_info["%s Failure" % message_type] += 1 310 else: 311 rat = self.dut.adb.getprop("gsm.network.type") 312 if "," in rat: 313 if self.dsds_esim: 314 rat = rat.split(',')[slot_id] 315 else: 316 rat = rat.split(',')[0] 317 self.dut.log.info("Network in RAT %s", rat) 318 if self.dut_incall and not is_rat_svd_capable(rat.upper()): 319 self.dut.log.info( 320 "In call data not supported, MMS failure expected") 321 self.result_info["Expected In-call MMS failure"] += 1 322 return True 323 else: 324 self.log.error("%s fails", log_msg) 325 self.result_info["MMS Failure"] += 1 326 try: 327 self._take_bug_report(test_name, begin_time) 328 except Exception as e: 329 self.log.exception(e) 330 return False 331 else: 332 self.result_info["%s Total" % message_type] += 1 333 self.log.info("%s succeed", log_msg) 334 self.result_info["%s Success" % message_type] += 1 335 return True 336 337 def _make_phone_call(self, call_verification_func=None): 338 ads = self.android_devices[:] 339 slot_id_callee = None 340 if not self.single_phone_test: 341 random.shuffle(ads) 342 if self.dsds_esim: 343 slot_id = random.randint(0,1) 344 sub_id = get_subid_from_slot_index(self.log, ads[0], slot_id) 345 ads[0].log.info("Voice - MO - slot_Id %d", slot_id) 346 set_subid_for_outgoing_call(ads[0], sub_id) 347 time.sleep(WAIT_TIME_CHANGE_VOICE_SUB_ID) 348 slot_id_callee = random.randint(0,1) 349 ads[1].log.info("Voice - MT - slot_id %d", slot_id_callee) 350 the_number = self.result_info["Call Total"] + 1 351 duration = random.randrange(self.min_phone_call_duration, 352 self.max_phone_call_duration) 353 result = True 354 test_name = "%s_No_%s_phone_call" % (self.test_name, the_number) 355 log_msg = "[Test Case] %s" % test_name 356 self.log.info("%s for %s seconds begin", log_msg, duration) 357 begin_time = get_device_epoch_time(ads[0]) 358 for ad in self.android_devices: 359 if self.user_params.get("turn_on_tcpdump", True): 360 start_adb_tcpdump(ad, interface="any", mask="all") 361 if not getattr(ad, "droid", None): 362 ad.droid, ad.ed = ad.get_droid() 363 ad.ed.start() 364 else: 365 try: 366 if not ad.droid.is_live: 367 ad.droid, ad.ed = ad.get_droid() 368 ad.ed.start() 369 else: 370 ad.ed.clear_all_events() 371 except Exception: 372 ad.log.info("Create new sl4a session for phone call") 373 ad.droid, ad.ed = ad.get_droid() 374 ad.ed.start() 375 ad.droid.logI("[BEGIN]%s" % log_msg) 376 start_qxdm_loggers(self.log, self.android_devices, begin_time) 377 failure_reasons = set() 378 self.dut_incall = True 379 if self.single_phone_test: 380 call_setup_result = initiate_call( 381 self.log, 382 self.dut, 383 self.call_server_number, 384 incall_ui_display=INCALL_UI_DISPLAY_BACKGROUND 385 ) and wait_for_in_call_active(self.dut, 60, 3) 386 else: 387 call_setup_result = call_setup_teardown( 388 self.log, 389 ads[0], 390 ads[1], 391 ad_hangup=None, 392 verify_caller_func=call_verification_func, 393 verify_callee_func=call_verification_func, 394 wait_time_in_call=0, 395 incall_ui_display=INCALL_UI_DISPLAY_BACKGROUND, 396 slot_id_callee=slot_id_callee) 397 if not call_setup_result: 398 get_telephony_signal_strength(ads[0]) 399 if not self.single_phone_test: 400 get_telephony_signal_strength(ads[1]) 401 call_logs = ads[0].search_logcat( 402 "ActivityManager: START u0 {act=android.intent.action.CALL", 403 begin_time) 404 messaging_logs = ads[0].search_logcat( 405 "com.google.android.apps.messaging/.ui.conversation.ConversationActivity", 406 begin_time) 407 if call_logs and messaging_logs: 408 if (messaging_logs[-1]["datetime_obj"] - 409 call_logs[-1]["datetime_obj"]).seconds < 5: 410 ads[0].log.info( 411 "Call setup failure due to simultaneous activities") 412 self.result_info[ 413 "Call Setup Failure With Simultaneous Activity"] += 1 414 return True 415 self.log.error("%s: Setup Call failed.", log_msg) 416 failure_reasons.add("Setup") 417 result = False 418 else: 419 elapsed_time = 0 420 check_interval = 5 421 while (elapsed_time < duration): 422 check_interval = min(check_interval, duration - elapsed_time) 423 time.sleep(check_interval) 424 elapsed_time += check_interval 425 time_message = "at <%s>/<%s> second." % (elapsed_time, 426 duration) 427 for ad in ads: 428 get_telephony_signal_strength(ad) 429 if not call_verification_func(self.log, ad): 430 ad.log.warning("Call is NOT in correct %s state at %s", 431 call_verification_func.__name__, 432 time_message) 433 if call_verification_func.__name__ == "is_phone_in_call_iwlan": 434 if is_phone_in_call(self.log, ad): 435 if getattr(ad, "data_rat_state_error_count", 436 0) < 1: 437 setattr(ad, "data_rat_state_error_count", 438 1) 439 continue 440 failure_reasons.add("Maintenance") 441 last_call_drop_reason(ad, begin_time) 442 hangup_call(self.log, ads[0]) 443 result = False 444 else: 445 ad.log.info("Call is in correct %s state at %s", 446 call_verification_func.__name__, 447 time_message) 448 if not result: 449 break 450 if not hangup_call(self.log, ads[0]): 451 failure_reasons.add("Teardown") 452 result = False 453 for ad in ads: 454 if not wait_for_call_id_clearing(ad, 455 []) or ad.droid.telecomIsInCall(): 456 ad.log.error("Fail to hang up call") 457 failure_reasons.add("Teardown") 458 result = False 459 self.result_info["Call Total"] += 1 460 for ad in self.android_devices: 461 try: 462 ad.droid.logI("[END]%s" % log_msg) 463 except: 464 pass 465 self.log.info("%s end", log_msg) 466 self.dut_incall = False 467 if not result: 468 self.log.info("%s failed", log_msg) 469 if self.gps_log_file: 470 gps_info = job.run( 471 "tail %s" % self.gps_log_file, ignore_status=True) 472 if gps_info.stdout: 473 gps_log_path = os.path.join(self.log_path, test_name) 474 utils.create_dir(gps_log_path) 475 job.run( 476 "tail %s > %s" % 477 (self.gps_log_file, 478 os.path.join(gps_log_path, "gps_logs.txt")), 479 ignore_status=True) 480 self.log.info("gps log:\n%s", gps_info.stdout) 481 else: 482 self.log.warning("Fail to get gps log %s", 483 self.user_params["gps_log_file"]) 484 for reason in failure_reasons: 485 self.result_info["Call %s Failure" % reason] += 1 486 for ad in ads: 487 log_path = os.path.join(self.log_path, test_name, 488 "%s_binder_logs" % ad.serial) 489 utils.create_dir(log_path) 490 ad.pull_files(BINDER_LOGS, log_path) 491 try: 492 self._take_bug_report(test_name, begin_time) 493 except Exception as e: 494 self.log.exception(e) 495 for ad in ads: 496 if ad.droid.telecomIsInCall(): 497 hangup_call_by_adb(ad) 498 else: 499 self.log.info("%s test succeed", log_msg) 500 self.result_info["Call Success"] += 1 501 if self.result_info["Call Total"] % 50 == 0: 502 for ad in ads: 503 synchronize_device_time(ad) 504 force_connectivity_metrics_upload(ad) 505 if self.get_binder_logs: 506 log_path = os.path.join(self.log_path, 507 "%s_binder_logs" % test_name, 508 "%s_binder_logs" % ad.serial) 509 utils.create_dir(log_path) 510 ad.pull_files(BINDER_LOGS, log_path) 511 return result 512 513 def _prefnetwork_mode_change(self, sub_id): 514 # ModePref change to non-LTE 515 begin_time = get_device_epoch_time(self.dut) 516 start_qxdm_loggers(self.log, self.android_devices) 517 self.result_info["Network Change Request Total"] += 1 518 test_name = "%s_network_change_iter_%s" % ( 519 self.test_name, self.result_info["Network Change Request Total"]) 520 log_msg = "[Test Case] %s" % test_name 521 self.log.info("%s begin", log_msg) 522 self.dut.droid.logI("[BEGIN]%s" % log_msg) 523 network_preference_list = [ 524 NETWORK_MODE_TDSCDMA_GSM_WCDMA, NETWORK_MODE_WCDMA_ONLY, 525 NETWORK_MODE_GLOBAL, NETWORK_MODE_CDMA, NETWORK_MODE_GSM_ONLY 526 ] 527 network_preference = random.choice(network_preference_list) 528 set_preferred_network_mode_pref(self.log, self.dut, sub_id, 529 network_preference) 530 time.sleep(WAIT_TIME_AFTER_MODE_CHANGE) 531 self.dut.log.info("Current Voice RAT is %s", 532 get_current_voice_rat(self.log, self.dut)) 533 534 # ModePref change back to with LTE 535 if not phone_setup_volte(self.log, self.dut): 536 self.dut.log.error("Phone failed to enable VoLTE.") 537 self.result_info["VoLTE Setup Failure"] += 1 538 self.dut.droid.logI("%s end" % log_msg) 539 self.dut.log.info("[END]%s", log_msg) 540 try: 541 self._ad_take_extra_logs(self.dut, test_name, begin_time) 542 self._ad_take_bugreport(self.dut, test_name, begin_time) 543 except Exception as e: 544 self.log.exception(e) 545 return False 546 else: 547 self.result_info["VoLTE Setup Success"] += 1 548 return True 549 550 def _mobile_data_toggling(self, setup="volte"): 551 # ModePref change to non-LTE 552 begin_time = get_device_epoch_time(self.dut) 553 start_qxdm_loggers(self.log, self.android_devices) 554 result = True 555 self.result_info["Data Toggling Request Total"] += 1 556 test_name = "%s_data_toggling_iter_%s" % ( 557 self.test_name, self.result_info["Data Toggling Request Total"]) 558 log_msg = "[Test Case] %s" % test_name 559 self.log.info("%s begin", log_msg) 560 self.dut.droid.logI("[BEGIN]%s" % log_msg) 561 self.dut.adb.shell("svc data disable") 562 time.sleep(WAIT_TIME_AFTER_MODE_CHANGE) 563 self.dut.adb.shell("svc data enable") 564 if not self._check_data(): 565 result = False 566 elif setup == "volte" and not phone_idle_volte(self.log, self.dut): 567 result = False 568 self.dut.droid.logI("%s end" % log_msg) 569 self.dut.log.info("[END]%s", log_msg) 570 if not result: 571 self.result_info["Data Toggling Failure"] += 1 572 try: 573 self._ad_take_extra_logs(self.dut, test_name, begin_time) 574 self._ad_take_bugreport(self.dut, test_name, begin_time) 575 except Exception as e: 576 self.log.exception(e) 577 return False 578 else: 579 self.result_info["Data Toggling Success"] += 1 580 return True 581 582 def _get_result_message(self): 583 msg_list = [ 584 "%s: %s" % (count, self.result_info[count]) 585 for count in sorted(self.result_info.keys()) 586 ] 587 return ", ".join(msg_list) 588 589 def _write_perf_json(self): 590 json_str = json.dumps(self.perf_data, indent=4, sort_keys=True) 591 with open(self.perf_file, 'w') as f: 592 f.write(json_str) 593 594 def _init_perf_json(self): 595 self.perf_file = os.path.join(self.log_path, "%s_perf_data_%s.json" % 596 (self.test_name, self.begin_time)) 597 self.perf_data = self.dut.build_info.copy() 598 self.perf_data["build_fingerprint"] = self.dut.adb.getprop( 599 "ro.build.fingerprint") 600 self.perf_data["model"] = self.dut.model 601 self.perf_data["carrier"] = self.dut.adb.getprop( 602 "gsm.sim.operator.alpha") 603 self._write_perf_json() 604 605 def _update_perf_json(self): 606 for result_key, result_value in self.result_info.items(): 607 self.perf_data[result_key] = result_value 608 self._write_perf_json() 609 610 def crash_check_test(self): 611 failure = 0 612 while time.time() < self.finishing_time: 613 try: 614 self.log.info(dict(self.result_info)) 615 self._update_perf_json() 616 begin_time = get_device_epoch_time(self.dut) 617 run_time_in_seconds = (begin_time - self.begin_time) / 1000 618 test_name = "%s_crash_%s_seconds_after_start" % ( 619 self.test_name, run_time_in_seconds) 620 time.sleep(self.crash_check_interval) 621 for ad in self.android_devices: 622 crash_report = ad.check_crash_report( 623 test_name, begin_time, log_crash_report=True) 624 if crash_report: 625 ad.log.error("Find new crash reports %s", crash_report) 626 failure += 1 627 self.result_info["Crashes"] += len(crash_report) 628 for crash in crash_report: 629 if "ramdump_modem" in crash: 630 self.result_info["Crashes-Modem"] += 1 631 try: 632 ad.take_bug_report(test_name, begin_time) 633 except Exception as e: 634 self.log.exception(e) 635 except Exception as e: 636 self.log.error("Exception error %s", str(e)) 637 self.result_info["Exception Errors"] += 1 638 self.log.info("Crashes found: %s", failure) 639 if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: 640 self.log.error("Too many exception errors, quit test") 641 return False 642 if failure: 643 return False 644 else: 645 return True 646 647 def call_test(self, call_verification_func=None): 648 while time.time() < self.finishing_time: 649 time.sleep( 650 random.randrange(self.min_sleep_time, self.max_sleep_time)) 651 try: 652 self._make_phone_call(call_verification_func) 653 except Exception as e: 654 self.log.exception("Exception error %s", str(e)) 655 self.result_info["Exception Errors"] += 1 656 if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: 657 self.log.error("Too many exception errors, quit test") 658 return False 659 self.log.info("%s", dict(self.result_info)) 660 if any([ 661 self.result_info["Call Setup Failure"], 662 self.result_info["Call Maintenance Failure"], 663 self.result_info["Call Teardown Failure"] 664 ]): 665 return False 666 else: 667 return True 668 669 def message_test(self, max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE): 670 while time.time() < self.finishing_time: 671 try: 672 self._send_message(max_wait_time=max_wait_time) 673 except Exception as e: 674 self.log.exception("Exception error %s", str(e)) 675 self.result_info["Exception Errors"] += 1 676 self.log.info(dict(self.result_info)) 677 if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: 678 self.log.error("Too many exception errors, quit test") 679 return False 680 time.sleep( 681 random.randrange(self.min_sleep_time, self.max_sleep_time)) 682 if self.result_info["SMS Failure"] or ( 683 self.result_info["MMS Failure"] / self.result_info["MMS Total"] 684 > 0.3): 685 return False 686 else: 687 return True 688 689 def _data_download(self, file_names=["5MB", "10MB", "20MB", "50MB"]): 690 begin_time = get_current_epoch_time() 691 slot_id = random.randint(0,1) 692 if self.dsds_esim: 693 sub_id = get_subid_from_slot_index(self.log, self.dut, slot_id) 694 self.dut.log.info("Data - slot_Id %d", slot_id) 695 set_subid_for_data(self.dut, sub_id) 696 self.dut.droid.telephonyToggleDataConnection(True) 697 start_qxdm_loggers(self.log, self.android_devices) 698 self.dut.log.info(dict(self.result_info)) 699 selection = random.randrange(0, len(file_names)) 700 file_name = file_names[selection] 701 self.result_info["Internet Connection Check Total"] += 1 702 if not self.internet_connection_check_method(self.log, self.dut): 703 rat = self.dut.adb.getprop("gsm.network.type") 704 if "," in rat: 705 if self.dsds_esim: 706 rat = rat.split(',')[slot_id] 707 else: 708 rat = rat.split(',')[0] 709 self.dut.log.info("Network in RAT %s", rat) 710 if self.dut_incall and not is_rat_svd_capable(rat.upper()): 711 self.result_info[ 712 "Expected Incall Internet Connection Check Failure"] += 1 713 return True 714 else: 715 self.result_info["Internet Connection Check Failure"] += 1 716 test_name = "%s_internet_connection_No_%s_failure" % ( 717 self.test_name, 718 self.result_info["Internet Connection Check Failure"]) 719 try: 720 self._ad_take_extra_logs(self.dut, test_name, begin_time) 721 self._ad_take_bugreport(self.dut, test_name, begin_time) 722 except Exception as e: 723 self.log.exception(e) 724 return False 725 else: 726 self.result_info["Internet Connection Check Success"] += 1 727 728 self.result_info["File Download Total"] += 1 729 if not active_file_download_test( 730 self.log, self.dut, file_name, 731 method=self.file_download_method): 732 self.result_info["File Download Failure"] += 1 733 if self.result_info["File Download Failure"] == 1: 734 try: 735 self._ad_take_extra_logs( 736 self.dut, "%s_file_download_failure" % self.test_name, 737 begin_time) 738 self._ad_take_bugreport( 739 self.dut, "%s_file_download_failure" % self.test_name, 740 begin_time) 741 except Exception as e: 742 self.log.exception(e) 743 return False 744 else: 745 self.result_info["File Download Success"] += 1 746 return True 747 748 def data_test(self): 749 while time.time() < self.finishing_time: 750 try: 751 self._data_download() 752 except Exception as e: 753 self.log.error("Exception error %s", str(e)) 754 self.result_info["Exception Errors"] += 1 755 self.log.info("%s", dict(self.result_info)) 756 if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: 757 self.log.error("Too many exception errors, quit test") 758 return False 759 time.sleep( 760 random.randrange(self.min_sleep_time, self.max_sleep_time)) 761 if self.result_info["Internet Connection Check Failure"]: 762 return False 763 else: 764 return True 765 766 def _check_data(self): 767 self.result_info["Data Connection Check Total"] += 1 768 if not wait_for_data_connection(self.log, self.dut, True): 769 self.result_info["Data Connection Setup Failure"] += 1 770 return False 771 if not self.internet_connection_check_method(self.log, self.dut): 772 rat = self.dut.adb.getprop("gsm.network.type") 773 self.dut.log.info("Network in RAT %s", rat) 774 self.result_info["Internet Connection Check Failure"] += 1 775 return False 776 return True 777 778 def _data_call_test(self, sub_id, generation): 779 self.dut.log.info(dict(self.result_info)) 780 begin_time = get_device_epoch_time(self.dut) 781 start_qxdm_loggers(self.log, self.android_devices) 782 self.result_info["Network Change Request Total"] += 1 783 test_name = "%s_network_change_test_iter_%s" % ( 784 self.test_name, self.result_info["Network Change Request Total"]) 785 log_msg = "[Test Case] %s" % test_name 786 self.log.info("%s begin", log_msg) 787 self.dut.droid.logI("[BEGIN]%s" % log_msg) 788 if not ensure_network_generation_for_subscription( 789 self.log, self.dut, sub_id, 790 generation) or not self._check_data(): 791 self.result_info["Network Change Failure"] += 1 792 self.dut.droid.logI("%s end" % log_msg) 793 self.dut.log.info("[END]%s", log_msg) 794 try: 795 self._ad_take_extra_logs(self.dut, test_name, begin_time) 796 self._ad_take_bugreport(self.dut, test_name, begin_time) 797 except Exception as e: 798 self.log.warning(e) 799 return False 800 if not self._mobile_data_toggling(setup=None): 801 return False 802 return True 803 804 def data_call_stress_test(self): 805 result = True 806 sub_id = self.dut.droid.subscriptionGetDefaultSubId() 807 while time.time() < self.finishing_time: 808 for generation in (GEN_4G, GEN_3G): 809 try: 810 if not self._data_call_test(sub_id, generation): 811 result = False 812 except Exception as e: 813 self.log.error("Exception error %s", str(e)) 814 self.result_info["Exception Errors"] += 1 815 if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: 816 self.log.error("Too many exception errors, quit test") 817 return False 818 return result 819 820 def check_incall_data(self): 821 if verify_internet_connection_by_ping(self.log, self.dut): 822 self.internet_connection_check_method = verify_internet_connection_by_ping 823 elif verify_http_connection(self.log, self.dut): 824 self.internet_connection_check_method = verify_http_connection 825 else: 826 self.dut.log.error("Data test failed") 827 raise signals.TestFailure("Data check failed") 828 if self.single_phone_test: 829 if not initiate_call( 830 self.log, self.dut, 831 self.call_server_number) and wait_for_in_call_active( 832 self.dut, 60, 3): 833 self._take_bug_report(self.test_name, self.begin_time) 834 raise signals.TestFailure("Unable to make phone call") 835 else: 836 if not call_setup_teardown( 837 self.log, self.dut, self.android_devices[1], 838 ad_hangup=None): 839 self._take_bug_report(self.test_name, self.begin_time) 840 raise signals.TestFailure("Unable to make phone call") 841 voice_rat = self.dut.droid.telephonyGetCurrentVoiceNetworkType() 842 data_rat = self.dut.droid.telephonyGetCurrentDataNetworkType() 843 self.dut.log.info("Voice in RAT %s, Data in RAT %s", voice_rat, 844 data_rat) 845 try: 846 if "wfc" in self.test_name or is_rat_svd_capable( 847 voice_rat.upper()) and is_rat_svd_capable( 848 data_rat.upper()): 849 self.dut.log.info("Capable for simultaneous voice and data") 850 851 if not self.internet_connection_check_method( 852 self.log, self.dut): 853 self.dut.log.error("Incall data check failed") 854 raise signals.TestFailure("Incall data check failed") 855 else: 856 return True 857 else: 858 self.dut.log.info( 859 "Not capable for simultaneous voice and data") 860 return False 861 hangup_call(self.log, self.dut) 862 finally: 863 for ad in self.android_devices: 864 if ad.droid.telecomIsInCall(): 865 hangup_call(self.log, ad) 866 867 def parallel_tests(self, setup_func=None, call_verification_func=None): 868 self.log.info(self._get_result_message()) 869 if setup_func and not setup_func(): 870 msg = "%s setup %s failed" % (self.test_name, setup_func.__name__) 871 self.log.error(msg) 872 self._take_bug_report("%s%s" % (self.test_name, 873 setup_func.__name__), 874 self.begin_time) 875 return False 876 if not call_verification_func: 877 call_verification_func = is_phone_in_call 878 self.finishing_time = time.time() + self.max_run_time 879 if not self.dsds_esim and self.check_incall_data(): 880 self.log.info( 881 "==== Start parallel voice/message/data stress test ====") 882 self.perf_data["testing method"] = "parallel" 883 results = run_multithread_func( 884 self.log, [(self.call_test, [call_verification_func]), 885 (self.message_test, []), (self.data_test, []), 886 (self.crash_check_test, [])]) 887 else: 888 self.log.info( 889 "==== Start sequential voice/message/data stress test ====") 890 self.perf_data["testing method"] = "sequential" 891 results = run_multithread_func( 892 self.log, [(self.sequential_tests, [call_verification_func]), 893 (self.crash_check_test, [])]) 894 result_message = self._get_result_message() 895 self.log.info(result_message) 896 self._update_perf_json() 897 self.result_detail = result_message 898 return all(results) 899 900 def sequential_tests(self, call_verification_func): 901 funcs = [(self._make_phone_call, [call_verification_func]), 902 (self._send_message, []), (self._data_download, [["5MB"]])] 903 while time.time() < self.finishing_time: 904 selection = random.randrange(0, 3) 905 try: 906 funcs[selection][0](*funcs[selection][1]) 907 except Exception as e: 908 self.log.error("Exception error %s", str(e)) 909 self.result_info["Exception Errors"] += 1 910 self.log.info("%s", dict(self.result_info)) 911 if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: 912 self.log.error("Too many exception errors, quit test") 913 return False 914 time.sleep( 915 random.randrange(self.min_sleep_time, self.max_sleep_time)) 916 if any([ 917 self.result_info["Call Setup Failure"], 918 self.result_info["Call Maintenance Failure"], 919 self.result_info["Call Teardown Failure"], 920 self.result_info["SMS Failure"], 921 self.result_info["MMS Failure"], 922 self.result_info["Internet Connection Check Failure"] 923 ]): 924 return False 925 return True 926 927 def volte_modechange_volte_test(self): 928 sub_id = self.dut.droid.subscriptionGetDefaultSubId() 929 result = True 930 while time.time() < self.finishing_time: 931 try: 932 if self._prefnetwork_mode_change(sub_id): 933 run_multithread_func( 934 self.log, 935 [(self._data_download, [["5MB"]]), 936 (self._make_phone_call, [is_phone_in_call_volte]), 937 (self._send_message, [])]) 938 else: 939 result = False 940 if self._mobile_data_toggling(): 941 run_multithread_func( 942 self.log, 943 [(self._data_download, [["5MB"]]), 944 (self._make_phone_call, [is_phone_in_call_volte]), 945 (self._send_message, [])]) 946 else: 947 result = False 948 except Exception as e: 949 self.log.error("Exception error %s", str(e)) 950 self.result_info["Exception Errors"] += 1 951 self.log.info(dict(self.result_info)) 952 if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: 953 self.log.error("Too many exception errors, quit test") 954 return False 955 return result 956 957 def parallel_with_network_change_tests(self, setup_func=None): 958 if setup_func and not setup_func(): 959 self.log.error("Test setup %s failed", setup_func.__name__) 960 return False 961 self.finishing_time = time.time() + self.max_run_time 962 results = run_multithread_func(self.log, 963 [(self.volte_modechange_volte_test, []), 964 (self.crash_check_test, [])]) 965 result_message = self._get_result_message() 966 self.log.info(result_message) 967 self._update_perf_json() 968 self.result_detail = result_message 969 return all(results) 970 971 """ Tests Begin """ 972 973 @test_tracker_info(uuid="d035e5b9-476a-4e3d-b4e9-6fd86c51a68d") 974 @TelephonyBaseTest.tel_test_wrap 975 def test_default_parallel_stress(self): 976 """ Default state stress test""" 977 return self.parallel_tests() 978 979 @test_tracker_info(uuid="c21e1f17-3282-4f0b-b527-19f048798098") 980 @TelephonyBaseTest.tel_test_wrap 981 def test_lte_volte_parallel_stress(self): 982 """ VoLTE on stress test""" 983 if CAPABILITY_VOLTE not in self.dut_capabilities: 984 raise signals.TestSkipClass("VoLTE is not supported") 985 return self.parallel_tests( 986 setup_func=self._setup_lte_volte_enabled, 987 call_verification_func=is_phone_in_call_volte) 988 989 @test_tracker_info(uuid="a317c23a-41e0-4ef8-af67-661451cfefcf") 990 @TelephonyBaseTest.tel_test_wrap 991 def test_csfb_parallel_stress(self): 992 """ LTE non-VoLTE stress test""" 993 return self.parallel_tests( 994 setup_func=self._setup_lte_volte_disabled, 995 call_verification_func=is_phone_in_call_csfb) 996 997 @test_tracker_info(uuid="fdb791bf-c414-4333-9fa3-cc18c9b3b234") 998 @TelephonyBaseTest.tel_test_wrap 999 def test_wfc_parallel_stress(self): 1000 """ Wifi calling APM mode off stress test""" 1001 if CAPABILITY_WFC not in self.dut_capabilities: 1002 raise signals.TestSkipClass("WFC is not supported") 1003 if WFC_MODE_WIFI_PREFERRED not in self.dut_wfc_modes: 1004 raise signals.TestSkip("WFC_MODE_WIFI_PREFERRED is not supported") 1005 return self.parallel_tests( 1006 setup_func=self._setup_wfc, 1007 call_verification_func=is_phone_in_call_iwlan) 1008 1009 @test_tracker_info(uuid="e334c1b3-4378-49bb-bf57-1573fa1b23fa") 1010 @TelephonyBaseTest.tel_test_wrap 1011 def test_wfc_apm_parallel_stress(self): 1012 """ Wifi calling in APM mode on stress test""" 1013 if CAPABILITY_WFC not in self.dut_capabilities: 1014 raise signals.TestSkipClass("WFC is not supported") 1015 return self.parallel_tests( 1016 setup_func=self._setup_wfc_apm, 1017 call_verification_func=is_phone_in_call_iwlan) 1018 1019 @test_tracker_info(uuid="4566eef6-55de-4ac8-87ee-58f2ef41a3e8") 1020 @TelephonyBaseTest.tel_test_wrap 1021 def test_3g_parallel_stress(self): 1022 """ 3G stress test""" 1023 return self.parallel_tests( 1024 setup_func=self._setup_3g, 1025 call_verification_func=is_phone_in_call_3g) 1026 1027 @test_tracker_info(uuid="f34f1a31-3948-4675-8698-372a83b8088d") 1028 @TelephonyBaseTest.tel_test_wrap 1029 def test_2g_parallel_stress(self): 1030 """ 2G call stress test""" 1031 return self.parallel_tests( 1032 setup_func=self._setup_2g, 1033 call_verification_func=is_phone_in_call_2g) 1034 1035 @test_tracker_info(uuid="af580fca-fea6-4ca5-b981-b8c710302d37") 1036 @TelephonyBaseTest.tel_test_wrap 1037 def test_volte_modeprefchange_parallel_stress(self): 1038 """ VoLTE Mode Pref call stress test""" 1039 if CAPABILITY_VOLTE not in self.dut_capabilities: 1040 raise signals.TestSkipClass("VoLTE is not supported") 1041 return self.parallel_with_network_change_tests( 1042 setup_func=self._setup_lte_volte_enabled) 1043 1044 @test_tracker_info(uuid="10e34247-5fd3-4f87-81bf-3c17a6b71ab2") 1045 @TelephonyBaseTest.tel_test_wrap 1046 def test_data_call_stress(self): 1047 """ Default state stress test""" 1048 self.finishing_time = time.time() + self.max_run_time 1049 results = run_multithread_func(self.log, 1050 [(self.data_call_stress_test, []), 1051 (self.crash_check_test, [])]) 1052 result_message = self._get_result_message() 1053 self.log.info(result_message) 1054 self._update_perf_json() 1055 self.result_detail = result_message 1056 return all(results) 1057 1058 """ Tests End """ 1059