| #!/usr/bin/env python3 |
| # |
| # Copyright 2016 - Google |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| """ |
| Base Class for Defining Common Telephony Test Functionality |
| """ |
| |
| import logging |
| import os |
| import re |
| import shutil |
| import time |
| |
| from acts import asserts |
| from acts import signals |
| from acts.base_test import BaseTestClass |
| from acts.controllers.android_device import DEFAULT_QXDM_LOG_PATH |
| from acts.keys import Config |
| from acts import records |
| from acts import utils |
| from acts.libs.utils.multithread import multithread_func |
| from acts.libs.utils.multithread import run_multithread_func |
| from acts_contrib.test_utils.tel.tel_defines import PRECISE_CALL_STATE_LISTEN_LEVEL_BACKGROUND |
| from acts_contrib.test_utils.tel.tel_defines import SINGLE_SIM_CONFIG, MULTI_SIM_CONFIG |
| from acts_contrib.test_utils.tel.tel_defines import PRECISE_CALL_STATE_LISTEN_LEVEL_FOREGROUND |
| from acts_contrib.test_utils.tel.tel_defines import PRECISE_CALL_STATE_LISTEN_LEVEL_RINGING |
| from acts_contrib.test_utils.tel.tel_defines import SIM_STATE_ABSENT |
| from acts_contrib.test_utils.tel.tel_defines import SIM_STATE_UNKNOWN |
| from acts_contrib.test_utils.tel.tel_defines import WIFI_VERBOSE_LOGGING_DISABLED |
| from acts_contrib.test_utils.tel.tel_defines import INVALID_SUB_ID |
| from acts_contrib.test_utils.tel.tel_defines import CHIPSET_MODELS_LIST |
| from acts_contrib.test_utils.tel.tel_bootloader_utils import flash_radio |
| from acts_contrib.test_utils.tel.tel_ims_utils import activate_wfc_on_device |
| from acts_contrib.test_utils.tel.tel_logging_utils import disable_qxdm_logger |
| from acts_contrib.test_utils.tel.tel_logging_utils import get_screen_shot_log |
| from acts_contrib.test_utils.tel.tel_logging_utils import set_qxdm_logger_command |
| from acts_contrib.test_utils.tel.tel_logging_utils import start_qxdm_logger |
| from acts_contrib.test_utils.tel.tel_logging_utils import start_qxdm_loggers |
| from acts_contrib.test_utils.tel.tel_logging_utils import stop_qxdm_logger |
| from acts_contrib.test_utils.tel.tel_logging_utils import start_sdm_loggers |
| from acts_contrib.test_utils.tel.tel_logging_utils import start_sdm_logger |
| from acts_contrib.test_utils.tel.tel_logging_utils import stop_sdm_logger |
| from acts_contrib.test_utils.tel.tel_logging_utils import start_tcpdumps |
| from acts_contrib.test_utils.tel.tel_logging_utils import stop_tcpdumps |
| from acts_contrib.test_utils.tel.tel_logging_utils import get_tcpdump_log |
| from acts_contrib.test_utils.tel.tel_phone_setup_utils import ensure_phone_default_state |
| from acts_contrib.test_utils.tel.tel_phone_setup_utils import ensure_phone_idle |
| from acts_contrib.test_utils.tel.tel_subscription_utils import initial_set_up_for_subid_information |
| from acts_contrib.test_utils.tel.tel_subscription_utils import set_default_sub_for_all_services |
| from acts_contrib.test_utils.tel.tel_subscription_utils import get_subid_from_slot_index |
| from acts_contrib.test_utils.tel.tel_test_utils import build_id_override |
| from acts_contrib.test_utils.tel.tel_test_utils import enable_connectivity_metrics |
| from acts_contrib.test_utils.tel.tel_test_utils import enable_radio_log_on |
| from acts_contrib.test_utils.tel.tel_test_utils import force_connectivity_metrics_upload |
| from acts_contrib.test_utils.tel.tel_test_utils import get_sim_state |
| from acts_contrib.test_utils.tel.tel_test_utils import install_apk |
| from acts_contrib.test_utils.tel.tel_test_utils import print_radio_info |
| from acts_contrib.test_utils.tel.tel_test_utils import reboot_device |
| from acts_contrib.test_utils.tel.tel_test_utils import recover_build_id |
| from acts_contrib.test_utils.tel.tel_test_utils import setup_droid_properties |
| from acts_contrib.test_utils.tel.tel_test_utils import set_phone_screen_on |
| from acts_contrib.test_utils.tel.tel_test_utils import set_phone_silent_mode |
| from acts_contrib.test_utils.tel.tel_test_utils import synchronize_device_time |
| from acts_contrib.test_utils.tel.tel_test_utils import unlock_sim |
| from acts_contrib.test_utils.tel.tel_test_utils import wait_for_sim_ready_by_adb |
| from acts_contrib.test_utils.tel.tel_test_utils import wait_for_sims_ready_by_adb |
| from acts_contrib.test_utils.tel.tel_test_utils import install_googleaccountutil_apk |
| from acts_contrib.test_utils.tel.tel_test_utils import add_google_account |
| from acts_contrib.test_utils.tel.tel_test_utils import install_googlefi_apk |
| from acts_contrib.test_utils.tel.tel_test_utils import activate_google_fi_account |
| from acts_contrib.test_utils.tel.tel_test_utils import check_google_fi_activated |
| from acts_contrib.test_utils.tel.tel_test_utils import phone_switch_to_msim_mode |
| from acts_contrib.test_utils.tel.tel_test_utils import activate_esim_using_suw |
| from acts_contrib.test_utils.tel.tel_wifi_utils import ensure_wifi_connected |
| |
| |
| class TelephonyBaseTest(BaseTestClass): |
| # Use for logging in the test cases to facilitate |
| # faster log lookup and reduce ambiguity in logging. |
| @staticmethod |
| def tel_test_wrap(fn): |
| def _safe_wrap_test_case(self, *args, **kwargs): |
| test_id = "%s:%s:%s" % (self.__class__.__name__, self.test_name, |
| self.log_begin_time.replace(' ', '-')) |
| self.test_id = test_id |
| self.result_detail = "" |
| self.testsignal_details = "" |
| self.testsignal_extras = {} |
| tries = int(self.user_params.get("telephony_auto_rerun", 1)) |
| for ad in self.android_devices: |
| ad.log_path = self.log_path |
| for i in range(tries + 1): |
| result = True |
| if i > 0: |
| log_string = "[Test Case] RERUN %s" % self.test_name |
| self.log.info(log_string) |
| self._teardown_test(self.test_name) |
| self._setup_test(self.test_name) |
| try: |
| result = fn(self, *args, **kwargs) |
| except signals.TestFailure as e: |
| self.testsignal_details = e.details |
| self.testsignal_extras = e.extras |
| result = False |
| except signals.TestSignal: |
| raise |
| except Exception as e: |
| self.log.exception(e) |
| asserts.fail(self.result_detail) |
| if result is False: |
| if i < tries: |
| continue |
| else: |
| break |
| if self.user_params.get("check_crash", True): |
| new_crash = ad.check_crash_report(self.test_name, |
| self.begin_time, True) |
| if new_crash: |
| msg = "Find new crash reports %s" % new_crash |
| ad.log.error(msg) |
| self.result_detail = "%s %s %s" % (self.result_detail, |
| ad.serial, msg) |
| result = False |
| if result is not False: |
| asserts.explicit_pass(self.result_detail) |
| else: |
| if self.result_detail: |
| asserts.fail(self.result_detail) |
| else: |
| asserts.fail(self.testsignal_details, self.testsignal_extras) |
| |
| return _safe_wrap_test_case |
| |
| def setup_class(self): |
| super().setup_class() |
| self.wifi_network_ssid = self.user_params.get( |
| "wifi_network_ssid") or self.user_params.get( |
| "wifi_network_ssid_2g") or self.user_params.get( |
| "wifi_network_ssid_5g") |
| self.wifi_network_pass = self.user_params.get( |
| "wifi_network_pass") or self.user_params.get( |
| "wifi_network_pass_2g") or self.user_params.get( |
| "wifi_network_ssid_5g") |
| |
| self.log_path = getattr(logging, "log_path", None) |
| self.qxdm_log = self.user_params.get("qxdm_log", True) |
| self.sdm_log = self.user_params.get("sdm_log", False) |
| self.enable_radio_log_on = self.user_params.get( |
| "enable_radio_log_on", False) |
| self.cbrs_esim = self.user_params.get("cbrs_esim", False) |
| self.account_util = self.user_params.get("account_util", None) |
| self.save_passing_logs = self.user_params.get("save_passing_logs", False) |
| if isinstance(self.account_util, list): |
| self.account_util = self.account_util[0] |
| self.fi_util = self.user_params.get("fi_util", None) |
| if isinstance(self.fi_util, list): |
| self.fi_util = self.fi_util[0] |
| self.radio_img = self.user_params.get("radio_img", None) |
| if isinstance(self.radio_img, list): |
| self.radio_img = self.radio_img[0] |
| self.modem_bin = self.user_params.get("modem_bin", None) |
| if isinstance(self.modem_bin, list): |
| self.modem_bin = self.modem_bin[0] |
| self.extra_apk = self.user_params.get("extra_apk", None) |
| if isinstance(self.extra_apk, list): |
| self.extra_apk = self.extra_apk[0] |
| self.extra_package = self.user_params.get("extra_package", None) |
| |
| if self.radio_img or self.modem_bin: |
| sideload_img = True |
| if self.radio_img: |
| file_path = self.radio_img |
| elif self.modem_bin: |
| file_path = self.modem_bin |
| sideload_img = False |
| tasks = [(flash_radio, [ad, file_path, True, sideload_img]) |
| for ad in self.android_devices] |
| multithread_func(self.log, tasks) |
| if self.extra_apk and self.extra_package: |
| tasks = [(install_apk, [ad, self.extra_apk, self.extra_package]) |
| for ad in self.android_devices] |
| multithread_func(self.log, tasks) |
| |
| tasks = [(self._init_device, [ad]) for ad in self.android_devices] |
| multithread_func(self.log, tasks) |
| self.skip_reset_between_cases = self.user_params.get( |
| "skip_reset_between_cases", True) |
| self.log_path = getattr(logging, "log_path", None) |
| self.sim_config = { |
| "config":SINGLE_SIM_CONFIG, |
| "number_of_sims":1 |
| } |
| |
| for ad in self.android_devices: |
| if getattr(ad, 'dsds', False): |
| self.sim_config = { |
| "config":MULTI_SIM_CONFIG, |
| "number_of_sims":2 |
| } |
| break |
| if "anritsu_md8475a_ip_address" in self.user_params: |
| return |
| qxdm_log_mask_cfg = self.user_params.get("qxdm_log_mask_cfg", None) |
| if isinstance(qxdm_log_mask_cfg, list): |
| qxdm_log_mask_cfg = qxdm_log_mask_cfg[0] |
| if qxdm_log_mask_cfg and "dev/null" in qxdm_log_mask_cfg: |
| qxdm_log_mask_cfg = None |
| sim_conf_file = self.user_params.get("sim_conf_file") |
| if not sim_conf_file: |
| self.log.info("\"sim_conf_file\" is not provided test bed config!") |
| else: |
| if isinstance(sim_conf_file, list): |
| sim_conf_file = sim_conf_file[0] |
| # If the sim_conf_file is not a full path, attempt to find it |
| # relative to the config file. |
| if not os.path.isfile(sim_conf_file): |
| sim_conf_file = os.path.join( |
| self.user_params[Config.key_config_path.value], |
| sim_conf_file) |
| if not os.path.isfile(sim_conf_file): |
| self.log.error("Unable to load user config %s ", |
| sim_conf_file) |
| |
| tasks = [(self._setup_device, [ad, sim_conf_file, qxdm_log_mask_cfg]) |
| for ad in self.android_devices] |
| return multithread_func(self.log, tasks) |
| |
| def _init_device(self, ad): |
| synchronize_device_time(ad) |
| ad.log_path = self.log_path |
| print_radio_info(ad) |
| unlock_sim(ad) |
| ad.wakeup_screen() |
| ad.adb.shell("input keyevent 82") |
| |
| def wait_for_sim_ready(self,ad): |
| wait_for_sim_ready_on_sim_config = { |
| SINGLE_SIM_CONFIG : lambda:wait_for_sim_ready_by_adb(self.log,ad), |
| MULTI_SIM_CONFIG : lambda:wait_for_sims_ready_by_adb(self.log,ad) |
| } |
| if not wait_for_sim_ready_on_sim_config[self.sim_config["config"]]: |
| raise signals.TestAbortClass("unable to load the SIM") |
| |
| def _setup_device(self, ad, sim_conf_file, qxdm_log_mask_cfg=None): |
| ad.qxdm_log = getattr(ad, "qxdm_log", self.qxdm_log) |
| ad.sdm_log = getattr(ad, "sdm_log", self.sdm_log) |
| if self.user_params.get("enable_connectivity_metrics", False): |
| enable_connectivity_metrics(ad) |
| if self.user_params.get("build_id_override", False): |
| build_postfix = self.user_params.get("build_id_postfix", |
| "LAB_TEST") |
| build_id_override( |
| ad, |
| new_build_id=self.user_params.get("build_id_override_with", |
| None), |
| postfix=build_postfix) |
| if self.enable_radio_log_on: |
| enable_radio_log_on(ad) |
| list_of_models = CHIPSET_MODELS_LIST |
| if any(model in ad.model for model in list_of_models): |
| phone_mode = "ssss" |
| if hasattr(ad, "mtp_dsds"): |
| phone_mode = "dsds" |
| if ad.adb.getprop("persist.radio.multisim.config") != phone_mode: |
| ad.adb.shell("setprop persist.radio.multisim.config %s" \ |
| % phone_mode) |
| reboot_device(ad) |
| |
| stop_qxdm_logger(ad) |
| if ad.qxdm_log: |
| qxdm_log_mask = getattr(ad, "qxdm_log_mask", None) |
| if qxdm_log_mask_cfg: |
| qxdm_mask_path = self.user_params.get("qxdm_log_path", |
| DEFAULT_QXDM_LOG_PATH) |
| ad.adb.shell("mkdir %s" % qxdm_mask_path, ignore_status=True) |
| ad.log.info("Push %s to %s", qxdm_log_mask_cfg, qxdm_mask_path) |
| ad.adb.push("%s %s" % (qxdm_log_mask_cfg, qxdm_mask_path)) |
| mask_file_name = os.path.split(qxdm_log_mask_cfg)[-1] |
| qxdm_log_mask = os.path.join(qxdm_mask_path, mask_file_name) |
| set_qxdm_logger_command(ad, mask=qxdm_log_mask) |
| start_qxdm_logger(ad, utils.get_current_epoch_time()) |
| elif ad.sdm_log: |
| start_sdm_logger(ad) |
| else: |
| disable_qxdm_logger(ad) |
| if not unlock_sim(ad): |
| raise signals.TestAbortClass("unable to unlock the SIM") |
| |
| # If device is setup already, skip the following setup procedures |
| if getattr(ad, "telephony_test_setup", None): |
| return True |
| |
| # eSIM enablement |
| if hasattr(ad, "fi_esim"): |
| if not ensure_wifi_connected(self.log, ad, self.wifi_network_ssid, |
| self.wifi_network_pass): |
| ad.log.error("Failed to connect to wifi") |
| if check_google_fi_activated(ad): |
| ad.log.info("Google Fi is already Activated") |
| else: |
| install_googleaccountutil_apk(ad, self.account_util) |
| add_google_account(ad) |
| install_googlefi_apk(ad, self.fi_util) |
| if not activate_google_fi_account(ad): |
| ad.log.error("Failed to activate Fi") |
| check_google_fi_activated(ad) |
| if getattr(ad, 'dsds', False): |
| sim_mode = ad.droid.telephonyGetPhoneCount() |
| if sim_mode == 1: |
| ad.log.info("Phone in Single SIM Mode") |
| if not phone_switch_to_msim_mode(ad): |
| ad.log.error("Failed to switch to Dual SIM Mode") |
| return False |
| elif sim_mode == 2: |
| ad.log.info("Phone already in Dual SIM Mode") |
| if get_sim_state(ad) in (SIM_STATE_ABSENT, SIM_STATE_UNKNOWN): |
| ad.log.info("Device has no or unknown SIM in it") |
| # eSIM needs activation |
| activate_esim_using_suw(ad) |
| ensure_phone_idle(self.log, ad) |
| setup_droid_properties(self.log, ad, sim_conf_file) |
| elif self.user_params.get("Attenuator"): |
| ad.log.info("Device in chamber room") |
| ensure_phone_idle(self.log, ad) |
| setup_droid_properties(self.log, ad, sim_conf_file) |
| else: |
| self.wait_for_sim_ready(ad) |
| ensure_phone_default_state(self.log, ad) |
| setup_droid_properties(self.log, ad, sim_conf_file) |
| |
| if getattr(ad, 'dsds', False): |
| default_slot = getattr(ad, "default_slot", 0) |
| if get_subid_from_slot_index(ad.log, ad, default_slot) != INVALID_SUB_ID: |
| ad.log.info("Slot %s is the default slot.", default_slot) |
| set_default_sub_for_all_services(ad, default_slot) |
| else: |
| ad.log.warning("Slot %s is NOT a valid slot. Slot %s will be used by default.", |
| default_slot, 1-default_slot) |
| set_default_sub_for_all_services(ad, 1-default_slot) |
| setattr(ad, "default_slot", 1-default_slot) |
| |
| # Activate WFC on Verizon, AT&T and Canada operators as per # b/33187374 & |
| # b/122327716 |
| activate_wfc_on_device(self.log, ad) |
| |
| # Sub ID setup |
| initial_set_up_for_subid_information(self.log, ad) |
| |
| |
| #try: |
| # ad.droid.wifiEnableVerboseLogging(WIFI_VERBOSE_LOGGING_ENABLED) |
| #except Exception: |
| # pass |
| |
| # Disable Emergency alerts |
| # Set chrome browser start with no-first-run verification and |
| # disable-fre. Give permission to read from and write to storage. |
| for cmd in ("pm disable com.android.cellbroadcastreceiver", |
| "pm grant com.android.chrome " |
| "android.permission.READ_EXTERNAL_STORAGE", |
| "pm grant com.android.chrome " |
| "android.permission.WRITE_EXTERNAL_STORAGE", |
| "rm /data/local/chrome-command-line", |
| "am set-debug-app --persistent com.android.chrome", |
| 'echo "chrome --no-default-browser-check --no-first-run ' |
| '--disable-fre" > /data/local/tmp/chrome-command-line'): |
| ad.adb.shell(cmd, ignore_status=True) |
| |
| # Curl for 2016/7 devices |
| if not getattr(ad, "curl_capable", False): |
| try: |
| out = ad.adb.shell("/data/curl --version") |
| if not out or "not found" in out: |
| if int(ad.adb.getprop("ro.product.first_api_level")) >= 25: |
| tel_data = self.user_params.get("tel_data", "tel_data") |
| if isinstance(tel_data, list): |
| tel_data = tel_data[0] |
| curl_file_path = os.path.join(tel_data, "curl") |
| if not os.path.isfile(curl_file_path): |
| curl_file_path = os.path.join( |
| self.user_params[Config.key_config_path.value], |
| curl_file_path) |
| if os.path.isfile(curl_file_path): |
| ad.log.info("Pushing Curl to /data dir") |
| ad.adb.push("%s /data" % (curl_file_path)) |
| ad.adb.shell( |
| "chmod 777 /data/curl", ignore_status=True) |
| else: |
| setattr(ad, "curl_capable", True) |
| except Exception: |
| ad.log.info("Failed to push curl on this device") |
| |
| # Ensure that a test class starts from a consistent state that |
| # improves chances of valid network selection and facilitates |
| # logging. |
| try: |
| if not set_phone_screen_on(self.log, ad): |
| self.log.error("Failed to set phone screen-on time.") |
| return False |
| if not set_phone_silent_mode(self.log, ad): |
| self.log.error("Failed to set phone silent mode.") |
| return False |
| ad.droid.telephonyAdjustPreciseCallStateListenLevel( |
| PRECISE_CALL_STATE_LISTEN_LEVEL_FOREGROUND, True) |
| ad.droid.telephonyAdjustPreciseCallStateListenLevel( |
| PRECISE_CALL_STATE_LISTEN_LEVEL_RINGING, True) |
| ad.droid.telephonyAdjustPreciseCallStateListenLevel( |
| PRECISE_CALL_STATE_LISTEN_LEVEL_BACKGROUND, True) |
| except Exception as e: |
| self.log.error("Failure with %s", e) |
| setattr(ad, "telephony_test_setup", True) |
| return True |
| |
| def _teardown_device(self, ad): |
| try: |
| stop_qxdm_logger(ad) |
| stop_sdm_logger(ad) |
| except Exception as e: |
| self.log.error("Failure with %s", e) |
| try: |
| ad.droid.disableDevicePassword() |
| except Exception as e: |
| self.log.error("Failure with %s", e) |
| if self.user_params.get("enable_connectivity_metrics", False): |
| if not ensure_wifi_connected(self.log, ad, self.wifi_network_ssid, |
| self.wifi_network_pass): |
| ad.log.error("Failed to connect to wifi") |
| force_connectivity_metrics_upload(ad) |
| time.sleep(30) |
| try: |
| ad.droid.wifiEnableVerboseLogging(WIFI_VERBOSE_LOGGING_DISABLED) |
| except Exception as e: |
| self.log.error("Failure with %s", e) |
| try: |
| if self.user_params.get("build_id_override", |
| False) and self.user_params.get( |
| "recover_build_id", False): |
| recover_build_id(ad) |
| except Exception as e: |
| self.log.error("Failure with %s", e) |
| |
| def teardown_class(self): |
| tasks = [(self._teardown_device, [ad]) for ad in self.android_devices] |
| multithread_func(self.log, tasks) |
| return True |
| |
| def setup_test(self): |
| if getattr(self, "qxdm_log", True): |
| if not self.user_params.get("qxdm_log_mask_cfg", None): |
| if "wfc" in self.test_name: |
| for ad in self.android_devices: |
| if not getattr(ad, "qxdm_logger_command", None) or ( |
| "IMS_DS_CNE_LnX_Golden.cfg" not in getattr( |
| ad, "qxdm_logger_command", "")): |
| set_qxdm_logger_command( |
| ad, "IMS_DS_CNE_LnX_Golden.cfg") |
| else: |
| for ad in self.android_devices: |
| if not getattr(ad, "qxdm_logger_command", None) or ( |
| "IMS_DS_CNE_LnX_Golden.cfg" in getattr( |
| ad, "qxdm_logger_command", "")): |
| set_qxdm_logger_command(ad, None) |
| start_qxdm_loggers(self.log, self.android_devices, self.begin_time) |
| if getattr(self, "sdm_log", False): |
| start_sdm_loggers(self.log, self.android_devices) |
| if getattr(self, "tcpdump_log", False) or "wfc" in self.test_name or ( |
| "iwlan" in self.test_name): |
| mask = getattr(self, "tcpdump_mask", "all") |
| interface = getattr(self, "tcpdump_interface", "wlan0") |
| start_tcpdumps( |
| self.android_devices, |
| begin_time=self.begin_time, |
| interface=interface, |
| mask=mask) |
| else: |
| stop_tcpdumps(self.android_devices) |
| for ad in self.android_devices: |
| if self.skip_reset_between_cases: |
| ensure_phone_idle(self.log, ad) |
| else: |
| ensure_phone_default_state(self.log, ad) |
| for session in ad._sl4a_manager.sessions.values(): |
| ed = session.get_event_dispatcher() |
| ed.clear_all_events() |
| output = ad.adb.logcat("-t 1") |
| match = re.search(r"\d+-\d+\s\d+:\d+:\d+.\d+", output) |
| if match: |
| ad.test_log_begin_time = match.group(0) |
| |
| def teardown_test(self): |
| stop_tcpdumps(self.android_devices) |
| |
| def on_fail(self, test_name, begin_time): |
| for ad in self.android_devices: |
| ad.screenshot(f"{ad.serial}_last_screen") |
| self._take_bug_report(test_name, begin_time) |
| |
| def on_pass(self, test_name, begin_time): |
| if self.save_passing_logs: |
| self._take_bug_report(test_name, begin_time) |
| |
| def _ad_take_extra_logs(self, ad, test_name, begin_time): |
| ad.adb.wait_for_device() |
| result = True |
| |
| try: |
| # get tcpdump and screen shot log |
| get_tcpdump_log(ad, test_name, begin_time) |
| get_screen_shot_log(ad, test_name, begin_time) |
| except Exception as e: |
| ad.log.error("Exception error %s", e) |
| result = False |
| |
| try: |
| ad.check_crash_report(test_name, begin_time, log_crash_report=True) |
| except Exception as e: |
| ad.log.error("Failed to check crash report for %s with error %s", |
| test_name, e) |
| result = False |
| |
| extra_qxdm_logs_in_seconds = self.user_params.get( |
| "extra_qxdm_logs_in_seconds", 60 * 3) |
| if getattr(ad, "qxdm_log", True): |
| # Gather qxdm log modified 3 minutes earlier than test start time |
| if begin_time: |
| qxdm_begin_time = begin_time - 1000 * extra_qxdm_logs_in_seconds |
| else: |
| qxdm_begin_time = None |
| try: |
| time.sleep(10) |
| ad.get_qxdm_logs(test_name, qxdm_begin_time) |
| except Exception as e: |
| ad.log.error("Failed to get QXDM log for %s with error %s", |
| test_name, e) |
| result = False |
| if getattr(ad, "sdm_log", False): |
| # Gather sdm log modified 3 minutes earlier than test start time |
| if begin_time: |
| sdm_begin_time = begin_time - 1000 * extra_qxdm_logs_in_seconds |
| else: |
| sdm_begin_time = None |
| try: |
| time.sleep(10) |
| ad.get_sdm_logs(test_name, sdm_begin_time) |
| except Exception as e: |
| ad.log.error("Failed to get SDM log for %s with error %s", |
| test_name, e) |
| result = False |
| |
| return result |
| |
| def _take_bug_report(self, test_name, begin_time): |
| if self._skip_bug_report(test_name): |
| return |
| dev_num = getattr(self, "number_of_devices", None) or len( |
| self.android_devices) |
| tasks = [(self._ad_take_bugreport, (ad, test_name, begin_time)) |
| for ad in self.android_devices[:dev_num]] |
| tasks.extend([(self._ad_take_extra_logs, (ad, test_name, begin_time)) |
| for ad in self.android_devices[:dev_num]]) |
| run_multithread_func(self.log, tasks) |
| for ad in self.android_devices[:dev_num]: |
| if getattr(ad, "reboot_to_recover", False): |
| reboot_device(ad) |
| ad.reboot_to_recover = False |
| # Zip log folder |
| if not self.user_params.get("zip_log", False): return |
| src_dir = os.path.join(self.log_path, test_name) |
| os.makedirs(src_dir, exist_ok=True) |
| file_name = "%s_%s" % (src_dir, begin_time) |
| self.log.info("Zip folder %s to %s.zip", src_dir, file_name) |
| shutil.make_archive(file_name, "zip", src_dir) |
| shutil.rmtree(src_dir) |
| |
| def _block_all_test_cases(self, tests, reason='Failed class setup'): |
| """Over-write _block_all_test_cases in BaseTestClass.""" |
| for (i, (test_name, test_func)) in enumerate(tests): |
| signal = signals.TestFailure(reason) |
| record = records.TestResultRecord(test_name, self.TAG) |
| record.test_begin() |
| # mark all test cases as FAIL |
| record.test_fail(signal) |
| self.results.add_record(record) |
| # only gather bug report for the first test case |
| if i == 0: |
| self.on_fail(test_name, record.begin_time) |
| |
| def get_stress_test_number(self): |
| """Gets the stress_test_number param from user params. |
| |
| Gets the stress_test_number param. If absent, returns default 100. |
| """ |
| return int(self.user_params.get("stress_test_number", 100)) |
