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