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