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