1 #!/usr/bin/env python3.4 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 20 import logging 21 import os 22 import re 23 import shutil 24 import traceback 25 26 import acts.controllers.diag_logger 27 28 from acts import asserts 29 from acts import logger as acts_logger 30 from acts.base_test import BaseTestClass 31 from acts.controllers.android_device import DEFAULT_QXDM_LOG_PATH 32 from acts.keys import Config 33 from acts.signals import TestSignal 34 from acts.signals import TestAbortClass 35 from acts.signals import TestAbortAll 36 from acts.signals import TestBlocked 37 from acts import records 38 from acts import utils 39 40 from acts.test_utils.tel.tel_subscription_utils import \ 41 initial_set_up_for_subid_infomation 42 from acts.test_utils.tel.tel_test_utils import abort_all_tests 43 from acts.test_utils.tel.tel_test_utils import ensure_phones_default_state 44 from acts.test_utils.tel.tel_test_utils import ensure_phones_idle 45 from acts.test_utils.tel.tel_test_utils import print_radio_info 46 from acts.test_utils.tel.tel_test_utils import reboot_device 47 from acts.test_utils.tel.tel_test_utils import refresh_sl4a_session 48 from acts.test_utils.tel.tel_test_utils import run_multithread_func 49 from acts.test_utils.tel.tel_test_utils import setup_droid_properties 50 from acts.test_utils.tel.tel_test_utils import set_phone_screen_on 51 from acts.test_utils.tel.tel_test_utils import set_phone_silent_mode 52 from acts.test_utils.tel.tel_test_utils import set_qxdm_logger_command 53 from acts.test_utils.tel.tel_test_utils import start_qxdm_loggers 54 from acts.test_utils.tel.tel_test_utils import stop_qxdm_loggers 55 from acts.test_utils.tel.tel_test_utils import unlock_sim 56 from acts.test_utils.tel.tel_defines import PRECISE_CALL_STATE_LISTEN_LEVEL_BACKGROUND 57 from acts.test_utils.tel.tel_defines import PRECISE_CALL_STATE_LISTEN_LEVEL_FOREGROUND 58 from acts.test_utils.tel.tel_defines import PRECISE_CALL_STATE_LISTEN_LEVEL_RINGING 59 from acts.test_utils.tel.tel_defines import WIFI_VERBOSE_LOGGING_ENABLED 60 from acts.test_utils.tel.tel_defines import WIFI_VERBOSE_LOGGING_DISABLED 61 62 63 class TelephonyBaseTest(BaseTestClass): 64 def __init__(self, controllers): 65 66 BaseTestClass.__init__(self, controllers) 67 self.logger_sessions = [] 68 69 self.log_path = getattr(logging, "log_path", None) 70 self.qxdm_log = self.user_params.get("qxdm_log", True) 71 qxdm_log_mask_cfg = self.user_params.get("qxdm_log_mask_cfg", None) 72 if isinstance(qxdm_log_mask_cfg, list): 73 qxdm_log_mask_cfg = qxdm_log_mask_cfg[0] 74 if qxdm_log_mask_cfg and "dev/null" in qxdm_log_mask_cfg: 75 qxdm_log_mask_cfg = None 76 stop_qxdm_loggers(self.log, self.android_devices) 77 for ad in self.android_devices: 78 try: 79 ad.adb.shell("killall -9 tcpdump") 80 except AdbError: 81 ad.log.warn("Killing existing tcpdump processes failed") 82 if not hasattr(ad, "init_log_path"): 83 ad.init_log_path = ad.log_path 84 ad.log_path = self.log_path 85 print_radio_info(ad) 86 if not unlock_sim(ad): 87 abort_all_tests(ad.log, "unable to unlock SIM") 88 ad.wakeup_screen() 89 ad.adb.shell("input keyevent 82") 90 ad.qxdm_log = getattr(ad, "qxdm_log", self.qxdm_log) 91 if ad.qxdm_log: 92 qxdm_log_mask = getattr(ad, "qxdm_log_mask", None) 93 if qxdm_log_mask_cfg: 94 qxdm_mask_path = DEFAULT_QXDM_LOG_PATH 95 ad.adb.shell("mkdir %s" % qxdm_mask_path) 96 ad.log.info("Push %s to %s", qxdm_log_mask_cfg, 97 qxdm_mask_path) 98 ad.adb.push("%s %s" % (qxdm_log_mask_cfg, qxdm_mask_path)) 99 mask_file_name = os.path.split(qxdm_log_mask_cfg)[-1] 100 qxdm_log_mask = os.path.join(qxdm_mask_path, 101 mask_file_name) 102 set_qxdm_logger_command(ad, mask=qxdm_log_mask) 103 ad.adb.pull( 104 "/firmware/image/qdsp6m.qdb %s" % ad.init_log_path, 105 ignore_status=True) 106 107 start_qxdm_loggers(self.log, self.android_devices, 108 utils.get_current_epoch_time()) 109 self.skip_reset_between_cases = self.user_params.get( 110 "skip_reset_between_cases", True) 111 112 # Use for logging in the test cases to facilitate 113 # faster log lookup and reduce ambiguity in logging. 114 @staticmethod 115 def tel_test_wrap(fn): 116 def _safe_wrap_test_case(self, *args, **kwargs): 117 test_id = "%s:%s:%s" % (self.__class__.__name__, self.test_name, 118 self.log_begin_time.replace(' ', '-')) 119 self.test_id = test_id 120 self.result_detail = "" 121 tries = 2 if self.user_params.get("telephony_auto_rerun") else 1 122 for i in range(tries): 123 result = True 124 log_string = "[Test ID] %s" % test_id 125 if i > 1: 126 log_string = "[Rerun]%s" % log_string 127 self.teardown_test() 128 self.setup_test() 129 self.log.info(log_string) 130 for ad in self.android_devices: 131 ad.log_path = self.log_path 132 try: 133 ad.droid.logI("Started %s" % log_string) 134 except Exception as e: 135 ad.log.warning(e) 136 refresh_sl4a_session(ad) 137 try: 138 result = fn(self, *args, **kwargs) 139 except (TestSignal, TestAbortClass, TestAbortAll) as signal: 140 if self.result_detail: 141 signal.details = self.result_detail 142 raise 143 except Exception as e: 144 self.log.error(traceback.format_exc()) 145 asserts.fail(self.result_detail) 146 for ad in self.android_devices: 147 try: 148 ad.droid.logI("Finished %s" % log_string) 149 except Exception as e: 150 ad.log.warning(e) 151 refresh_sl4a_session(ad) 152 if result: break 153 if self.user_params.get("check_crash", True): 154 new_crash = ad.check_crash_report(self.test_name, 155 self.begin_time, True) 156 if new_crash: 157 msg = "Find new crash reports %s" % new_crash 158 ad.log.error(msg) 159 self.result_detail = "%s %s %s" % (self.result_detail, 160 ad.serial, msg) 161 result = False 162 if result: 163 asserts.explicit_pass(self.result_detail) 164 else: 165 asserts.fail(self.result_detail) 166 167 return _safe_wrap_test_case 168 169 def setup_class(self): 170 sim_conf_file = self.user_params.get("sim_conf_file") 171 if not sim_conf_file: 172 self.log.info("\"sim_conf_file\" is not provided test bed config!") 173 else: 174 if isinstance(sim_conf_file, list): 175 sim_conf_file = sim_conf_file[0] 176 # If the sim_conf_file is not a full path, attempt to find it 177 # relative to the config file. 178 if not os.path.isfile(sim_conf_file): 179 sim_conf_file = os.path.join( 180 self.user_params[Config.key_config_path], sim_conf_file) 181 if not os.path.isfile(sim_conf_file): 182 self.log.error("Unable to load user config %s ", 183 sim_conf_file) 184 185 setattr(self, "diag_logger", 186 self.register_controller( 187 acts.controllers.diag_logger, required=False)) 188 189 if not self.user_params.get("Attenuator"): 190 ensure_phones_default_state(self.log, self.android_devices) 191 else: 192 ensure_phones_idle(self.log, self.android_devices) 193 for ad in self.android_devices: 194 setup_droid_properties(self.log, ad, sim_conf_file) 195 196 # Setup VoWiFi MDN for Verizon. b/33187374 197 build_id = ad.build_info["build_id"] 198 if "vzw" in [ 199 sub["operator"] for sub in ad.cfg["subscription"].values() 200 ] and ad.is_apk_installed("com.google.android.wfcactivation"): 201 ad.log.info("setup VoWiFi MDN per b/33187374") 202 ad.adb.shell("setprop dbg.vzw.force_wfc_nv_enabled true") 203 ad.adb.shell("am start --ei EXTRA_LAUNCH_CARRIER_APP 0 -n " 204 "\"com.google.android.wfcactivation/" 205 ".VzwEmergencyAddressActivity\"") 206 # Sub ID setup 207 initial_set_up_for_subid_infomation(self.log, ad) 208 if "enable_wifi_verbose_logging" in self.user_params: 209 ad.droid.wifiEnableVerboseLogging(WIFI_VERBOSE_LOGGING_ENABLED) 210 # If device is setup already, skip the following setup procedures 211 if getattr(ad, "telephony_test_setup", None): 212 continue 213 # Disable Emergency alerts 214 # Set chrome browser start with no-first-run verification and 215 # disable-fre. Give permission to read from and write to storage. 216 for cmd in ( 217 "pm disable com.android.cellbroadcastreceiver", 218 "pm grant com.android.chrome " 219 "android.permission.READ_EXTERNAL_STORAGE", 220 "pm grant com.android.chrome " 221 "android.permission.WRITE_EXTERNAL_STORAGE", 222 "rm /data/local/chrome-command-line", 223 "am set-debug-app --persistent com.android.chrome", 224 'echo "chrome --no-default-browser-check --no-first-run ' 225 '--disable-fre" > /data/local/tmp/chrome-command-line'): 226 ad.adb.shell(cmd) 227 228 # Curl for 2016/7 devices 229 try: 230 if int(ad.adb.getprop("ro.product.first_api_level")) >= 25: 231 out = ad.adb.shell("/data/curl --version") 232 if not out or "not found" in out: 233 tel_data = self.user_params.get("tel_data", "tel_data") 234 if isinstance(tel_data, list): 235 tel_data = tel_data[0] 236 curl_file_path = os.path.join(tel_data, "curl") 237 if not os.path.isfile(curl_file_path): 238 curl_file_path = os.path.join( 239 self.user_params[Config.key_config_path], 240 curl_file_path) 241 if os.path.isfile(curl_file_path): 242 ad.log.info("Pushing Curl to /data dir") 243 ad.adb.push("%s /data" % (curl_file_path)) 244 ad.adb.shell( 245 "chmod 777 /data/curl", ignore_status=True) 246 except Exception: 247 ad.log.info("Failed to push curl on this device") 248 249 # Ensure that a test class starts from a consistent state that 250 # improves chances of valid network selection and facilitates 251 # logging. 252 try: 253 if not set_phone_screen_on(self.log, ad): 254 self.log.error("Failed to set phone screen-on time.") 255 return False 256 if not set_phone_silent_mode(self.log, ad): 257 self.log.error("Failed to set phone silent mode.") 258 return False 259 ad.droid.telephonyAdjustPreciseCallStateListenLevel( 260 PRECISE_CALL_STATE_LISTEN_LEVEL_FOREGROUND, True) 261 ad.droid.telephonyAdjustPreciseCallStateListenLevel( 262 PRECISE_CALL_STATE_LISTEN_LEVEL_RINGING, True) 263 ad.droid.telephonyAdjustPreciseCallStateListenLevel( 264 PRECISE_CALL_STATE_LISTEN_LEVEL_BACKGROUND, True) 265 except Exception as e: 266 self.log.error("Failure with %s", e) 267 setattr(ad, "telephony_test_setup", True) 268 269 return True 270 271 def teardown_class(self): 272 stop_qxdm_loggers(self.log, self.android_devices) 273 ensure_phones_default_state(self.log, self.android_devices) 274 try: 275 for ad in self.android_devices: 276 ad.droid.disableDevicePassword() 277 if "enable_wifi_verbose_logging" in self.user_params: 278 ad.droid.wifiEnableVerboseLogging( 279 WIFI_VERBOSE_LOGGING_DISABLED) 280 if hasattr(ad, "init_log_path"): 281 ad.log_path = ad.init_log_path 282 return True 283 except Exception as e: 284 self.log.error("Failure with %s", e) 285 286 def setup_test(self): 287 for ad in self.android_devices: 288 ad.ed.clear_all_events() 289 output = ad.adb.logcat("-t 1") 290 match = re.search(r"\d+-\d+\s\d+:\d+:\d+.\d+", output) 291 if match: 292 ad.test_log_begin_time = match.group(0) 293 if getattr(self, "qxdm_log", True): 294 start_qxdm_loggers(self.log, self.android_devices, self.begin_time) 295 if getattr(self, "diag_logger", None): 296 for logger in self.diag_logger: 297 self.log.info("Starting a diagnostic session %s", logger) 298 self.logger_sessions.append((logger, logger.start())) 299 if self.skip_reset_between_cases: 300 ensure_phones_idle(self.log, self.android_devices) 301 else: 302 ensure_phones_default_state(self.log, self.android_devices) 303 304 def on_exception(self, test_name, begin_time): 305 self._pull_diag_logs(test_name, begin_time) 306 self._take_bug_report(test_name, begin_time) 307 self._cleanup_logger_sessions() 308 309 def on_fail(self, test_name, begin_time): 310 self._pull_diag_logs(test_name, begin_time) 311 self._take_bug_report(test_name, begin_time) 312 self._cleanup_logger_sessions() 313 314 def on_blocked(self, test_name, begin_time): 315 self.on_fail(test_name, begin_time) 316 317 def _ad_take_extra_logs(self, ad, test_name, begin_time): 318 extra_qxdm_logs_in_seconds = self.user_params.get( 319 "extra_qxdm_logs_in_seconds", 60 * 3) 320 result = True 321 if getattr(ad, "qxdm_log", True): 322 # Gather qxdm log modified 3 minutes earlier than test start time 323 if begin_time: 324 qxdm_begin_time = begin_time - 1000 * extra_qxdm_logs_in_seconds 325 else: 326 qxdm_begin_time = None 327 try: 328 ad.get_qxdm_logs(test_name, qxdm_begin_time) 329 except Exception as e: 330 ad.log.error("Failed to get QXDM log for %s with error %s", 331 test_name, e) 332 result = False 333 334 try: 335 ad.check_crash_report(test_name, begin_time, log_crash_report=True) 336 except Exception as e: 337 ad.log.error("Failed to check crash report for %s with error %s", 338 test_name, e) 339 result = False 340 341 log_begin_time = getattr( 342 ad, "test_log_begin_time", None 343 ) or acts_logger.epoch_to_log_line_timestamp(begin_time - 1000 * 60) 344 log_path = os.path.join(self.log_path, test_name, 345 "%s_%s.logcat" % (ad.serial, begin_time)) 346 try: 347 ad.adb.logcat( 348 'b all -d -v year -t "%s" > %s' % (log_begin_time, log_path), 349 timeout=120) 350 except Exception as e: 351 ad.log.error("Failed to get logcat with error %s", e) 352 result = False 353 return result 354 355 def _take_bug_report(self, test_name, begin_time): 356 if self._skip_bug_report(): 357 return 358 dev_num = getattr(self, "number_of_devices", None) or len( 359 self.android_devices) 360 tasks = [(self._ad_take_bugreport, (ad, test_name, begin_time)) 361 for ad in self.android_devices[:dev_num]] 362 tasks.extend([(self._ad_take_extra_logs, (ad, test_name, begin_time)) 363 for ad in self.android_devices[:dev_num]]) 364 run_multithread_func(self.log, tasks) 365 for ad in self.android_devices[:dev_num]: 366 if getattr(ad, "reboot_to_recover", False): 367 reboot_device(ad) 368 ad.reboot_to_recover = False 369 if not self.user_params.get("zip_log", False): return 370 src_dir = os.path.join(self.log_path, test_name) 371 file_name = "%s_%s" % (src_dir, begin_time) 372 self.log.info("Zip folder %s to %s.zip", src_dir, file_name) 373 shutil.make_archive(file_name, "zip", src_dir) 374 shutil.rmtree(src_dir) 375 376 def _block_all_test_cases(self, tests): 377 """Over-write _block_all_test_case in BaseTestClass.""" 378 for (i, (test_name, test_func)) in enumerate(tests): 379 signal = TestBlocked("Failed class setup") 380 record = records.TestResultRecord(test_name, self.TAG) 381 record.test_begin() 382 # mark all test cases as FAIL 383 record.test_fail(signal) 384 self.results.add_record(record) 385 # only gather bug report for the first test case 386 if i == 0: 387 self.on_fail(test_name, record.begin_time) 388 389 def on_pass(self, test_name, begin_time): 390 self._cleanup_logger_sessions() 391 392 def get_stress_test_number(self): 393 """Gets the stress_test_number param from user params. 394 395 Gets the stress_test_number param. If absent, returns default 100. 396 """ 397 return int(self.user_params.get("stress_test_number", 100)) 398
