| #!/usr/bin/env python3.4 |
| # |
| # Copyright 2022 - 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. |
| """ |
| Test Script for Telephony Stress Call Test |
| """ |
| |
| import collections |
| import json |
| import os |
| import random |
| import time |
| |
| from acts import context |
| from acts import signals |
| from acts.libs.proc import job |
| from acts.test_decorators import test_tracker_info |
| from acts_contrib.test_utils.tel.loggers.telephony_metric_logger import TelephonyMetricLogger |
| from acts_contrib.test_utils.tel.loggers.telephony_stress_metric_logger import TelephonyStressMetricLogger |
| from acts_contrib.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest |
| from acts_contrib.test_utils.tel.tel_defines import GEN_3G |
| from acts_contrib.test_utils.tel.tel_defines import GEN_4G |
| from acts_contrib.test_utils.tel.tel_defines import GOOGLE_CBRS_CARRIER_ID |
| from acts_contrib.test_utils.tel.tel_defines import GOOGLE_FI_CARRIER_ID |
| from acts_contrib.test_utils.tel.tel_defines import INCALL_UI_DISPLAY_BACKGROUND |
| from acts_contrib.test_utils.tel.tel_defines import MAX_WAIT_TIME_SMS_RECEIVE |
| from acts_contrib.test_utils.tel.tel_defines import NETWORK_MODE_WCDMA_ONLY |
| from acts_contrib.test_utils.tel.tel_defines import NETWORK_MODE_GLOBAL |
| from acts_contrib.test_utils.tel.tel_defines import NETWORK_MODE_CDMA |
| from acts_contrib.test_utils.tel.tel_defines import NETWORK_MODE_GSM_ONLY |
| from acts_contrib.test_utils.tel.tel_defines import NETWORK_MODE_TDSCDMA_GSM_WCDMA |
| from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_AFTER_MODE_CHANGE |
| from acts_contrib.test_utils.tel.tel_defines import WFC_MODE_CELLULAR_PREFERRED |
| from acts_contrib.test_utils.tel.tel_defines import WFC_MODE_WIFI_PREFERRED |
| from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_CHANGE_MESSAGE_SUB_ID |
| from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_CHANGE_VOICE_SUB_ID |
| from acts_contrib.test_utils.tel.tel_5g_test_utils import provision_device_for_5g |
| from acts_contrib.test_utils.tel.tel_5g_utils import is_current_network_5g |
| from acts_contrib.test_utils.tel.tel_data_utils import active_file_download_test |
| from acts_contrib.test_utils.tel.tel_ims_utils import set_wfc_mode |
| from acts_contrib.test_utils.tel.tel_logging_utils import extract_test_log |
| from acts_contrib.test_utils.tel.tel_logging_utils import start_qxdm_loggers |
| from acts_contrib.test_utils.tel.tel_logging_utils import start_sdm_loggers |
| from acts_contrib.test_utils.tel.tel_logging_utils import start_adb_tcpdump |
| from acts_contrib.test_utils.tel.tel_lookup_tables import is_rat_svd_capable |
| from acts_contrib.test_utils.tel.tel_message_utils import sms_send_receive_verify |
| from acts_contrib.test_utils.tel.tel_message_utils import mms_send_receive_verify |
| from acts_contrib.test_utils.tel.tel_phone_setup_utils import ensure_network_generation_for_subscription |
| from acts_contrib.test_utils.tel.tel_phone_setup_utils import phone_setup_csfb |
| from acts_contrib.test_utils.tel.tel_phone_setup_utils import phone_setup_iwlan |
| from acts_contrib.test_utils.tel.tel_phone_setup_utils import phone_setup_voice_3g |
| from acts_contrib.test_utils.tel.tel_phone_setup_utils import phone_setup_voice_2g |
| from acts_contrib.test_utils.tel.tel_phone_setup_utils import phone_setup_volte |
| from acts_contrib.test_utils.tel.tel_phone_setup_utils import phone_idle_iwlan |
| from acts_contrib.test_utils.tel.tel_phone_setup_utils import phone_idle_volte |
| from acts_contrib.test_utils.tel.tel_subscription_utils import get_subid_from_slot_index |
| from acts_contrib.test_utils.tel.tel_subscription_utils import get_operatorname_from_slot_index |
| from acts_contrib.test_utils.tel.tel_subscription_utils import get_carrierid_from_slot_index |
| from acts_contrib.test_utils.tel.tel_subscription_utils import get_isopportunistic_from_slot_index |
| from acts_contrib.test_utils.tel.tel_subscription_utils import set_subid_for_data |
| from acts_contrib.test_utils.tel.tel_subscription_utils import set_subid_for_message |
| from acts_contrib.test_utils.tel.tel_subscription_utils import set_subid_for_outgoing_call |
| from acts_contrib.test_utils.tel.tel_subscription_utils import set_always_allow_mms_data |
| from acts_contrib.test_utils.tel.tel_test_utils import STORY_LINE |
| 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_device_epoch_time |
| from acts_contrib.test_utils.tel.tel_test_utils import get_telephony_signal_strength |
| from acts_contrib.test_utils.tel.tel_test_utils import synchronize_device_time |
| from acts_contrib.test_utils.tel.tel_test_utils import set_preferred_network_mode_pref |
| from acts_contrib.test_utils.tel.tel_test_utils import verify_internet_connection |
| from acts_contrib.test_utils.tel.tel_test_utils import verify_internet_connection_by_ping |
| from acts_contrib.test_utils.tel.tel_test_utils import verify_http_connection |
| from acts_contrib.test_utils.tel.tel_data_utils import wait_for_data_connection |
| from acts_contrib.test_utils.tel.tel_test_utils import is_current_data_on_cbrs |
| from acts_contrib.test_utils.tel.tel_test_utils import check_voice_network_type |
| from acts_contrib.test_utils.tel.tel_voice_utils import call_setup_teardown |
| from acts_contrib.test_utils.tel.tel_voice_utils import hangup_call |
| from acts_contrib.test_utils.tel.tel_voice_utils import hangup_call_by_adb |
| from acts_contrib.test_utils.tel.tel_voice_utils import initiate_call |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_3g |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_2g |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_csfb |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_iwlan |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_volte |
| from acts_contrib.test_utils.tel.tel_voice_utils import last_call_drop_reason |
| from acts_contrib.test_utils.tel.tel_voice_utils import get_current_voice_rat |
| from acts_contrib.test_utils.tel.tel_voice_utils import wait_for_call_id_clearing |
| from acts_contrib.test_utils.tel.tel_voice_utils import wait_for_in_call_active |
| from acts_contrib.test_utils.tel.tel_wifi_utils import ensure_wifi_connected |
| from acts.utils import get_current_epoch_time |
| from acts.utils import rand_ascii_str |
| from acts.libs.utils.multithread import run_multithread_func |
| |
| EXCEPTION_TOLERANCE = 5 |
| BINDER_LOGS = ["/sys/kernel/debug/binder"] |
| DEFAULT_FILE_DOWNLOADS = ["1MB", "5MB", "10MB", "20MB", "50MB"] |
| RESULTS_LIST = {-2: "UNAVAILABLE_NETWORK_TYPE", |
| -1: "CALL_SETUP_FAILURE", |
| 0: "SUCCESS", |
| 1: "INITIATE_FAILED", |
| 2: "NO_RING_EVENT_OR_ANSWER_FAILED", |
| 3: "NO_CALL_ID_FOUND", |
| 4: "CALL_STATE_NOT_ACTIVE_DURING_ESTABLISHMENT", |
| 5: "AUDIO_STATE_NOT_INCALL_DURING_ESTABLISHMENT", |
| 6: "AUDIO_STATE_NOT_INCALL_AFTER_CONNECTED", |
| 7: "CALL_DROP_OR_WRONG_STATE_DURING_ESTABLISHMENT", |
| 8: "CALL_DROP_OR_WRONG_STATE_AFTER_CONNECTED", |
| 9: "CALL_HANGUP_FAIL", |
| 10: "CALL_ID_CLEANUP_FAIL"} |
| voice_call_failure_dict = {} |
| |
| |
| class TelLiveStressTest(TelephonyBaseTest): |
| def setup_class(self): |
| super(TelLiveStressTest, self).setup_class() |
| self.dut = self.android_devices[0] |
| self.single_phone_test = self.user_params.get("single_phone_test", |
| False) |
| # supported file download methods: chrome, sl4a, curl |
| self.file_download_method = self.user_params.get( |
| "file_download_method", "curl") |
| self.get_binder_logs = self.user_params.get("get_binder_logs", False) |
| if len(self.android_devices) == 1: |
| self.single_phone_test = True |
| if self.single_phone_test: |
| self.android_devices = self.android_devices[:1] |
| self.call_server_number = self.user_params.get( |
| "call_server_number", STORY_LINE) |
| if self.file_download_method == "sl4a": |
| # with single device, do not use sl4a file download |
| # due to stability issue |
| self.file_download_method = "curl" |
| else: |
| self.android_devices = self.android_devices[:2] |
| self.sdm_log = self.user_params.get("sdm_log", False) |
| for ad in self.android_devices: |
| setattr(ad, "sdm_log", self.sdm_log) |
| ad.adb.shell("setprop nfc.debug_enable 1") |
| if self.user_params.get("turn_on_tcpdump", False): |
| start_adb_tcpdump(ad, interface="any", mask="all") |
| self.user_params["telephony_auto_rerun"] = 0 |
| self.phone_call_iteration = int( |
| self.user_params.get("phone_call_iteration", 500)) |
| self.max_phone_call_duration = int( |
| self.user_params.get("max_phone_call_duration", 600)) |
| self.min_sleep_time = int(self.user_params.get("min_sleep_time", 30)) |
| self.max_sleep_time = int(self.user_params.get("max_sleep_time", 60)) |
| self.max_run_time = int(self.user_params.get("max_run_time", 14400)) |
| self.max_sms_length = int(self.user_params.get("max_sms_length", 1000)) |
| self.max_mms_length = int(self.user_params.get("max_mms_length", 160)) |
| self.min_sms_length = int(self.user_params.get("min_sms_length", 1)) |
| self.min_mms_length = int(self.user_params.get("min_mms_length", 1)) |
| self.min_phone_call_duration = int( |
| self.user_params.get("min_phone_call_duration", 10)) |
| self.crash_check_interval = int( |
| self.user_params.get("crash_check_interval", 300)) |
| self.cbrs_check_interval = int( |
| self.user_params.get("cbrs_check_interval", 100)) |
| self.dut_incall = False |
| self.wfc_nw_gen = self.user_params.get("wfc_nw_gen", None) |
| self.dsds_esim = self.user_params.get("dsds_esim", False) |
| self.cbrs_esim = self.user_params.get("cbrs_esim", False) |
| telephony_info = getattr(self.dut, "telephony", {}) |
| self.dut_capabilities = telephony_info.get("capabilities", []) |
| self.dut_wfc_modes = telephony_info.get("wfc_modes", []) |
| self.gps_log_file = self.user_params.get("gps_log_file", None) |
| self.file_name_list = self.user_params.get("file_downloads", DEFAULT_FILE_DOWNLOADS) |
| self.tel_logger = TelephonyMetricLogger.for_test_case() |
| self.tel_logger = TelephonyStressMetricLogger.for_test_case() |
| self.result_collection = {"UNAVAILABLE_NETWORK_TYPE" : 0, |
| "CALL_SETUP_FAILURE" : 0, |
| "SUCCESS" : 0, |
| "INITIATE_FAILED" : 0, |
| "NO_RING_EVENT_OR_ANSWER_FAILED" : 0, |
| "NO_CALL_ID_FOUND" : 0, |
| "CALL_STATE_NOT_ACTIVE_DURING_ESTABLISHMENT" : 0, |
| "AUDIO_STATE_NOT_INCALL_DURING_ESTABLISHMENT" : 0, |
| "AUDIO_STATE_NOT_INCALL_AFTER_CONNECTED" : 0, |
| "CALL_DROP_OR_WRONG_STATE_DURING_ESTABLISHMENT" : 0, |
| "CALL_DROP_OR_WRONG_STATE_AFTER_CONNECTED": 0, |
| "CALL_HANGUP_FAIL": 0, |
| "CALL_ID_CLEANUP_FAIL": 0 } |
| self.call_stats_check = self.user_params.get("call_stats_check", False) |
| self.nsa_5g_for_stress = self.user_params.get("nsa_5g_for_stress", False) |
| self.nr_type = self.user_params.get("nr_type", 'nsa') |
| self.idle_period = self.user_params.get("idle_period", False) |
| self.min_pause_duration = int(self.user_params.get("min_pause_duration", 180)) |
| self.max_pause_duration = int(self.user_params.get("max_pause_duration", 600)) |
| self.call_pause_intervals = int(self.user_params.get("call_pause_intervals", 10)) |
| self.pause=False |
| return True |
| |
| def setup_test(self): |
| super(TelLiveStressTest, self).setup_test() |
| self.result_info = collections.defaultdict(int) |
| self._init_perf_json() |
| self.internet_connection_check_method = verify_internet_connection |
| |
| def on_fail(self, test_name, begin_time): |
| pass |
| |
| def _take_bug_report(self, test_name, begin_time): |
| if self._skip_bug_report(test_name): |
| return |
| src_dir = context.get_current_context().get_full_output_path() |
| dst_dir = os.path.join(self.log_path, test_name) |
| |
| # Extract test_run_info.txt, test_run_debug.txt |
| for file_name in ("test_run_info.txt", "test_run_debug.txt"): |
| extract_test_log(self.log, os.path.join(src_dir, file_name), |
| os.path.join(dst_dir, |
| "%s_%s" % (test_name, file_name)), |
| "\[Test Case\] %s " % test_name) |
| super()._take_bug_report(test_name, begin_time) |
| |
| def _ad_take_extra_logs(self, ad, test_name, begin_time): |
| src_file = os.path.join(ad.device_log_path, |
| 'adblog_%s_debug.txt' % ad.serial) |
| dst_file = os.path.join(ad.device_log_path, test_name, |
| "%s_%s.logcat" % (ad.serial, test_name)) |
| extract_test_log(self.log, src_file, dst_file, test_name) |
| return super()._ad_take_extra_logs(ad, test_name, begin_time) |
| |
| def _setup_wfc(self): |
| for ad in self.android_devices: |
| if not ensure_wifi_connected( |
| self.log, |
| ad, |
| self.wifi_network_ssid, |
| self.wifi_network_pass, |
| retries=3): |
| ad.log.error("Bringing up Wifi connection fails.") |
| return False |
| ad.log.info("Phone WIFI is connected successfully.") |
| if not set_wfc_mode(self.log, ad, WFC_MODE_WIFI_PREFERRED): |
| ad.log.error("Phone failed to enable Wifi-Calling.") |
| return False |
| ad.log.info("Phone is set in Wifi-Calling successfully.") |
| if not phone_idle_iwlan(self.log, ad): |
| ad.log.error("Phone is not in WFC enabled state.") |
| return False |
| ad.log.info("Phone is in WFC enabled state.") |
| return True |
| |
| def _setup_wfc_apm(self): |
| for ad in self.android_devices: |
| if not phone_setup_iwlan( |
| self.log, ad, True, WFC_MODE_CELLULAR_PREFERRED, |
| self.wifi_network_ssid, self.wifi_network_pass, |
| nw_gen=self.wfc_nw_gen, nr_type= self.nr_type): |
| ad.log.error("Failed to setup WFC.") |
| return False |
| return True |
| |
| def _setup_lte_volte_enabled(self): |
| for ad in self.android_devices: |
| if not phone_setup_volte(self.log, ad): |
| ad.log.error("Phone failed to enable VoLTE.") |
| return False |
| ad.log.info("Phone VOLTE is enabled successfully.") |
| # TODO: b/186865335 Move 5G methods to NR directory |
| if self.nsa_5g_for_stress: |
| if not provision_device_for_5g( |
| self.log, ad, nr_type=self.nr_type): |
| ad.log.error("Phone failed to attach 5G NSA.") |
| return False |
| ad.log.info("Phone 5G NSA VOLTE is enabled successfully.") |
| return True |
| |
| def _setup_lte_volte_disabled(self): |
| for ad in self.android_devices: |
| if not phone_setup_csfb(self.log, ad): |
| ad.log.error("Phone failed to setup CSFB.") |
| return False |
| ad.log.info("Phone VOLTE is disabled successfully.") |
| return True |
| |
| def _setup_3g(self): |
| for ad in self.android_devices: |
| if not phone_setup_voice_3g(self.log, ad): |
| ad.log.error("Phone failed to setup 3g.") |
| return False |
| ad.log.info("Phone RAT 3G is enabled successfully.") |
| return True |
| |
| def _setup_2g(self): |
| for ad in self.android_devices: |
| if not phone_setup_voice_2g(self.log, ad): |
| ad.log.error("Phone failed to setup 2g.") |
| return False |
| ad.log.info("RAT 2G is enabled successfully.") |
| return True |
| |
| def _get_network_rat(self, slot_id): |
| rat = self.dut.adb.getprop("gsm.network.type") |
| if "," in rat: |
| if self.dsds_esim: |
| rat = rat.split(',')[slot_id] |
| else: |
| (rat1, rat2) = rat.split(',') |
| if rat1 == "Unknown": |
| rat = rat2 |
| else: |
| rat = rat1 |
| return rat |
| |
| def _send_message(self, max_wait_time=2 * MAX_WAIT_TIME_SMS_RECEIVE): |
| slot_id_rx = None |
| if self.pause and self.idle_period: |
| self.log.info("PAUSE MESSAGE TEST FOR %s seconds", self.pause_duration) |
| time.sleep(self.pause_duration) |
| if self.single_phone_test: |
| ads = [self.dut, self.dut] |
| else: |
| ads = self.android_devices[:] |
| random.shuffle(ads) |
| slot_id = random.randint(0,1) |
| if self.dsds_esim: |
| sub_id = get_subid_from_slot_index(self.log, ads[0], slot_id) |
| ads[0].log.info("Message - MO - slot_Id %d", slot_id) |
| set_subid_for_message(ads[0], sub_id) |
| time.sleep(WAIT_TIME_CHANGE_MESSAGE_SUB_ID) |
| slot_id_rx = random.randint(0,1) |
| ads[1].log.info("Message - MT - slot_id %d", slot_id_rx) |
| selection = random.randrange(0, 2) |
| message_type_map = {0: "SMS", 1: "MMS"} |
| max_length_map = {0: self.max_sms_length, 1: self.max_mms_length} |
| min_length_map = {0: self.min_sms_length, 1: self.min_mms_length} |
| length = random.randrange(min_length_map[selection], |
| max_length_map[selection] + 1) |
| message_func_map = { |
| 0: sms_send_receive_verify, |
| 1: mms_send_receive_verify |
| } |
| rat = self._get_network_rat(slot_id) |
| self.dut.log.info("Network in RAT %s", rat) |
| if self.dut_incall and not is_rat_svd_capable(rat.upper()): |
| self.dut.log.info("In call data not supported, test SMS only") |
| selection = 0 |
| message_type = message_type_map[selection] |
| the_number = self.result_info["%s Total" % message_type] + 1 |
| begin_time = get_device_epoch_time(self.dut) |
| test_name = "%s_No_%s_%s" % (self.test_name, the_number, message_type) |
| if self.sdm_log: |
| start_sdm_loggers(self.log, self.android_devices) |
| else: |
| start_qxdm_loggers(self.log, self.android_devices) |
| log_msg = "[Test Case] %s" % test_name |
| self.log.info("%s begin", log_msg) |
| for ad in self.android_devices: |
| if self.user_params.get("turn_on_tcpdump", False): |
| start_adb_tcpdump(ad, interface="any", mask="all") |
| if not getattr(ad, "messaging_droid", None): |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| else: |
| try: |
| if not ad.messaging_droid.is_live: |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| else: |
| ad.messaging_ed.clear_all_events() |
| except Exception: |
| ad.log.info("Create new sl4a session for messaging") |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| ad.messaging_droid.logI("[BEGIN]%s" % log_msg) |
| |
| text = "%s:" % test_name |
| text_length = len(text) |
| if length < text_length: |
| text = text[:length] |
| else: |
| text += rand_ascii_str(length - text_length) |
| message_content_map = {0: [text], 1: [(test_name, text, None)]} |
| |
| result = message_func_map[selection](self.log, ads[0], ads[1], |
| message_content_map[selection], |
| max_wait_time, |
| slot_id_rx=slot_id_rx) |
| self.log.info("%s end", log_msg) |
| for ad in self.android_devices: |
| ad.messaging_droid.logI("[END]%s" % log_msg) |
| if not result: |
| self.result_info["%s Total" % message_type] += 1 |
| if message_type == "SMS": |
| self.log.error("%s fails", log_msg) |
| self.result_info["%s Failure" % message_type] += 1 |
| else: |
| rat = self._get_network_rat(slot_id) |
| self.dut.log.info("Network in RAT %s", rat) |
| if self.dut_incall and not is_rat_svd_capable(rat.upper()): |
| self.dut.log.info( |
| "In call data not supported, MMS failure expected") |
| self.result_info["Expected In-call MMS failure"] += 1 |
| return True |
| else: |
| self.log.error("%s fails", log_msg) |
| self.result_info["MMS Failure"] += 1 |
| try: |
| self._take_bug_report(test_name, begin_time) |
| except Exception as e: |
| self.log.exception(e) |
| return False |
| else: |
| self.result_info["%s Total" % message_type] += 1 |
| self.log.info("%s succeed", log_msg) |
| self.result_info["%s Success" % message_type] += 1 |
| return True |
| |
| def _make_phone_call(self, call_verification_func=None, voice_stress_only = False): |
| ads = self.android_devices[:] |
| slot_id_callee = None |
| if not voice_stress_only: |
| if not self.single_phone_test: |
| random.shuffle(ads) |
| if self.dsds_esim: |
| slot_id = random.randint(0,1) |
| sub_id = get_subid_from_slot_index(self.log, ads[0], slot_id) |
| ads[0].log.info("Voice - MO - slot_Id %d", slot_id) |
| set_subid_for_outgoing_call(ads[0], sub_id) |
| time.sleep(WAIT_TIME_CHANGE_VOICE_SUB_ID) |
| slot_id_callee = random.randint(0,1) |
| ads[1].log.info("Voice - MT - slot_id %d", slot_id_callee) |
| the_number = self.result_info["Call Total"] + 1 |
| if voice_stress_only: |
| duration = 30 |
| else: |
| duration = random.randrange(self.min_phone_call_duration, |
| self.max_phone_call_duration) |
| result = True |
| test_name = "%s_No_%s_phone_call" % (self.test_name, the_number) |
| log_msg = "[Test Case] %s" % test_name |
| self.log.info("%s for %s seconds begin", log_msg, duration) |
| |
| if self.idle_period: |
| call_iteration = self.call_pause_intervals if self.call_pause_intervals != 0 else 1 |
| if the_number % call_iteration == 0: |
| self.pause=True |
| self.pause_duration = random.randrange( |
| self.min_pause_duration, self.max_pause_duration) |
| self.log.info("PAUSE CALLING TEST FOR %s seconds", self.pause_duration) |
| time.sleep(self.pause_duration) |
| self.pause=False |
| |
| if self.call_stats_check: |
| voice_type_init = check_voice_network_type(ads, voice_init=True) |
| else: |
| voice_type_init = None |
| |
| begin_time = get_device_epoch_time(ads[0]) |
| for ad in self.android_devices: |
| if self.user_params.get("turn_on_tcpdump", False): |
| start_adb_tcpdump(ad, interface="any", mask="all") |
| if not getattr(ad, "droid", None): |
| ad.droid, ad.ed = ad.get_droid() |
| ad.ed.start() |
| else: |
| try: |
| if not ad.droid.is_live: |
| ad.droid, ad.ed = ad.get_droid() |
| ad.ed.start() |
| else: |
| ad.ed.clear_all_events() |
| except Exception: |
| ad.log.info("Create new sl4a session for phone call") |
| ad.droid, ad.ed = ad.get_droid() |
| ad.ed.start() |
| ad.droid.logI("[BEGIN]%s" % log_msg) |
| if self.sdm_log: |
| for ad in ads: |
| ad.adb.shell("i2cset -fy 3 64 6 1 b", ignore_status=True) |
| ad.adb.shell("i2cset -fy 3 65 6 1 b", ignore_status=True) |
| start_sdm_loggers(self.log, self.android_devices) |
| else: |
| start_qxdm_loggers(self.log, self.android_devices) |
| if self.cbrs_esim: |
| self._cbrs_data_check_test(begin_time, expected_cbrs=True, |
| test_time="before") |
| failure_reasons = set() |
| self.dut_incall = True |
| if self.single_phone_test: |
| call_setup_result = initiate_call( |
| self.log, |
| self.dut, |
| self.call_server_number, |
| incall_ui_display=INCALL_UI_DISPLAY_BACKGROUND |
| ) and wait_for_in_call_active(self.dut, 60, 3) |
| else: |
| call_setup_result = call_setup_teardown( |
| self.log, |
| ads[0], |
| ads[1], |
| ad_hangup=None, |
| verify_caller_func=call_verification_func, |
| verify_callee_func=call_verification_func, |
| wait_time_in_call=0, |
| incall_ui_display=INCALL_UI_DISPLAY_BACKGROUND, |
| slot_id_callee=slot_id_callee, |
| call_stats_check=self.call_stats_check, |
| voice_type_init=voice_type_init, |
| result_info = self.result_info) |
| |
| if not call_setup_result: |
| get_telephony_signal_strength(ads[0]) |
| if not self.single_phone_test: |
| get_telephony_signal_strength(ads[1]) |
| call_logs = ads[0].search_logcat( |
| "ActivityManager: START u0 {act=android.intent.action.CALL", |
| begin_time) |
| messaging_logs = ads[0].search_logcat( |
| "com.google.android.apps.messaging/.ui.conversation.ConversationActivity", |
| begin_time) |
| if call_logs and messaging_logs: |
| if (messaging_logs[-1]["datetime_obj"] - |
| call_logs[-1]["datetime_obj"]).seconds < 5: |
| ads[0].log.info( |
| "Call setup failure due to simultaneous activities") |
| self.result_info[ |
| "Call Setup Failure With Simultaneous Activity"] += 1 |
| return True |
| self.log.error("%s: Setup Call failed.", log_msg) |
| failure_reasons.add("Setup") |
| if self.call_stats_check: |
| network = ads[0].droid.telephonyGetCurrentVoiceNetworkType() |
| ads[0].log.debug("Call Setup failure RAT is %s", network) |
| self.result_info["Call Failures"] = self._update_call_failure(str(ads[0].serial), |
| "Call Setup Failure", |
| network) |
| result = False |
| else: |
| elapsed_time = 0 |
| check_interval = 5 |
| if self.sdm_log: |
| for ad in ads: |
| ad.adb.shell("i2cset -fy 3 64 6 1 b", ignore_status=True) |
| ad.adb.shell("i2cset -fy 3 65 6 1 b", ignore_status=True) |
| if self.cbrs_esim: |
| time.sleep(5) |
| self._cbrs_data_check_test(begin_time, expected_cbrs=False, |
| test_time="during") |
| while (elapsed_time < duration): |
| check_interval = min(check_interval, duration - elapsed_time) |
| time.sleep(check_interval) |
| elapsed_time += check_interval |
| time_message = "at <%s>/<%s> second." % (elapsed_time, |
| duration) |
| for ad in ads: |
| get_telephony_signal_strength(ad) |
| if not call_verification_func(self.log, ad): |
| ad.log.warning("Call is NOT in correct %s state at %s", |
| call_verification_func.__name__, |
| time_message) |
| if call_verification_func.__name__ == "is_phone_in_call_iwlan": |
| if is_phone_in_call(self.log, ad): |
| if getattr(ad, "data_rat_state_error_count", |
| 0) < 1: |
| setattr(ad, "data_rat_state_error_count", |
| 1) |
| continue |
| failure_reasons.add("Maintenance") |
| if self.call_stats_check: |
| network = ad.droid.telephonyGetCurrentVoiceNetworkType() |
| ad.log.debug("Call Maintenance failure RAT is %s", network) |
| self.result_info["Call Failures"] = self._update_call_failure( |
| str(ad.serial), |
| "Call Maintenance Failure", |
| network) |
| last_call_drop_reason(ad, begin_time) |
| hangup_call(self.log, ads[0]) |
| result = False |
| else: |
| ad.log.info("Call is in correct %s state at %s", |
| call_verification_func.__name__, |
| time_message) |
| if not result: |
| break |
| if not hangup_call(self.log, ads[0]): |
| failure_reasons.add("Teardown") |
| result = False |
| else: |
| if self.nsa_5g_for_stress: |
| for ad in (ads[0], ads[1]): |
| if not is_current_network_5g(ad, nr_type=self.nr_type): |
| ad.log.error("Phone not attached on 5G") |
| for ad in ads: |
| if not wait_for_call_id_clearing(ad, |
| []) or ad.droid.telecomIsInCall(): |
| ad.log.error("Fail to hang up call") |
| failure_reasons.add("Teardown") |
| result = False |
| self.result_info["Call Total"] += 1 |
| for ad in self.android_devices: |
| try: |
| ad.droid.logI("[END]%s" % log_msg) |
| except: |
| pass |
| self.log.info("%s end", log_msg) |
| self.dut_incall = False |
| if self.cbrs_esim: |
| time.sleep(30) |
| self._cbrs_data_check_test(begin_time, expected_cbrs=True, |
| test_time="after") |
| if not result: |
| self.log.info("%s failed", log_msg) |
| if self.gps_log_file: |
| gps_info = job.run( |
| "tail %s" % self.gps_log_file, ignore_status=True) |
| if gps_info.stdout: |
| gps_log_path = os.path.join(self.log_path, test_name) |
| os.makedirs(gps_log_path, exist_ok=True) |
| job.run( |
| "tail %s > %s" % |
| (self.gps_log_file, |
| os.path.join(gps_log_path, "gps_logs.txt")), |
| ignore_status=True) |
| self.log.info("gps log:\n%s", gps_info.stdout) |
| else: |
| self.log.warning("Fail to get gps log %s", |
| self.user_params["gps_log_file"]) |
| for reason in failure_reasons: |
| self.result_info["Call %s Failure" % reason] += 1 |
| if self.get_binder_logs: |
| for ad in ads: |
| log_path = os.path.join(self.log_path, test_name, |
| "%s_binder_logs" % ad.serial) |
| os.makedirs(log_path, exist_ok=True) |
| ad.pull_files(BINDER_LOGS, log_path) |
| try: |
| self._take_bug_report(test_name, begin_time) |
| except Exception as e: |
| self.log.exception(e) |
| for ad in ads: |
| if ad.droid.telecomIsInCall(): |
| hangup_call_by_adb(ad) |
| else: |
| self.log.info("%s test succeed", log_msg) |
| self.result_info["Call Success"] += 1 |
| if self.result_info["Call Total"] % 50 == 0: |
| for ad in ads: |
| synchronize_device_time(ad) |
| force_connectivity_metrics_upload(ad) |
| if self.get_binder_logs: |
| log_path = os.path.join(self.log_path, test_name, |
| "%s_binder_logs" % ad.serial) |
| os.makedirs(log_path, exist_ok=True) |
| ad.pull_files(BINDER_LOGS, log_path) |
| return result |
| |
| def _prefnetwork_mode_change(self, sub_id): |
| # ModePref change to non-LTE |
| begin_time = get_device_epoch_time(self.dut) |
| if self.sdm_log: |
| start_sdm_loggers(self.log, self.android_devices) |
| else: |
| start_qxdm_loggers(self.log, self.android_devices) |
| self.result_info["Network Change Request Total"] += 1 |
| test_name = "%s_network_change_iter_%s" % ( |
| self.test_name, self.result_info["Network Change Request Total"]) |
| log_msg = "[Test Case] %s" % test_name |
| self.log.info("%s begin", log_msg) |
| self.dut.droid.logI("[BEGIN]%s" % log_msg) |
| network_preference_list = [ |
| NETWORK_MODE_TDSCDMA_GSM_WCDMA, NETWORK_MODE_WCDMA_ONLY, |
| NETWORK_MODE_GLOBAL, NETWORK_MODE_CDMA, NETWORK_MODE_GSM_ONLY |
| ] |
| network_preference = random.choice(network_preference_list) |
| set_preferred_network_mode_pref(self.log, self.dut, sub_id, |
| network_preference) |
| time.sleep(WAIT_TIME_AFTER_MODE_CHANGE) |
| self.dut.log.info("Current Voice RAT is %s", |
| get_current_voice_rat(self.log, self.dut)) |
| |
| # ModePref change back to with LTE |
| if not phone_setup_volte(self.log, self.dut): |
| self.dut.log.error("Phone failed to enable VoLTE.") |
| self.result_info["VoLTE Setup Failure"] += 1 |
| self.dut.droid.logI("%s end" % log_msg) |
| self.dut.log.info("[END]%s", log_msg) |
| try: |
| self._ad_take_extra_logs(self.dut, test_name, begin_time) |
| self._ad_take_bugreport(self.dut, test_name, begin_time) |
| except Exception as e: |
| self.log.exception(e) |
| return False |
| else: |
| self.result_info["VoLTE Setup Success"] += 1 |
| return True |
| |
| def _mobile_data_toggling(self, setup="volte"): |
| # ModePref change to non-LTE |
| begin_time = get_device_epoch_time(self.dut) |
| if self.sdm_log: |
| start_sdm_loggers(self.log, self.android_devices) |
| else: |
| start_qxdm_loggers(self.log, self.android_devices) |
| result = True |
| self.result_info["Data Toggling Request Total"] += 1 |
| test_name = "%s_data_toggling_iter_%s" % ( |
| self.test_name, self.result_info["Data Toggling Request Total"]) |
| log_msg = "[Test Case] %s" % test_name |
| self.log.info("%s begin", log_msg) |
| self.dut.droid.logI("[BEGIN]%s" % log_msg) |
| self.dut.adb.shell("svc data disable") |
| time.sleep(WAIT_TIME_AFTER_MODE_CHANGE) |
| self.dut.adb.shell("svc data enable") |
| if not self._check_data(): |
| result = False |
| elif setup == "volte" and not phone_idle_volte(self.log, self.dut): |
| result = False |
| self.dut.droid.logI("%s end" % log_msg) |
| self.dut.log.info("[END]%s", log_msg) |
| if not result: |
| self.result_info["Data Toggling Failure"] += 1 |
| try: |
| self._ad_take_extra_logs(self.dut, test_name, begin_time) |
| self._ad_take_bugreport(self.dut, test_name, begin_time) |
| except Exception as e: |
| self.log.exception(e) |
| return False |
| else: |
| self.result_info["Data Toggling Success"] += 1 |
| return True |
| |
| def _get_result_message(self): |
| msg_list = [ |
| "%s: %s" % (count, self.result_info[count]) |
| for count in sorted(self.result_info.keys()) |
| ] |
| return ", ".join(msg_list) |
| |
| def _write_perf_json(self): |
| json_str = json.dumps(self.perf_data, indent=4, sort_keys=True) |
| with open(self.perf_file, 'w') as f: |
| f.write(json_str) |
| |
| def _init_perf_json(self): |
| self.perf_file = os.path.join(self.log_path, "%s_perf_data_%s.json" % |
| (self.test_name, self.begin_time)) |
| self.perf_data = self.dut.build_info.copy() |
| self.perf_data["build_fingerprint"] = self.dut.adb.getprop( |
| "ro.build.fingerprint") |
| self.perf_data["model"] = self.dut.model |
| self.perf_data["carrier"] = self.dut.adb.getprop( |
| "gsm.sim.operator.alpha") |
| self._write_perf_json() |
| |
| def _update_perf_json(self): |
| for result_key, result_value in self.result_info.items(): |
| self.perf_data[result_key] = result_value |
| self._write_perf_json() |
| |
| def crash_check_test(self): |
| failure = 0 |
| while time.time() < self.finishing_time: |
| try: |
| self.log.info(dict(self.result_info)) |
| self._update_perf_json() |
| begin_time = get_device_epoch_time(self.dut) |
| run_time_in_seconds = (begin_time - self.begin_time) / 1000 |
| test_name = "%s_crash_%s_seconds_after_start" % ( |
| self.test_name, run_time_in_seconds) |
| time.sleep(self.crash_check_interval) |
| for ad in self.android_devices: |
| crash_report = ad.check_crash_report( |
| test_name, begin_time, log_crash_report=True) |
| if crash_report: |
| ad.log.error("Find new crash reports %s", crash_report) |
| failure += 1 |
| self.result_info["Crashes"] += len(crash_report) |
| for crash in crash_report: |
| if "ramdump_modem" in crash: |
| self.result_info["Crashes-Modem"] += 1 |
| try: |
| ad.take_bug_report(test_name, begin_time) |
| except Exception as e: |
| self.log.exception(e) |
| except Exception as e: |
| self.log.error("Exception error %s", str(e)) |
| self.result_info["Exception Errors"] += 1 |
| self.log.info("Crashes found: %s", failure) |
| if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: |
| self.log.error("Too many exception errors, quit test") |
| return False |
| if failure: |
| return False |
| else: |
| return True |
| |
| def _update_call_failure(self, dut, key, network): |
| if dut not in voice_call_failure_dict.keys(): |
| voice_call_failure_dict[dut] = {key:{network:0}} |
| if key not in voice_call_failure_dict[dut].keys(): |
| voice_call_failure_dict[dut].update({key:{network:0}}) |
| if network not in voice_call_failure_dict[dut][key].keys(): |
| voice_call_failure_dict[dut][key].update({network:0}) |
| voice_call_failure_dict[dut][key][network] += 1 |
| return voice_call_failure_dict |
| |
| def _cbrs_data_check_test(self, begin_time, expected_cbrs=True, |
| test_time="before"): |
| cbrs_fail_count = 0 |
| the_number = self.result_info["CBRS Total"] + 1 |
| test_name = "%s_cbrs_%s_call_No_%s" % (self.test_name, |
| test_time, the_number) |
| for ad in self.android_devices: |
| current_state = is_current_data_on_cbrs(ad, ad.cbrs) |
| if current_state == expected_cbrs: |
| self.result_info["CBRS-Check-Pass"] += 1 |
| else: |
| self.result_info["CBRS-Check-Fail"] += 1 |
| cbrs_fail_count += 1 |
| try: |
| self._ad_take_extra_logs(ad, test_name, begin_time) |
| self._ad_take_bugreport(ad, test_name, begin_time) |
| except Exception as e: |
| self.log.warning(e) |
| if cbrs_fail_count > 0: |
| ad.log.error("Found %d checks failed, expected cbrs %s", |
| cbrs_fail_count, expected_cbrs) |
| cbrs_fail_count += 1 |
| self.result_info["CBRS Total"] += 1 |
| return True |
| |
| def call_test(self, call_verification_func=None): |
| while time.time() < self.finishing_time: |
| time.sleep( |
| random.randrange(self.min_sleep_time, self.max_sleep_time)) |
| try: |
| self._make_phone_call(call_verification_func) |
| except Exception as e: |
| self.log.exception("Exception error %s", str(e)) |
| self.result_info["Exception Errors"] += 1 |
| if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: |
| self.log.error("Too many exception errors, quit test") |
| return False |
| self.log.info("%s", dict(self.result_info)) |
| if any([ |
| self.result_info["Call Setup Failure"], |
| self.result_info["Call Maintenance Failure"], |
| self.result_info["Call Teardown Failure"] |
| ]): |
| return False |
| else: |
| return True |
| |
| def message_test(self, max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE): |
| while time.time() < self.finishing_time: |
| try: |
| self._send_message(max_wait_time=max_wait_time) |
| except Exception as e: |
| self.log.exception("Exception error %s", str(e)) |
| self.result_info["Exception Errors"] += 1 |
| self.log.info(dict(self.result_info)) |
| if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: |
| self.log.error("Too many exception errors, quit test") |
| return False |
| time.sleep( |
| random.randrange(self.min_sleep_time, self.max_sleep_time)) |
| if self.result_info["SMS Failure"] or ( |
| self.result_info["MMS Failure"] / self.result_info["MMS Total"] |
| > 0.3): |
| return False |
| else: |
| return True |
| |
| def _data_download(self, file_names=[]): |
| begin_time = get_current_epoch_time() |
| slot_id = random.randint(0,1) |
| if self.pause and self.idle_period: |
| self.log.info("PAUSE DATA TEST FOR %s seconds", self.pause_duration) |
| time.sleep(self.pause_duration) |
| if self.dsds_esim: |
| sub_id = get_subid_from_slot_index(self.log, self.dut, slot_id) |
| self.dut.log.info("Data - slot_Id %d", slot_id) |
| set_subid_for_data(self.dut, sub_id) |
| self.dut.droid.telephonyToggleDataConnection(True) |
| if self.sdm_log: |
| start_sdm_loggers(self.log, self.android_devices) |
| else: |
| start_qxdm_loggers(self.log, self.android_devices) |
| self.dut.log.info(dict(self.result_info)) |
| selection = random.randrange(0, len(file_names)) |
| file_name = file_names[selection] |
| self.result_info["Internet Connection Check Total"] += 1 |
| |
| rat = self._get_network_rat(slot_id) |
| if not self.internet_connection_check_method(self.log, self.dut): |
| self.dut.log.info("Network in RAT %s", rat) |
| if self.dut_incall and not is_rat_svd_capable(rat.upper()): |
| self.result_info[ |
| "Expected Incall Internet Connection Check Failure"] += 1 |
| return True |
| else: |
| self.result_info["Internet Connection Check Failure"] += 1 |
| test_name = "%s_internet_connection_No_%s_failure" % ( |
| self.test_name, |
| self.result_info["Internet Connection Check Failure"]) |
| try: |
| self._ad_take_extra_logs(self.dut, test_name, begin_time) |
| self._ad_take_bugreport(self.dut, test_name, begin_time) |
| except Exception as e: |
| self.log.exception(e) |
| return False |
| else: |
| self.result_info["Internet Connection Check Success"] += 1 |
| |
| self.result_info["File Download Total"] += 1 |
| self.result_info["Data Download Total"] += int(file_name[:-2]) |
| if not active_file_download_test( |
| self.log, self.dut, file_name, |
| method=self.file_download_method): |
| self.result_info["File Download Failure"] += 1 |
| self.result_info["Data Failure Total"] += int(file_name[:-2]) |
| if self.result_info["File Download Failure"] == 1: |
| try: |
| self._ad_take_extra_logs( |
| self.dut, "%s_file_download_failure" % self.test_name, |
| begin_time) |
| self._ad_take_bugreport( |
| self.dut, "%s_file_download_failure" % self.test_name, |
| begin_time) |
| except Exception as e: |
| self.log.exception(e) |
| return False |
| else: |
| self.result_info["File Download Success"] += 1 |
| self.result_info["Data Success Total"] += int(file_name[:-2]) |
| return True |
| |
| def data_test(self): |
| while time.time() < self.finishing_time: |
| try: |
| self._data_download(self.file_name_list) |
| except Exception as e: |
| self.log.error("Exception error %s", str(e)) |
| self.result_info["Exception Errors"] += 1 |
| self.log.info("%s", dict(self.result_info)) |
| if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: |
| self.log.error("Too many exception errors, quit test") |
| return False |
| time.sleep( |
| random.randrange(self.min_sleep_time, self.max_sleep_time)) |
| if self.result_info["Internet Connection Check Failure"]: |
| return False |
| else: |
| return True |
| |
| def _check_data(self): |
| self.result_info["Data Connection Check Total"] += 1 |
| if not wait_for_data_connection(self.log, self.dut, True): |
| self.result_info["Data Connection Setup Failure"] += 1 |
| return False |
| if not self.internet_connection_check_method(self.log, self.dut): |
| rat = self.dut.adb.getprop("gsm.network.type") |
| self.dut.log.info("Network in RAT %s", rat) |
| self.result_info["Internet Connection Check Failure"] += 1 |
| return False |
| return True |
| |
| def _data_call_test(self, sub_id, generation): |
| self.dut.log.info(dict(self.result_info)) |
| begin_time = get_device_epoch_time(self.dut) |
| if self.sdm_log: |
| start_sdm_loggers(self.log, self.android_devices) |
| else: |
| start_qxdm_loggers(self.log, self.android_devices) |
| self.result_info["Network Change Request Total"] += 1 |
| test_name = "%s_network_change_test_iter_%s" % ( |
| self.test_name, self.result_info["Network Change Request Total"]) |
| log_msg = "[Test Case] %s" % test_name |
| self.log.info("%s begin", log_msg) |
| self.dut.droid.logI("[BEGIN]%s" % log_msg) |
| if not ensure_network_generation_for_subscription( |
| self.log, self.dut, sub_id, |
| generation) or not self._check_data(): |
| self.result_info["Network Change Failure"] += 1 |
| self.dut.droid.logI("%s end" % log_msg) |
| self.dut.log.info("[END]%s", log_msg) |
| try: |
| self._ad_take_extra_logs(self.dut, test_name, begin_time) |
| self._ad_take_bugreport(self.dut, test_name, begin_time) |
| except Exception as e: |
| self.log.warning(e) |
| return False |
| if not self._mobile_data_toggling(setup=None): |
| return False |
| return True |
| |
| def data_call_stress_test(self): |
| result = True |
| sub_id = self.dut.droid.subscriptionGetDefaultSubId() |
| while time.time() < self.finishing_time: |
| for generation in (GEN_4G, GEN_3G): |
| try: |
| if not self._data_call_test(sub_id, generation): |
| result = False |
| except Exception as e: |
| self.log.error("Exception error %s", str(e)) |
| self.result_info["Exception Errors"] += 1 |
| if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: |
| self.log.error("Too many exception errors, quit test") |
| return False |
| return result |
| |
| def check_incall_data(self): |
| if verify_internet_connection_by_ping(self.log, self.dut): |
| self.internet_connection_check_method = verify_internet_connection_by_ping |
| elif verify_http_connection(self.log, self.dut): |
| self.internet_connection_check_method = verify_http_connection |
| else: |
| self.dut.log.error("Data test failed") |
| raise signals.TestFailure("Data check failed") |
| if self.single_phone_test: |
| if not initiate_call( |
| self.log, self.dut, |
| self.call_server_number) and wait_for_in_call_active( |
| self.dut, 60, 3): |
| self._take_bug_report(self.test_name, self.begin_time) |
| raise signals.TestFailure("Unable to make phone call") |
| else: |
| if not call_setup_teardown( |
| self.log, self.dut, self.android_devices[1], |
| ad_hangup=None): |
| self._take_bug_report(self.test_name, self.begin_time) |
| raise signals.TestFailure("Unable to make phone call") |
| voice_rat = self.dut.droid.telephonyGetCurrentVoiceNetworkType() |
| data_rat = self.dut.droid.telephonyGetCurrentDataNetworkType() |
| self.dut.log.info("Voice in RAT %s, Data in RAT %s", voice_rat, |
| data_rat) |
| try: |
| if "wfc" in self.test_name or is_rat_svd_capable( |
| voice_rat.upper()) and is_rat_svd_capable( |
| data_rat.upper()): |
| self.dut.log.info("Capable for simultaneous voice and data") |
| |
| if not self.internet_connection_check_method( |
| self.log, self.dut): |
| self.dut.log.error("Incall data check failed") |
| raise signals.TestFailure("Incall data check failed") |
| else: |
| return True |
| else: |
| self.dut.log.info( |
| "Not capable for simultaneous voice and data") |
| return False |
| hangup_call(self.log, self.dut) |
| finally: |
| for ad in self.android_devices: |
| if ad.droid.telecomIsInCall(): |
| hangup_call(self.log, ad) |
| |
| def parallel_tests(self, setup_func=None, call_verification_func=None): |
| self.log.info(self._get_result_message()) |
| if setup_func and not setup_func(): |
| msg = "%s setup %s failed" % (self.test_name, setup_func.__name__) |
| self.log.error(msg) |
| self._take_bug_report("%s%s" % (self.test_name, |
| setup_func.__name__), |
| self.begin_time) |
| return False |
| if not call_verification_func: |
| call_verification_func = is_phone_in_call |
| self.finishing_time = time.time() + self.max_run_time |
| # CBRS setup |
| if self.cbrs_esim: |
| cbrs_sub_count = 0 |
| for ad in self.android_devices: |
| if not getattr(ad, 'cbrs', {}): |
| setattr(ad, 'cbrs', None) |
| for i in range(0, 2): |
| sub_id = get_subid_from_slot_index(ad.log, ad, i) |
| operator = get_operatorname_from_slot_index(ad, i) |
| carrier_id = get_carrierid_from_slot_index(ad, i) |
| is_opportunistic = get_isopportunistic_from_slot_index(ad, i) |
| ad.log.info("Slot %d - Sub %s - %s - %d", i, sub_id, operator, carrier_id) |
| if carrier_id == GOOGLE_CBRS_CARRIER_ID or (carrier_id == GOOGLE_FI_CARRIER_ID and is_opportunistic): |
| ad.cbrs = sub_id |
| cbrs_sub_count += 1 |
| if cbrs_sub_count != 2: |
| self.log.error("Expecting - 2 CBRS subs, found - %d", cbrs_sub_count) |
| raise signals.TestAbortClass("Cannot find all expected CBRS subs") |
| # DSDS setup |
| if self.dsds_esim: |
| for ad in self.android_devices: |
| for i in range(0, 2): |
| sub_id = get_subid_from_slot_index(ad.log, ad, i) |
| set_always_allow_mms_data(ad, sub_id) |
| operator = get_operatorname_from_slot_index(ad, i) |
| ad.log.info("Slot %d - Sub %s - %s", i, sub_id, operator) |
| # Actual test trigger |
| if not self.dsds_esim and self.check_incall_data(): |
| self.log.info( |
| "==== Start parallel voice/message/data stress test ====") |
| self.perf_data["testing method"] = "parallel" |
| results = run_multithread_func( |
| self.log, [(self.call_test, [call_verification_func]), |
| (self.message_test, []), (self.data_test, []), |
| (self.crash_check_test, [])]) |
| else: |
| self.log.info( |
| "==== Start sequential voice/message/data stress test ====") |
| self.perf_data["testing method"] = "sequential" |
| results = run_multithread_func( |
| self.log, [(self.sequential_tests, [call_verification_func]), |
| (self.crash_check_test, [])]) |
| result_message = self._get_result_message() |
| self.log.info(result_message) |
| self._update_perf_json() |
| self.result_detail = result_message |
| return all(results) |
| |
| def sequential_tests(self, call_verification_func): |
| funcs = [(self._make_phone_call, [call_verification_func]), |
| (self._send_message, []), (self._data_download, [["5MB"]])] |
| while time.time() < self.finishing_time: |
| selection = random.randrange(0, 3) |
| try: |
| funcs[selection][0](*funcs[selection][1]) |
| except Exception as e: |
| self.log.error("Exception error %s", str(e)) |
| self.result_info["Exception Errors"] += 1 |
| self.log.info("%s", dict(self.result_info)) |
| if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: |
| self.log.error("Too many exception errors, quit test") |
| return False |
| time.sleep( |
| random.randrange(self.min_sleep_time, self.max_sleep_time)) |
| if any([ |
| self.result_info["Call Setup Failure"], |
| self.result_info["Call Maintenance Failure"], |
| self.result_info["Call Teardown Failure"], |
| self.result_info["SMS Failure"], |
| self.result_info["MMS Failure"], |
| self.result_info["Internet Connection Check Failure"] |
| ]): |
| return False |
| return True |
| |
| def volte_modechange_volte_test(self): |
| sub_id = self.dut.droid.subscriptionGetDefaultSubId() |
| result = True |
| while time.time() < self.finishing_time: |
| try: |
| if self._prefnetwork_mode_change(sub_id): |
| run_multithread_func( |
| self.log, |
| [(self._data_download, [["5MB"]]), |
| (self._make_phone_call, [is_phone_in_call_volte]), |
| (self._send_message, [])]) |
| else: |
| result = False |
| if self._mobile_data_toggling(): |
| run_multithread_func( |
| self.log, |
| [(self._data_download, [["5MB"]]), |
| (self._make_phone_call, [is_phone_in_call_volte]), |
| (self._send_message, [])]) |
| else: |
| result = False |
| except Exception as e: |
| self.log.error("Exception error %s", str(e)) |
| self.result_info["Exception Errors"] += 1 |
| self.log.info(dict(self.result_info)) |
| if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: |
| self.log.error("Too many exception errors, quit test") |
| return False |
| return result |
| |
| def parallel_with_network_change_tests(self, setup_func=None): |
| if setup_func and not setup_func(): |
| self.log.error("Test setup %s failed", setup_func.__name__) |
| return False |
| self.finishing_time = time.time() + self.max_run_time |
| results = run_multithread_func(self.log, |
| [(self.volte_modechange_volte_test, []), |
| (self.crash_check_test, [])]) |
| result_message = self._get_result_message() |
| self.log.info(result_message) |
| self._update_perf_json() |
| self.result_detail = result_message |
| return all(results) |
| |
| def connect_to_wifi(self): |
| for ad in self.android_devices: |
| if not ensure_wifi_connected( |
| self.log, |
| ad, |
| self.wifi_network_ssid, |
| self.wifi_network_pass, |
| retries=3): |
| ad.log.error("Bringing up Wifi connection fails.") |
| return False |
| ad.log.info("Phone WIFI is connected successfully.") |
| return True |
| |
| def performance_tests(self, setup_func=None, call_verification_func=None): |
| self.log.info(self._get_result_message()) |
| if setup_func and not setup_func(): |
| msg = "%s setup %s failed" % (self.test_name, setup_func.__name__) |
| self.log.error(msg) |
| self._take_bug_report("%s%s" % (self.test_name, |
| setup_func.__name__), |
| self.begin_time) |
| return False |
| if not call_verification_func: |
| call_verification_func = is_phone_in_call |
| self.finishing_time = time.time() + self.max_run_time |
| |
| self.log.info( |
| "==== Start voice stress test ====") |
| self.perf_data["testing method"] = "parallel" |
| results = self.call_performance_test(call_verification_func) |
| |
| result_message = self._get_result_message() |
| self.log.info(result_message) |
| self._update_perf_json() |
| self.result_detail = result_message |
| total_call = self.result_info["Call Total"] |
| success_call = self.result_info["Call Success Total"] |
| call_fail = total_call - success_call |
| if call_fail != 0: |
| call_fail_rate = ( call_fail / total_call ) * 100 |
| else: |
| call_fail_rate = 0 |
| call_success_rate = (success_call / total_call) * 100 |
| |
| self.log.info("Call Success Rate is %s", call_success_rate) |
| self.log.info("Call Drop Rate is %s", call_success_rate) |
| |
| return results |
| |
| def _update_initiate_call_fail_count(self): |
| self.result_info["Call Initiate Fail"] += 1 |
| |
| def call_performance_test(self, call_verification_func=None): |
| count = 0 |
| while count < self.phone_call_iteration: |
| time.sleep(15) |
| count += 1 |
| try: |
| self._make_phone_call(call_verification_func, True) |
| except Exception as e: |
| self.log.exception("Exception error %s", str(e)) |
| self.result_info["Exception Errors"] += 1 |
| if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE: |
| self.log.error("Too many exception errors, quit test") |
| return False |
| self.log.info("%s", dict(self.result_info)) |
| self.result_info["Call Success Total"] += 1 |
| if any([ |
| self.result_info["Call Setup Failure"], |
| self.result_info["Call Maintenance Failure"], |
| self.result_info["Call Teardown Failure"] |
| ]): |
| return False |
| else: |
| return True |
| |
| """ Tests Begin """ |
| |
| @test_tracker_info(uuid="d035e5b9-476a-4e3d-b4e9-6fd86c51a68d") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_default_parallel_stress(self): |
| """ Default state stress test""" |
| return self.parallel_tests() |
| |
| @test_tracker_info(uuid="798a3c34-db75-4bcf-b8ef-e1116414a7fe") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_default_parallel_stress_with_wifi(self): |
| """ Default state stress test with Wifi enabled.""" |
| if self.connect_to_wifi(): |
| return self.parallel_tests() |
| |
| @test_tracker_info(uuid="c21e1f17-3282-4f0b-b527-19f048798098") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_lte_volte_parallel_stress(self): |
| """ VoLTE on stress test""" |
| return self.parallel_tests( |
| setup_func=self._setup_lte_volte_enabled, |
| call_verification_func=is_phone_in_call_volte) |
| |
| @test_tracker_info(uuid="a317c23a-41e0-4ef8-af67-661451cfefcf") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_csfb_parallel_stress(self): |
| """ LTE non-VoLTE stress test""" |
| return self.parallel_tests( |
| setup_func=self._setup_lte_volte_disabled, |
| call_verification_func=is_phone_in_call_csfb) |
| |
| @test_tracker_info(uuid="fdb791bf-c414-4333-9fa3-cc18c9b3b234") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_wfc_parallel_stress(self): |
| """ Wifi calling APM mode off stress test""" |
| if WFC_MODE_WIFI_PREFERRED not in self.dut_wfc_modes: |
| raise signals.TestSkip("WFC_MODE_WIFI_PREFERRED is not supported") |
| return self.parallel_tests( |
| setup_func=self._setup_wfc, |
| call_verification_func=is_phone_in_call_iwlan) |
| |
| @test_tracker_info(uuid="e334c1b3-4378-49bb-bf57-1573fa1b23fa") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_wfc_apm_parallel_stress(self): |
| """ Wifi calling in APM mode on stress test""" |
| return self.parallel_tests( |
| setup_func=self._setup_wfc_apm, |
| call_verification_func=is_phone_in_call_iwlan) |
| |
| @test_tracker_info(uuid="4566eef6-55de-4ac8-87ee-58f2ef41a3e8") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_3g_parallel_stress(self): |
| """ 3G stress test""" |
| return self.parallel_tests( |
| setup_func=self._setup_3g, |
| call_verification_func=is_phone_in_call_3g) |
| |
| @test_tracker_info(uuid="f34f1a31-3948-4675-8698-372a83b8088d") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_2g_parallel_stress(self): |
| """ 2G call stress test""" |
| return self.parallel_tests( |
| setup_func=self._setup_2g, |
| call_verification_func=is_phone_in_call_2g) |
| |
| @test_tracker_info(uuid="af580fca-fea6-4ca5-b981-b8c710302d37") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_volte_modeprefchange_parallel_stress(self): |
| """ VoLTE Mode Pref call stress test""" |
| return self.parallel_with_network_change_tests( |
| setup_func=self._setup_lte_volte_enabled) |
| |
| @test_tracker_info(uuid="10e34247-5fd3-4f87-81bf-3c17a6b71ab2") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_data_call_stress(self): |
| """ Default state stress test""" |
| self.finishing_time = time.time() + self.max_run_time |
| results = run_multithread_func(self.log, |
| [(self.data_call_stress_test, []), |
| (self.crash_check_test, [])]) |
| result_message = self._get_result_message() |
| self.log.info(result_message) |
| self._update_perf_json() |
| self.result_detail = result_message |
| return all(results) |
| |
| @test_tracker_info(uuid="4212d0e0-fb87-47e5-ba48-9df9a4a6bb9b") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_voice_performance_stress(self): |
| """ Voice Performance stress test""" |
| return self.performance_tests() |
| |
| @test_tracker_info(uuid="a126793e-6e78-4920-a8c4-b382a444c4b7") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_voice_performance_stress_nr(self): |
| """ Voice Performance stress test""" |
| return self.performance_tests( |
| setup_func=self._setup_lte_volte_enabled) |
| |
| """ Tests End """ |
