#!/usr/bin/env python3.4
#
# Copyright 2017 - 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 utils
from acts.asserts import fail
from acts.test_decorators import test_tracker_info
from acts.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
from acts.test_utils.tel.tel_defines import MAX_WAIT_TIME_SMS_RECEIVE
from acts.test_utils.tel.tel_defines import NETWORK_MODE_WCDMA_ONLY
from acts.test_utils.tel.tel_defines import NETWORK_MODE_GLOBAL
from acts.test_utils.tel.tel_defines import NETWORK_MODE_CDMA
from acts.test_utils.tel.tel_defines import NETWORK_MODE_GSM_ONLY
from acts.test_utils.tel.tel_defines import NETWORK_MODE_TDSCDMA_GSM_WCDMA
from acts.test_utils.tel.tel_defines import NETWORK_MODE_LTE_CDMA_EVDO_GSM_WCDMA
from acts.test_utils.tel.tel_defines import WAIT_TIME_AFTER_MODE_CHANGE
from acts.test_utils.tel.tel_defines import WFC_MODE_WIFI_PREFERRED
from acts.test_utils.tel.tel_test_utils import STORY_LINE
from acts.test_utils.tel.tel_test_utils import active_file_download_test
from acts.test_utils.tel.tel_test_utils import is_phone_in_call
from acts.test_utils.tel.tel_test_utils import call_setup_teardown
from acts.test_utils.tel.tel_test_utils import ensure_wifi_connected
from acts.test_utils.tel.tel_test_utils import hangup_call
from acts.test_utils.tel.tel_test_utils import hangup_call_by_adb
from acts.test_utils.tel.tel_test_utils import initiate_call
from acts.test_utils.tel.tel_test_utils import run_multithread_func
from acts.test_utils.tel.tel_test_utils import set_wfc_mode
from acts.test_utils.tel.tel_test_utils import sms_send_receive_verify
from acts.test_utils.tel.tel_test_utils import start_adb_tcpdump
from acts.test_utils.tel.tel_test_utils import stop_adb_tcpdump
from acts.test_utils.tel.tel_test_utils import start_qxdm_loggers
from acts.test_utils.tel.tel_test_utils import mms_send_receive_verify
from acts.test_utils.tel.tel_test_utils import set_preferred_network_mode_pref
from acts.test_utils.tel.tel_test_utils import wait_for_in_call_active
from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_3g
from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_2g
from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_csfb
from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_iwlan
from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_volte
from acts.test_utils.tel.tel_voice_utils import phone_setup_csfb
from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_3g
from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_2g
from acts.test_utils.tel.tel_voice_utils import phone_setup_volte
from acts.test_utils.tel.tel_voice_utils import phone_idle_iwlan
from acts.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
EXCEPTION_TOLERANCE = 5
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", "sl4a")
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 = "chrome"
else:
self.android_devices = self.android_devices[:2]
self.user_params["telephony_auto_rerun"] = False
self.wifi_network_ssid = self.user_params.get(
"wifi_network_ssid") or self.user_params.get(
"wifi_network_ssid_2g")
self.wifi_network_pass = self.user_params.get(
"wifi_network_pass") or self.user_params.get(
"wifi_network_pass_2g")
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", 10))
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", 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))
return True
def setup_test(self):
super(TelLiveStressTest, self).setup_test()
self.result_info = collections.defaultdict(int)
self._init_perf_json()
def on_fail(self, test_name, begin_time):
pass
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("Phone 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_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.")
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 _send_message(self, max_wait_time=2 * MAX_WAIT_TIME_SMS_RECEIVE):
if self.single_phone_test:
ads = [self.dut, self.dut]
else:
ads = self.android_devices[:]
random.shuffle(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)
message_func_map = {
0: sms_send_receive_verify,
1: mms_send_receive_verify
}
message_type = message_type_map[selection]
the_number = self.result_info["%s Total" % message_type] + 1
begin_time = get_current_epoch_time()
start_qxdm_loggers(self.log, self.android_devices)
log_msg = "The %s-th %s test: of length %s from %s to %s" % (
the_number, message_type, length, ads[0].serial, ads[1].serial)
self.log.info(log_msg)
for ad in self.android_devices:
for session in ad._sl4a_manager.sessions.values():
try:
session.rpc_client.logI(log_msg)
break
except Exception as e:
ad.log.warning(e)
text = "%s: " % log_msg
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: [(log_msg, text, None)]}
incall_non_ims = False
for ad in self.android_devices:
if ad.droid.telecomIsInCall() and (
not ad.droid.telephonyIsImsRegistered()):
incall_non_ims = True
break
if not message_func_map[selection](self.log, ads[0], ads[1],
message_content_map[selection],
max_wait_time):
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
self._take_bug_report("%s_%s_No_%s_failure" %
(self.test_name, message_type,
the_number), begin_time)
else:
if incall_non_ims:
self.log.info(
"Device not in IMS, MMS in call is not support")
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
if self.result_info["MMS Failure"] == 1:
self._take_bug_report("%s_%s_No_%s_failure" %
(self.test_name, message_type,
the_number), begin_time)
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):
ads = self.android_devices[:]
if not self.single_phone_test:
random.shuffle(ads)
for ad in ads:
hangup_call_by_adb(ad)
the_number = self.result_info["Call Total"] + 1
duration = random.randrange(self.min_phone_call_duration,
self.max_phone_call_duration)
result = True
if self.single_phone_test:
log_msg = "The %s-th phone call test for %ssec duration" % (
the_number, duration)
else:
log_msg = "The %s-th phone call test from %s to %s for %ssec" % (
the_number, ads[0].serial, ads[1].serial, duration)
self.log.info(log_msg)
for ad in ads:
try:
ad.droid.logI(log_msg)
except Exception as e:
ad.log.warning(e)
begin_time = get_current_epoch_time()
start_qxdm_loggers(self.log, self.android_devices, begin_time)
if self.single_phone_test:
call_setup_result = initiate_call(
self.log,
self.dut,
self.call_server_number,
wait_time_betwn_call_initcheck=5) 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, wait_time_in_call=0)
if not call_setup_result:
self.log.error("%s: Setup Call failed.", log_msg)
self.result_info["Call Setup Failure"] += 1
failure_reason = "setup"
result = False
else:
elapsed_time = 0
check_interval = 5
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:
if not call_verification_func(self.log, ad):
ad.log.error("Call is NOT in correct %s state at %s",
call_verification_func.__name__,
time_message)
self.result_info["Call Maintenance Failure"] += 1
failure_reason = "maintenance"
reasons = ad.search_logcat(
"qcril_qmi_voice_map_qmi_to_ril_last_call_failure_cause",
begin_time)
if reasons:
ad.log.info(reasons[-1]["log_message"])
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]):
time.sleep(10)
for ad in ads:
if ad.droid.telecomIsInCall():
ad.log.error("Still in call after hungup")
self.result_info["Call Teardown Failure"] += 1
failure_reason = "teardown"
result = False
self.result_info["Call Total"] += 1
if not result:
self.log.info("%s test failed", log_msg)
test_name = "%s_call_No_%s_%s_failure" % (self.test_name,
the_number,
failure_reason)
for ad in ads:
log_path = os.path.join(self.log_path, test_name,
"%s_binder" % ad.serial)
utils.create_dir(log_path)
ad.adb.pull("/sys/kernel/debug/binder %s" % log_path)
self._take_bug_report(test_name, begin_time)
else:
self.log.info("%s test succeed", log_msg)
self.result_info["Call Success"] += 1
if self.result_info["Call Total"] % 50 == 0:
test_name = "%s_call_No_%s_success_binder_logs" % (
self.test_name, the_number)
for ad in ads:
log_path = os.path.join(self.log_path, test_name,
"%s_binder" % ad.serial)
utils.create_dir(log_path)
ad.adb.pull("/sys/kernel/debug/binder %s" % log_path)
return result
def _prefnetwork_mode_change(self, sub_id):
# ModePref change to non-LTE
begin_time = get_current_epoch_time()
start_qxdm_loggers(self.log, self.android_devices)
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
set_preferred_network_mode_pref(self.log, self.dut, sub_id,
NETWORK_MODE_LTE_CDMA_EVDO_GSM_WCDMA)
time.sleep(WAIT_TIME_AFTER_MODE_CHANGE)
rat = get_current_voice_rat(self.log, self.dut)
self.dut.log.info("Current Voice RAT is %s", rat)
self.result_info["RAT Change Total"] += 1
if rat != "LTE":
self.result_info["RAT Change Failure"] += 1
self._take_bug_report("%s_rat_change_failure" % self.test_name,
begin_time)
return False
else:
self.result_info["RAT Change 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.android_devices[0].build_info.copy()
self.perf_data["model"] = self.android_devices[0].model
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_current_epoch_time()
time.sleep(self.crash_check_interval)
for ad in self.android_devices:
crash_report = ad.check_crash_report(
"checking_crash", 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"] += 1
for crash in crash_report:
if "ramdump_modem" in crash:
self.result_info["Crashes-Modem"] += 1
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 call_test(self, call_verification_func=None):
while time.time() < self.finishing_time:
try:
self._make_phone_call(call_verification_func)
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
self.log.info("%s", dict(self.result_info))
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"]
]):
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.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
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 = ["5MB", "10MB", "20MB", "50MB", "200MB", "512MB", "1GB"]
file_names = ["5MB", "10MB", "20MB", "50MB", "200MB"]
begin_time = get_current_epoch_time()
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["File Download Total"] += 1
if not active_file_download_test(
self.log, self.dut, file_name,
method=self.file_download_method):
self.result_info["File Download Failure"] += 1
if self.result_info["File Download Failure"] == 1:
self._take_bug_report(
"%s_file_download_failure" % self.test_name, begin_time)
return False
else:
self.result_info["File Download Success"] += 1
return True
def data_test(self):
while time.time() < self.finishing_time:
try:
self._data_download()
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["File Download Failure"] / self.result_info["File Download Total"] > 0.1:
return False
else:
return True
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 = "Test setup %s failed" % setup_func.__name__
self.log.error(msg)
fail(msg)
if not call_verification_func:
call_verification_func = is_phone_in_call
self.finishing_time = time.time() + self.max_run_time
results = run_multithread_func(
self.log, [(self.call_test, [call_verification_func]),
(self.message_test, []), (self.data_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 volte_modechange_volte_test(self):
sub_id = self.dut.droid.subscriptionGetDefaultSubId()
while time.time() < self.finishing_time:
try:
run_multithread_func(
self.log,
[(self._data_download, []),
(self._make_phone_call, [is_phone_in_call_volte]),
(self._send_message, [])])
self._prefnetwork_mode_change(sub_id)
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
if self.result_info["Call Failure"] or self.result_info["RAT Change Failure"] or self.result_info["SMS Failure"]:
return False
else:
return True
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)
""" 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="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 on stress test"""
return self.parallel_tests(
setup_func=self._setup_wfc,
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_call_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)
""" Tests End """