1#!/usr/bin/env python3 2# 3# Copyright 2016 - 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 Base Class for Defining Common Telephony Test Functionality 18""" 19 20import logging 21import os 22import re 23import shutil 24import time 25 26from acts import asserts 27from acts import records 28from acts import signals 29from acts import utils 30from acts.base_test import BaseTestClass 31from acts.controllers.adb_lib.error import AdbCommandError 32from acts.controllers.android_device import DEFAULT_QXDM_LOG_PATH 33from acts.keys import Config 34from acts.libs.utils.multithread import multithread_func 35from acts.libs.utils.multithread import run_multithread_func 36from acts_contrib.test_utils.tel.tel_bootloader_utils import flash_radio 37from acts_contrib.test_utils.tel.tel_defines import CHIPSET_MODELS_LIST 38from acts_contrib.test_utils.tel.tel_defines import INVALID_SUB_ID 39from acts_contrib.test_utils.tel.tel_defines import MULTI_SIM_CONFIG, SINGLE_SIM_CONFIG 40from acts_contrib.test_utils.tel.tel_defines import PRECISE_CALL_STATE_LISTEN_LEVEL_BACKGROUND 41from acts_contrib.test_utils.tel.tel_defines import PRECISE_CALL_STATE_LISTEN_LEVEL_FOREGROUND 42from acts_contrib.test_utils.tel.tel_defines import PRECISE_CALL_STATE_LISTEN_LEVEL_RINGING 43from acts_contrib.test_utils.tel.tel_defines import SIM_STATE_ABSENT 44from acts_contrib.test_utils.tel.tel_defines import SIM_STATE_UNKNOWN 45from acts_contrib.test_utils.tel.tel_defines import WIFI_VERBOSE_LOGGING_DISABLED 46from acts_contrib.test_utils.tel.tel_ims_utils import activate_wfc_on_device 47from acts_contrib.test_utils.tel.tel_logging_utils import disable_qxdm_logger 48from acts_contrib.test_utils.tel.tel_logging_utils import get_screen_shot_log 49from acts_contrib.test_utils.tel.tel_logging_utils import get_tcpdump_log 50from acts_contrib.test_utils.tel.tel_logging_utils import set_qxdm_logger_command 51from acts_contrib.test_utils.tel.tel_logging_utils import start_dsp_logger 52from acts_contrib.test_utils.tel.tel_logging_utils import start_qxdm_logger 53from acts_contrib.test_utils.tel.tel_logging_utils import start_qxdm_loggers 54from acts_contrib.test_utils.tel.tel_logging_utils import start_sdm_logger 55from acts_contrib.test_utils.tel.tel_logging_utils import start_sdm_loggers 56from acts_contrib.test_utils.tel.tel_logging_utils import start_tcpdumps 57from acts_contrib.test_utils.tel.tel_logging_utils import stop_qxdm_logger 58from acts_contrib.test_utils.tel.tel_logging_utils import stop_sdm_logger 59from acts_contrib.test_utils.tel.tel_logging_utils import stop_tcpdumps 60from acts_contrib.test_utils.tel.tel_phone_setup_utils import ensure_phone_default_state 61from acts_contrib.test_utils.tel.tel_phone_setup_utils import ensure_phones_default_state 62from acts_contrib.test_utils.tel.tel_phone_setup_utils import ensure_phone_idle 63from acts_contrib.test_utils.tel.tel_subscription_utils import get_subid_from_slot_index 64from acts_contrib.test_utils.tel.tel_subscription_utils import initial_set_up_for_subid_information 65from acts_contrib.test_utils.tel.tel_subscription_utils import set_default_sub_for_all_services 66from acts_contrib.test_utils.tel.tel_test_utils import activate_esim_using_suw 67from acts_contrib.test_utils.tel.tel_test_utils import activate_google_fi_account 68from acts_contrib.test_utils.tel.tel_test_utils import adb_disable_verity 69from acts_contrib.test_utils.tel.tel_test_utils import add_google_account 70from acts_contrib.test_utils.tel.tel_test_utils import build_id_override 71from acts_contrib.test_utils.tel.tel_test_utils import check_google_fi_activated 72from acts_contrib.test_utils.tel.tel_test_utils import enable_connectivity_metrics 73from acts_contrib.test_utils.tel.tel_test_utils import enable_radio_log_on 74from acts_contrib.test_utils.tel.tel_test_utils import force_connectivity_metrics_upload 75from acts_contrib.test_utils.tel.tel_test_utils import get_sim_state 76from acts_contrib.test_utils.tel.tel_test_utils import install_apk 77from acts_contrib.test_utils.tel.tel_test_utils import install_googleaccountutil_apk 78from acts_contrib.test_utils.tel.tel_test_utils import install_googlefi_apk 79from acts_contrib.test_utils.tel.tel_test_utils import phone_switch_to_msim_mode 80from acts_contrib.test_utils.tel.tel_test_utils import print_radio_info 81from acts_contrib.test_utils.tel.tel_test_utils import reboot_device 82from acts_contrib.test_utils.tel.tel_test_utils import recover_build_id 83from acts_contrib.test_utils.tel.tel_test_utils import set_phone_screen_on 84from acts_contrib.test_utils.tel.tel_test_utils import set_phone_silent_mode 85from acts_contrib.test_utils.tel.tel_test_utils import setup_droid_properties 86from acts_contrib.test_utils.tel.tel_test_utils import synchronize_device_time 87from acts_contrib.test_utils.tel.tel_test_utils import toggle_airplane_mode 88from acts_contrib.test_utils.tel.tel_test_utils import unlock_sim 89from acts_contrib.test_utils.tel.tel_test_utils import wait_for_sim_ready_by_adb 90from acts_contrib.test_utils.tel.tel_test_utils import wait_for_sims_ready_by_adb 91from acts_contrib.test_utils.tel.tel_wifi_utils import ensure_wifi_connected 92 93 94REMOUNT_REBOOT_MSG = "Now reboot your device for settings to take effect" 95 96 97class TelephonyBaseTest(BaseTestClass): 98 # Use for logging in the test cases to facilitate 99 # faster log lookup and reduce ambiguity in logging. 100 @staticmethod 101 def tel_test_wrap(fn): 102 def _safe_wrap_test_case(self, *args, **kwargs): 103 test_id = "%s:%s:%s" % (self.__class__.__name__, self.test_name, 104 self.log_begin_time.replace(' ', '-')) 105 self.test_id = test_id 106 self.result_detail = "" 107 self.testsignal_details = "" 108 self.testsignal_extras = {} 109 tries = int(self.user_params.get("telephony_auto_rerun", 1)) 110 for ad in self.android_devices: 111 ad.log_path = self.log_path 112 for i in range(tries + 1): 113 result = True 114 if i > 0: 115 log_string = "[Test Case] RERUN %s" % self.test_name 116 self.log.info(log_string) 117 self._teardown_test(self.test_name) 118 self._setup_test(self.test_name) 119 try: 120 result = fn(self, *args, **kwargs) 121 except signals.TestFailure as e: 122 self.testsignal_details = e.details 123 self.testsignal_extras = e.extras 124 result = False 125 except signals.TestSignal: 126 raise 127 except Exception as e: 128 self.log.exception(e) 129 asserts.fail(self.result_detail) 130 if result is False: 131 if i < tries: 132 continue 133 else: 134 break 135 if self.user_params.get("check_crash", True): 136 new_crash = ad.check_crash_report(self.test_name, 137 self.begin_time, True) 138 if new_crash: 139 msg = "Find new crash reports %s" % new_crash 140 ad.log.error(msg) 141 self.result_detail = "%s %s %s" % (self.result_detail, 142 ad.serial, msg) 143 result = False 144 if result is not False: 145 asserts.explicit_pass(self.result_detail) 146 else: 147 if self.result_detail: 148 asserts.fail(self.result_detail) 149 else: 150 asserts.fail(self.testsignal_details, self.testsignal_extras) 151 152 return _safe_wrap_test_case 153 154 def setup_class(self): 155 super().setup_class() 156 self.wifi_network_ssid = self.user_params.get( 157 "wifi_network_ssid") or self.user_params.get( 158 "wifi_network_ssid_2g") or self.user_params.get( 159 "wifi_network_ssid_5g") 160 self.wifi_network_pass = self.user_params.get( 161 "wifi_network_pass") or self.user_params.get( 162 "wifi_network_pass_2g") or self.user_params.get( 163 "wifi_network_ssid_5g") 164 165 self.log_path = getattr(logging, "log_path", None) 166 self.qxdm_log = self.user_params.get("qxdm_log", True) 167 self.sdm_log = self.user_params.get("sdm_log", False) 168 self.tcpdump_log = self.user_params.get("tcpdump_log", True) 169 self.dsp_log = self.user_params.get("dsp_log", False) 170 self.dsp_log_p21 = self.user_params.get("dsp_log_p21", False) 171 self.enable_radio_log_on = self.user_params.get( 172 "enable_radio_log_on", False) 173 self.cbrs_esim = self.user_params.get("cbrs_esim", False) 174 self.account_util = self.user_params.get("account_util", None) 175 self.save_passing_logs = self.user_params.get("save_passing_logs", False) 176 if isinstance(self.account_util, list): 177 self.account_util = self.account_util[0] 178 self.fi_util = self.user_params.get("fi_util", None) 179 if isinstance(self.fi_util, list): 180 self.fi_util = self.fi_util[0] 181 self.radio_img = self.user_params.get("radio_img", None) 182 if isinstance(self.radio_img, list): 183 self.radio_img = self.radio_img[0] 184 self.modem_bin = self.user_params.get("modem_bin", None) 185 if isinstance(self.modem_bin, list): 186 self.modem_bin = self.modem_bin[0] 187 self.extra_apk = self.user_params.get("extra_apk", None) 188 if isinstance(self.extra_apk, list): 189 self.extra_apk = self.extra_apk[0] 190 self.extra_package = self.user_params.get("extra_package", None) 191 192 if self.radio_img or self.modem_bin: 193 sideload_img = True 194 if self.radio_img: 195 file_path = self.radio_img 196 elif self.modem_bin: 197 file_path = self.modem_bin 198 sideload_img = False 199 tasks = [(flash_radio, [ad, file_path, True, sideload_img]) 200 for ad in self.android_devices] 201 multithread_func(self.log, tasks) 202 if self.extra_apk and self.extra_package: 203 tasks = [(install_apk, [ad, self.extra_apk, self.extra_package]) 204 for ad in self.android_devices] 205 multithread_func(self.log, tasks) 206 207 tasks = [(self._init_device, [ad]) for ad in self.android_devices] 208 multithread_func(self.log, tasks) 209 self.reboot_before_test = self.user_params.get( 210 "reboot_before_test", False) 211 self.skip_reset_between_cases = self.user_params.get( 212 "skip_reset_between_cases", True) 213 self.log_path = getattr(logging, "log_path", None) 214 self.sim_config = { 215 "config":SINGLE_SIM_CONFIG, 216 "number_of_sims":1 217 } 218 219 for ad in self.android_devices: 220 if getattr(ad, 'dsds', False): 221 self.sim_config = { 222 "config":MULTI_SIM_CONFIG, 223 "number_of_sims":2 224 } 225 break 226 if "anritsu_md8475a_ip_address" in self.user_params: 227 return 228 qxdm_log_mask_cfg = self.user_params.get("qxdm_log_mask_cfg", None) 229 if isinstance(qxdm_log_mask_cfg, list): 230 qxdm_log_mask_cfg = qxdm_log_mask_cfg[0] 231 if qxdm_log_mask_cfg and "dev/null" in qxdm_log_mask_cfg: 232 qxdm_log_mask_cfg = None 233 sim_conf_file = self.user_params.get("sim_conf_file") 234 if not sim_conf_file: 235 self.log.info("\"sim_conf_file\" is not provided test bed config!") 236 else: 237 if isinstance(sim_conf_file, list): 238 sim_conf_file = sim_conf_file[0] 239 # If the sim_conf_file is not a full path, attempt to find it 240 # relative to the config file. 241 if not os.path.isfile(sim_conf_file): 242 sim_conf_file = os.path.join( 243 self.user_params[Config.key_config_path.value], 244 sim_conf_file) 245 if not os.path.isfile(sim_conf_file): 246 self.log.error("Unable to load user config %s ", 247 sim_conf_file) 248 249 tasks = [(self._setup_device, [ad, sim_conf_file, qxdm_log_mask_cfg]) 250 for ad in self.android_devices] 251 return multithread_func(self.log, tasks) 252 253 def _init_device(self, ad): 254 synchronize_device_time(ad) 255 ad.log_path = self.log_path 256 print_radio_info(ad) 257 unlock_sim(ad) 258 ad.wakeup_screen() 259 ad.adb.shell("input keyevent 82") 260 261 def wait_for_sim_ready(self,ad): 262 wait_for_sim_ready_on_sim_config = { 263 SINGLE_SIM_CONFIG : lambda:wait_for_sim_ready_by_adb(self.log,ad), 264 MULTI_SIM_CONFIG : lambda:wait_for_sims_ready_by_adb(self.log,ad) 265 } 266 if not wait_for_sim_ready_on_sim_config[self.sim_config["config"]]: 267 raise signals.TestAbortClass("unable to load the SIM") 268 269 def _setup_device(self, ad, sim_conf_file, qxdm_log_mask_cfg=None): 270 ad.qxdm_log = getattr(ad, "qxdm_log", self.qxdm_log) 271 ad.sdm_log = getattr(ad, "sdm_log", self.sdm_log) 272 ad.dsp_log = getattr(ad, "dsp_log", self.dsp_log) 273 ad.dsp_log_p21 = getattr(ad, "dsp_log_p21", self.dsp_log_p21) 274 if self.user_params.get("enable_connectivity_metrics", False): 275 enable_connectivity_metrics(ad) 276 if self.user_params.get("build_id_override", False): 277 build_postfix = self.user_params.get("build_id_postfix", 278 "LAB_TEST") 279 build_id_override( 280 ad, 281 new_build_id=self.user_params.get("build_id_override_with", 282 None), 283 postfix=build_postfix) 284 285 if self.enable_radio_log_on: 286 enable_radio_log_on(ad) 287 list_of_models = CHIPSET_MODELS_LIST 288 if any(model in ad.model for model in list_of_models): 289 phone_mode = "ssss" 290 if hasattr(ad, "mtp_dsds"): 291 phone_mode = "dsds" 292 if ad.adb.getprop("persist.radio.multisim.config") != phone_mode: 293 ad.adb.shell("setprop persist.radio.multisim.config %s" \ 294 % phone_mode) 295 reboot_device(ad) 296 297 if "_test" not in ad.build_info["build_id"]: 298 ad.ensure_verity_disabled() 299 try: 300 ad.adb.remount() 301 except AdbCommandError as e: 302 if REMOUNT_REBOOT_MSG in e.stderr: 303 ad.reboot() 304 ad.adb.remount() 305 build_id = ad.build_info["build_id"].replace(".", r"\.") 306 ad.adb.shell("sed -i '/^ro.build.id=/ " 307 f"s/{build_id}/&_test/g' /system/build.prop") 308 ad.adb.shell("sed -i '/^ro.build.description=/ " 309 f"s/{build_id}/&_test/g' /system/build.prop") 310 311 if ad.dsp_log: 312 start_dsp_logger(ad) 313 elif ad.dsp_log_p21: 314 start_dsp_logger(ad, p21=True) 315 else: 316 ad.reboot() 317 stop_qxdm_logger(ad) 318 if ad.qxdm_log: 319 qxdm_log_mask = getattr(ad, "qxdm_log_mask", None) 320 if qxdm_log_mask_cfg: 321 qxdm_mask_path = self.user_params.get("qxdm_log_path", 322 DEFAULT_QXDM_LOG_PATH) 323 ad.adb.shell("mkdir %s" % qxdm_mask_path, ignore_status=True) 324 ad.log.info("Push %s to %s", qxdm_log_mask_cfg, qxdm_mask_path) 325 ad.adb.push("%s %s" % (qxdm_log_mask_cfg, qxdm_mask_path)) 326 mask_file_name = os.path.split(qxdm_log_mask_cfg)[-1] 327 qxdm_log_mask = os.path.join(qxdm_mask_path, mask_file_name) 328 set_qxdm_logger_command(ad, mask=qxdm_log_mask) 329 start_qxdm_logger(ad, utils.get_current_epoch_time()) 330 elif ad.sdm_log: 331 start_sdm_logger(ad) 332 else: 333 disable_qxdm_logger(ad) 334 if not unlock_sim(ad): 335 raise signals.TestAbortClass("unable to unlock the SIM") 336 337 # If device is setup already, skip the following setup procedures 338 if getattr(ad, "telephony_test_setup", None): 339 return True 340 341 # eSIM enablement 342 if hasattr(ad, "fi_esim"): 343 if not ensure_wifi_connected(self.log, ad, self.wifi_network_ssid, 344 self.wifi_network_pass): 345 ad.log.error("Failed to connect to wifi") 346 if check_google_fi_activated(ad): 347 ad.log.info("Google Fi is already Activated") 348 else: 349 install_googleaccountutil_apk(ad, self.account_util) 350 add_google_account(ad) 351 install_googlefi_apk(ad, self.fi_util) 352 if not activate_google_fi_account(ad): 353 ad.log.error("Failed to activate Fi") 354 check_google_fi_activated(ad) 355 if getattr(ad, 'dsds', False): 356 sim_mode = ad.droid.telephonyGetPhoneCount() 357 if sim_mode == 1: 358 ad.log.info("Phone in Single SIM Mode") 359 if not phone_switch_to_msim_mode(ad): 360 ad.log.error("Failed to switch to Dual SIM Mode") 361 return False 362 elif sim_mode == 2: 363 ad.log.info("Phone already in Dual SIM Mode") 364 if get_sim_state(ad) in (SIM_STATE_ABSENT, SIM_STATE_UNKNOWN): 365 ad.log.info("Device has no or unknown SIM in it") 366 # eSIM needs activation 367 activate_esim_using_suw(ad) 368 ensure_phone_idle(self.log, ad) 369 if getattr(ad, 'mep', False): 370 setup_droid_properties(self.log, ad, sim_conf_file, True) 371 else: 372 setup_droid_properties(self.log, ad, sim_conf_file) 373 elif self.user_params.get("Attenuator"): 374 ad.log.info("Device in chamber room") 375 ensure_phone_idle(self.log, ad) 376 if getattr(ad, 'mep', False): 377 setup_droid_properties(self.log, ad, sim_conf_file, True) 378 else: 379 setup_droid_properties(self.log, ad, sim_conf_file) 380 else: 381 self.wait_for_sim_ready(ad) 382 ensure_phone_default_state(self.log, ad) 383 if getattr(ad, 'mep', False): 384 setup_droid_properties(self.log, ad, sim_conf_file, True) 385 else: 386 setup_droid_properties(self.log, ad, sim_conf_file) 387 388 if getattr(ad, 'mep', False): 389 default_slot = getattr(ad, "default_slot", 1) 390 if get_subid_from_slot_index(ad.log, ad, default_slot) != INVALID_SUB_ID: 391 ad.log.info("Slot %s is the default slot.", default_slot) 392 set_default_sub_for_all_services(ad, default_slot) 393 else: 394 ad.log.warning("Slot %s is NOT a valid slot. Slot %s will be used by default.", 395 default_slot, 1-default_slot) 396 set_default_sub_for_all_services(ad, 1-default_slot) 397 setattr(ad, "default_slot", 1-default_slot) 398 elif getattr(ad, 'dsds', False): 399 default_slot = getattr(ad, "default_slot", 0) 400 if get_subid_from_slot_index(ad.log, ad, default_slot) != INVALID_SUB_ID: 401 ad.log.info("Slot %s is the default slot.", default_slot) 402 set_default_sub_for_all_services(ad, default_slot) 403 else: 404 ad.log.warning("Slot %s is NOT a valid slot. Slot %s will be used by default.", 405 default_slot, 1-default_slot) 406 set_default_sub_for_all_services(ad, 1-default_slot) 407 setattr(ad, "default_slot", 1-default_slot) 408 409 # Activate WFC on Verizon, AT&T and Canada operators as per # b/33187374 & 410 # b/122327716 411 activate_wfc_on_device(self.log, ad) 412 413 # Sub ID setup 414 initial_set_up_for_subid_information(self.log, ad) 415 416 417 #try: 418 # ad.droid.wifiEnableVerboseLogging(WIFI_VERBOSE_LOGGING_ENABLED) 419 #except Exception: 420 # pass 421 422 # Disable Emergency alerts 423 # Set chrome browser start with no-first-run verification and 424 # disable-fre. Give permission to read from and write to storage. 425 for cmd in ("pm disable com.android.cellbroadcastreceiver", 426 "pm grant com.android.chrome " 427 "android.permission.READ_EXTERNAL_STORAGE", 428 "pm grant com.android.chrome " 429 "android.permission.WRITE_EXTERNAL_STORAGE", 430 "rm /data/local/chrome-command-line", 431 "am set-debug-app --persistent com.android.chrome", 432 'echo "chrome --no-default-browser-check --no-first-run ' 433 '--disable-fre" > /data/local/tmp/chrome-command-line'): 434 ad.adb.shell(cmd, ignore_status=True) 435 436 # Curl for 2016/7 devices 437 if not getattr(ad, "curl_capable", False): 438 try: 439 out = ad.adb.shell("/data/curl --version") 440 if not out or "not found" in out: 441 if int(ad.adb.getprop("ro.product.first_api_level")) >= 25: 442 tel_data = self.user_params.get("tel_data", "tel_data") 443 if isinstance(tel_data, list): 444 tel_data = tel_data[0] 445 curl_file_path = os.path.join(tel_data, "curl") 446 if not os.path.isfile(curl_file_path): 447 curl_file_path = os.path.join( 448 self.user_params[Config.key_config_path.value], 449 curl_file_path) 450 if os.path.isfile(curl_file_path): 451 ad.log.info("Pushing Curl to /data dir") 452 ad.adb.push("%s /data" % (curl_file_path)) 453 ad.adb.shell( 454 "chmod 777 /data/curl", ignore_status=True) 455 else: 456 setattr(ad, "curl_capable", True) 457 except Exception: 458 ad.log.info("Failed to push curl on this device") 459 460 # Ensure that a test class starts from a consistent state that 461 # improves chances of valid network selection and facilitates 462 # logging. 463 try: 464 if not set_phone_screen_on(self.log, ad): 465 self.log.error("Failed to set phone screen-on time.") 466 return False 467 if not set_phone_silent_mode(self.log, ad): 468 self.log.error("Failed to set phone silent mode.") 469 return False 470 ad.droid.telephonyAdjustPreciseCallStateListenLevel( 471 PRECISE_CALL_STATE_LISTEN_LEVEL_FOREGROUND, True) 472 ad.droid.telephonyAdjustPreciseCallStateListenLevel( 473 PRECISE_CALL_STATE_LISTEN_LEVEL_RINGING, True) 474 ad.droid.telephonyAdjustPreciseCallStateListenLevel( 475 PRECISE_CALL_STATE_LISTEN_LEVEL_BACKGROUND, True) 476 except Exception as e: 477 self.log.error("Failure with %s", e) 478 setattr(ad, "telephony_test_setup", True) 479 return True 480 481 def _teardown_device(self, ad): 482 try: 483 stop_qxdm_logger(ad) 484 stop_sdm_logger(ad) 485 except Exception as e: 486 self.log.error("Failure with %s", e) 487 try: 488 ad.droid.disableDevicePassword() 489 except Exception as e: 490 self.log.error("Failure with %s", e) 491 if self.user_params.get("enable_connectivity_metrics", False): 492 if not ensure_wifi_connected(self.log, ad, self.wifi_network_ssid, 493 self.wifi_network_pass): 494 ad.log.error("Failed to connect to wifi") 495 force_connectivity_metrics_upload(ad) 496 time.sleep(30) 497 try: 498 ad.droid.wifiEnableVerboseLogging(WIFI_VERBOSE_LOGGING_DISABLED) 499 except Exception as e: 500 self.log.error("Failure with %s", e) 501 try: 502 if self.user_params.get("build_id_override", 503 False) and self.user_params.get( 504 "recover_build_id", False): 505 recover_build_id(ad) 506 except Exception as e: 507 self.log.error("Failure with %s", e) 508 509 def teardown_class(self): 510 tasks = [(self._teardown_device, [ad]) for ad in self.android_devices] 511 multithread_func(self.log, tasks) 512 return True 513 514 def setup_test(self): 515 if getattr(self, "qxdm_log", True): 516 if not self.user_params.get("qxdm_log_mask_cfg", None): 517 if "wfc" in self.test_name: 518 for ad in self.android_devices: 519 if not getattr(ad, "qxdm_logger_command", None) or ( 520 "IMS_DS_CNE_LnX_Golden.cfg" not in getattr( 521 ad, "qxdm_logger_command", "")): 522 set_qxdm_logger_command( 523 ad, "IMS_DS_CNE_LnX_Golden.cfg") 524 else: 525 for ad in self.android_devices: 526 if not getattr(ad, "qxdm_logger_command", None) or ( 527 "IMS_DS_CNE_LnX_Golden.cfg" in getattr( 528 ad, "qxdm_logger_command", "")): 529 set_qxdm_logger_command(ad, None) 530 start_qxdm_loggers(self.log, self.android_devices, self.begin_time) 531 if getattr(self, "sdm_log", False): 532 start_sdm_loggers(self.log, self.android_devices) 533 if getattr(self, "tcpdump_log", True): 534 mask = getattr(self, "tcpdump_mask", "all") 535 if "wfc" in self.test_name or "iwlan" in self.test_name: 536 interface = getattr(self, "tcpdump_interface", "wlan0") 537 else: 538 interface = getattr(self, "tcpdump_interface", "any") 539 start_tcpdumps( 540 self.android_devices, 541 begin_time=self.begin_time, 542 interface=interface, 543 mask=mask) 544 else: 545 stop_tcpdumps(self.android_devices) 546 for ad in self.android_devices: 547 if self.reboot_before_test: 548 ad.reboot() 549 if self.skip_reset_between_cases: 550 ensure_phone_idle(self.log, ad) 551 else: 552 ensure_phone_default_state(self.log, ad) 553 for session in ad._sl4a_manager.sessions.values(): 554 ed = session.get_event_dispatcher() 555 ed.clear_all_events() 556 output = ad.adb.logcat("-t 1") 557 match = re.search(r"\d+-\d+\s\d+:\d+:\d+.\d+", output) 558 if match: 559 ad.test_log_begin_time = match.group(0) 560 if self.user_params.get("apm_before_test", None): 561 toggle_airplane_mode(self.log, ad, True, False) 562 toggle_airplane_mode(self.log, ad, False, False) 563 564 def teardown_test(self): 565 stop_tcpdumps(self.android_devices) 566 567 def on_fail(self, test_name, begin_time): 568 for ad in self.android_devices: 569 # open Phone information page 570 ad.adb.shell("am start -n com.android.phone/.settings.RadioInfo") 571 time.sleep(3) 572 ad.screenshot(f"{ad.serial}_last_screen") 573 self._take_bug_report(test_name, begin_time) 574 ensure_phones_default_state(self.log, self.android_devices) 575 576 def on_pass(self, test_name, begin_time): 577 if self.save_passing_logs: 578 self._take_bug_report(test_name, begin_time) 579 580 def _ad_take_extra_logs(self, ad, test_name, begin_time): 581 ad.adb.wait_for_device() 582 result = True 583 584 try: 585 # get tcpdump and screen shot log 586 get_tcpdump_log(ad, test_name, begin_time) 587 get_screen_shot_log(ad, test_name, begin_time) 588 except Exception as e: 589 ad.log.error("Exception error %s", e) 590 result = False 591 592 try: 593 ad.check_crash_report(test_name, begin_time, log_crash_report=True) 594 except Exception as e: 595 ad.log.error("Failed to check crash report for %s with error %s", 596 test_name, e) 597 result = False 598 599 extra_qxdm_logs_in_seconds = self.user_params.get( 600 "extra_qxdm_logs_in_seconds", 60 * 3) 601 if getattr(ad, "qxdm_log", True): 602 # Gather qxdm log modified 3 minutes earlier than test start time 603 if begin_time: 604 qxdm_begin_time = begin_time - 1000 * extra_qxdm_logs_in_seconds 605 else: 606 qxdm_begin_time = None 607 try: 608 time.sleep(10) 609 ad.get_qxdm_logs(test_name, qxdm_begin_time) 610 except Exception as e: 611 ad.log.error("Failed to get QXDM log for %s with error %s", 612 test_name, e) 613 result = False 614 if getattr(ad, "sdm_log", False): 615 # Gather sdm log modified 3 minutes earlier than test start time 616 if begin_time: 617 sdm_begin_time = begin_time - 1000 * extra_qxdm_logs_in_seconds 618 else: 619 sdm_begin_time = None 620 try: 621 time.sleep(10) 622 ad.get_sdm_logs(test_name, sdm_begin_time) 623 except Exception as e: 624 ad.log.error("Failed to get SDM log for %s with error %s", 625 test_name, e) 626 result = False 627 628 return result 629 630 def _take_bug_report(self, test_name, begin_time): 631 if self._skip_bug_report(test_name): 632 return 633 dev_num = getattr(self, "number_of_devices", None) or len( 634 self.android_devices) 635 tasks = [(self._ad_take_bugreport, (ad, test_name, begin_time)) 636 for ad in self.android_devices[:dev_num]] 637 tasks.extend([(self._ad_take_extra_logs, (ad, test_name, begin_time)) 638 for ad in self.android_devices[:dev_num]]) 639 run_multithread_func(self.log, tasks) 640 for ad in self.android_devices[:dev_num]: 641 if getattr(ad, "reboot_to_recover", False): 642 reboot_device(ad) 643 ad.reboot_to_recover = False 644 # Zip log folder 645 if not self.user_params.get("zip_log", False): return 646 src_dir = os.path.join(self.log_path, test_name) 647 os.makedirs(src_dir, exist_ok=True) 648 file_name = "%s_%s" % (src_dir, begin_time) 649 self.log.info("Zip folder %s to %s.zip", src_dir, file_name) 650 shutil.make_archive(file_name, "zip", src_dir) 651 shutil.rmtree(src_dir) 652 653 def _block_all_test_cases(self, tests, reason='Failed class setup'): 654 """Over-write _block_all_test_cases in BaseTestClass.""" 655 for (i, (test_name, test_func)) in enumerate(tests): 656 signal = signals.TestFailure(reason) 657 record = records.TestResultRecord(test_name, self.TAG) 658 record.test_begin() 659 # mark all test cases as FAIL 660 record.test_fail(signal) 661 self.results.add_record(record) 662 # only gather bug report for the first test case 663 if i == 0: 664 self.on_fail(test_name, record.begin_time) 665 666 def get_stress_test_number(self): 667 """Gets the stress_test_number param from user params. 668 669 Gets the stress_test_number param. If absent, returns default 100. 670 """ 671 return int(self.user_params.get("stress_test_number", 100)) 672