#!/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 Mobility Stress Test """ import collections import random import time from acts.asserts import explicit_pass from acts.asserts import fail from acts.test_decorators import test_tracker_info from acts_contrib.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest from acts_contrib.test_utils.tel.tel_atten_utils import set_rssi from acts_contrib.test_utils.tel.tel_defines import CELL_WEAK_RSSI_VALUE from acts_contrib.test_utils.tel.tel_defines import MAX_RSSI_RESERVED_VALUE from acts_contrib.test_utils.tel.tel_defines import MIN_RSSI_RESERVED_VALUE from acts_contrib.test_utils.tel.tel_defines import WFC_MODE_CELLULAR_PREFERRED from acts_contrib.test_utils.tel.tel_defines import WFC_MODE_DISABLED from acts_contrib.test_utils.tel.tel_defines import WFC_MODE_WIFI_PREFERRED from acts_contrib.test_utils.tel.tel_defines import WFC_MODE_CELLULAR_PREFERRED from acts_contrib.test_utils.tel.tel_logging_utils import start_qxdm_loggers 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_data_utils import active_file_download_test from acts_contrib.test_utils.tel.tel_voice_utils import call_setup_teardown from acts_contrib.test_utils.tel.tel_voice_utils import get_current_voice_rat from acts.utils import get_current_epoch_time from acts.utils import rand_ascii_str from acts.libs.utils.multithread import run_multithread_func from TelWifiVoiceTest import TelWifiVoiceTest from TelWifiVoiceTest import ATTEN_NAME_FOR_WIFI_2G from TelWifiVoiceTest import ATTEN_NAME_FOR_WIFI_5G from TelWifiVoiceTest import ATTEN_NAME_FOR_CELL_3G from TelWifiVoiceTest import ATTEN_NAME_FOR_CELL_4G from acts.controllers.sl4a_lib.rpc_client import Sl4aProtocolError IGNORE_EXCEPTIONS = (BrokenPipeError, Sl4aProtocolError) EXCEPTION_TOLERANCE = 20 class TelLiveMobilityStressTest(TelWifiVoiceTest): def setup_class(self): super().setup_class() #super(TelWifiVoiceTest, self).setup_class() self.user_params["telephony_auto_rerun"] = False self.max_phone_call_duration = int( self.user_params.get("max_phone_call_duration", 600)) self.max_sleep_time = int(self.user_params.get("max_sleep_time", 120)) self.max_run_time = int(self.user_params.get("max_run_time", 7200)) 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.crash_check_interval = int( self.user_params.get("crash_check_interval", 300)) self.signal_change_interval = int( self.user_params.get("signal_change_interval", 10)) self.signal_change_step = int( self.user_params.get("signal_change_step", 5)) 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.dut = self.android_devices[0] self.helper = self.android_devices[1] return True def on_fail(self, test_name, begin_time): pass def _setup_volte_wfc_wifi_preferred(self): return self._wfc_phone_setup( False, WFC_MODE_WIFI_PREFERRED, volte_mode=True) def _setup_volte_wfc_cell_preferred(self): return self._wfc_phone_setup( False, WFC_MODE_CELLULAR_PREFERRED, volte_mode=True) def _setup_csfb_wfc_wifi_preferred(self): return self._wfc_phone_setup( False, WFC_MODE_WIFI_PREFERRED, volte_mode=False) def _setup_csfb_wfc_cell_preferred(self): return self._wfc_phone_setup( False, WFC_MODE_CELLULAR_PREFERRED, volte_mode=False) def _setup_volte_wfc_disabled(self): return self._wfc_phone_setup(False, WFC_MODE_DISABLED, volte_mode=True) def _setup_csfb_wfc_disabled(self): return self._wfc_phone_setup( False, WFC_MODE_DISABLED, volte_mode=False) def _send_message(self, ads): 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) text = rand_ascii_str(length) message_content_map = {0: [text], 1: [("Mms Message", text, None)]} message_func_map = { 0: sms_send_receive_verify, 1: mms_send_receive_verify } message_type = message_type_map[selection] self.result_info["Total %s" % message_type] += 1 begin_time = get_current_epoch_time() start_qxdm_loggers(self.log, self.android_devices) if not message_func_map[selection](self.log, ads[0], ads[1], message_content_map[selection]): self.log.error("%s of length %s from %s to %s fails", message_type, length, ads[0].serial, ads[1].serial) self.result_info["%s failure" % message_type] += 1 if message_type == "SMS" or self.result_info["%s failure" % message_type] == 1: self._take_bug_report("%s_%s_failure" % (self.test_name, message_type), begin_time) return False else: self.log.info("%s of length %s from %s to %s succeed", message_type_map[selection], length, ads[0].serial, ads[1].serial) return True def _make_phone_call(self, ads): self.result_info["Total Calls"] += 1 begin_time = get_current_epoch_time() start_qxdm_loggers(self.log, self.android_devices) if not call_setup_teardown( self.log, ads[0], ads[1], ad_hangup=ads[random.randrange(0, 2)], wait_time_in_call=random.randrange( self.min_phone_call_duration, self.max_phone_call_duration)): self.log.error("Call setup and teardown failed.") self.result_info["Call Failure"] += 1 self._take_bug_report("%s_call_failure" % self.test_name, begin_time) return False self.log.info("Call setup and teardown succeed.") return True def crash_check_test(self): failure = 0 while time.time() < self.finishing_time: self.dut.log.info(dict(self.result_info)) try: begin_time = get_current_epoch_time() time.sleep(self.crash_check_interval) crash_report = self.dut.check_crash_report( "checking_crash", begin_time, True) if crash_report: self.dut.log.error("Find new crash reports %s", crash_report) failure += 1 self.result_info["Crashes"] += 1 except IGNORE_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: return False except Exception as e: return False self.dut.log.info("Crashes found: %s", failure) if failure: return False else: return True def environment_change_4g_wifi(self): #block cell 3G, WIFI 2G set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_3G], 0, MIN_RSSI_RESERVED_VALUE) set_rssi(self.log, self.attens[ATTEN_NAME_FOR_WIFI_2G], 0, MIN_RSSI_RESERVED_VALUE) while time.time() < self.finishing_time: #set strong wifi 5G and LTE set_rssi(self.log, self.attens[ATTEN_NAME_FOR_WIFI_5G], 0, MAX_RSSI_RESERVED_VALUE) set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_4G], 0, MAX_RSSI_RESERVED_VALUE) #gratually decrease wifi 5g set_rssi(self.log, self.attens[ATTEN_NAME_FOR_WIFI_5G], self.wifi_rssi_with_no_atten, MIN_RSSI_RESERVED_VALUE, self.signal_change_step, self.signal_change_interval) #gratually increase wifi 5g set_rssi(self.log, self.attens[ATTEN_NAME_FOR_WIFI_5G], MIN_RSSI_RESERVED_VALUE, MAX_RSSI_RESERVED_VALUE, self.signal_change_step, self.signal_change_interval) #gratually decrease cell 4G set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_4G], self.cell_rssi_with_no_atten, CELL_WEAK_RSSI_VALUE, self.signal_change_step, self.signal_change_interval) #gradtually increase cell 4G set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_4G], MIN_RSSI_RESERVED_VALUE, MAX_RSSI_RESERVED_VALUE, self.signal_change_step, self.signal_change_interval) return "" def environment_change_4g_3g(self): #block wifi 2G and 5G set_rssi(self.log, self.attens[ATTEN_NAME_FOR_WIFI_2G], 0, MIN_RSSI_RESERVED_VALUE) set_rssi(self.log, self.attens[ATTEN_NAME_FOR_WIFI_5G], 0, MIN_RSSI_RESERVED_VALUE) while time.time() < self.finishing_time: #set strong cell 4G and 3G set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_4G], 0, MAX_RSSI_RESERVED_VALUE) set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_3G], 0, MAX_RSSI_RESERVED_VALUE) #gratually decrease cell 4G set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_4G], self.cell_rssi_with_no_atten, MIN_RSSI_RESERVED_VALUE, self.signal_change_step, self.signal_change_interval) #gradtually increase cell 4G set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_4G], MIN_RSSI_RESERVED_VALUE, MAX_RSSI_RESERVED_VALUE, self.signal_change_step, self.signal_change_interval) #gratually decrease cell 3G set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_3G], self.cell_rssi_with_no_atten, MIN_RSSI_RESERVED_VALUE, self.signal_change_step, self.signal_change_interval) #gradtually increase cell 3G set_rssi(self.log, self.attens[ATTEN_NAME_FOR_CELL_3G], MIN_RSSI_RESERVED_VALUE, MAX_RSSI_RESERVED_VALUE, self.signal_change_step, self.signal_change_interval) return "" def call_test(self): failure = 0 total_count = 0 while time.time() < self.finishing_time: try: ads = [self.dut, self.helper] random.shuffle(ads) total_count += 1 # Current Voice RAT self.dut.log.info("Current Voice RAT is %s", get_current_voice_rat(self.log, self.dut)) self.helper.log.info("Current Voice RAT is %s", get_current_voice_rat( self.log, self.helper)) if not self._make_phone_call(ads): failure += 1 self.dut.droid.goToSleepNow() time.sleep(random.randrange(0, self.max_sleep_time)) except IGNORE_EXCEPTIONS 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 %s", IGNORE_EXCEPTIONS) return False except Exception as e: self.log.error(e) return False self.dut.log.info("Call test failure: %s/%s", failure, total_count) if failure: return False else: return True def message_test(self): failure = 0 total_count = 0 while time.time() < self.finishing_time: try: ads = [self.dut, self.helper] random.shuffle(ads) total_count += 1 if not self._send_message(ads): failure += 1 self.dut.droid.goToSleepNow() time.sleep(random.randrange(0, self.max_sleep_time)) except IGNORE_EXCEPTIONS 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 %s", IGNORE_EXCEPTIONS) return False except Exception as e: self.log.error(e) return False self.dut.log.info("Messaging test failure: %s/%s", failure, total_count) if failure / total_count > 0.1: return False else: return True def data_test(self): failure = 0 total_count = 0 #file_names = ["5MB", "10MB", "20MB", "50MB", "200MB", "512MB", "1GB"] #wifi download is very slow in lab, limit the file size upto 200MB file_names = ["5MB", "10MB", "20MB", "50MB", "200MB"] while time.time() < self.finishing_time: total_count += 1 begin_time = get_current_epoch_time() start_qxdm_loggers(self.log, self.android_devices) try: self.dut.log.info(dict(self.result_info)) self.result_info["Total file download"] += 1 selection = random.randrange(0, len(file_names)) file_name = file_names[selection] if not active_file_download_test(self.log, self.dut, file_name): self.result_info["File download failure"] += 1 failure += 1 if self.result_info["File download failure"] == 1: self._take_bug_report( "%s_file_download_failure" % self.test_name, begin_time) self.dut.droid.goToSleepNow() time.sleep(random.randrange(0, self.max_sleep_time)) except IGNORE_EXCEPTIONS 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 error %s", IGNORE_EXCEPTIONS) return False except Exception as e: self.log.error(e) return False self.dut.log.info("File download test failure: %s/%s", failure, total_count) if failure / total_count > 0.1: return False else: return True def parallel_tests(self, change_env_func, setup_func=None): if setup_func and not setup_func(): self.log.error("Test setup %s failed", setup_func.__name__) return False self.result_info = collections.defaultdict(int) self.finishing_time = time.time() + self.max_run_time results = run_multithread_func(self.log, [(self.call_test, []), ( self.message_test, []), (self.data_test, []), ( self.crash_check_test, []), (change_env_func, [])]) result_message = "%s" % dict(self.result_info) self.log.info(result_message) if all(results): explicit_pass(result_message) else: fail(result_message) """ Tests Begin """ @test_tracker_info(uuid="6fcba97c-3572-47d7-bcac-9608f1aa5304") @TelephonyBaseTest.tel_test_wrap def test_volte_wfc_wifi_preferred_parallel_stress(self): return self.parallel_tests( self.environment_change_4g_wifi, setup_func=self._setup_volte_wfc_wifi_preferred) @test_tracker_info(uuid="df78a9a8-2a14-40bf-a7aa-719502f975be") @TelephonyBaseTest.tel_test_wrap def test_volte_wfc_cell_preferred_parallel_stress(self): return self.parallel_tests( self.environment_change_4g_wifi, setup_func=self._setup_volte_wfc_cell_preferred) @test_tracker_info(uuid="4cb47315-c420-44c2-ac47-a8bdca6d0e25") @TelephonyBaseTest.tel_test_wrap def test_csfb_wfc_wifi_preferred_parallel_stress(self): return self.parallel_tests( self.environment_change_4g_wifi, setup_func=self._setup_csfb_wfc_wifi_preferred) @test_tracker_info(uuid="92821ef7-542a-4139-b3b0-22278e2b06c4") @TelephonyBaseTest.tel_test_wrap def test_csfb_wfc_cell_preferred_parallel_stress(self): return self.parallel_tests( self.self.environment_change_4g_wifi, setup_func=self._setup_csfb_wfc_cell_preferred) @test_tracker_info(uuid="dd23923e-ebbc-461e-950a-0657e845eacf") @TelephonyBaseTest.tel_test_wrap def test_4g_volte_3g_parallel_stress(self): return self.parallel_tests( self.environment_change_4g_3g, setup_func=self._setup_volte_wfc_disabled) @test_tracker_info(uuid="faef9384-a5b0-4640-8cfa-f9f34ce6d977") @TelephonyBaseTest.tel_test_wrap def test_4g_csfb_3g_parallel_stress(self): return self.parallel_tests( self.environment_change_4g_3g, setup_func=self._setup_csfb_wfc_disabled) """ Tests End """