1#!/usr/bin/env python3.4
2#
3#   Copyright 2017 - 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    Test Script for Telephony Stress Call Test
18"""
19
20import collections
21import json
22import os
23import random
24import time
25
26from acts import context
27from acts import signals
28from acts.libs.proc import job
29from acts.test_decorators import test_tracker_info
30from acts.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
31from acts.test_utils.tel.tel_defines import CAPABILITY_VOLTE
32from acts.test_utils.tel.tel_defines import CAPABILITY_WFC
33from acts.test_utils.tel.tel_defines import GEN_3G
34from acts.test_utils.tel.tel_defines import GEN_4G
35from acts.test_utils.tel.tel_defines import INCALL_UI_DISPLAY_BACKGROUND
36from acts.test_utils.tel.tel_defines import MAX_WAIT_TIME_SMS_RECEIVE
37from acts.test_utils.tel.tel_defines import NETWORK_MODE_WCDMA_ONLY
38from acts.test_utils.tel.tel_defines import NETWORK_MODE_GLOBAL
39from acts.test_utils.tel.tel_defines import NETWORK_MODE_CDMA
40from acts.test_utils.tel.tel_defines import NETWORK_MODE_GSM_ONLY
41from acts.test_utils.tel.tel_defines import NETWORK_MODE_TDSCDMA_GSM_WCDMA
42from acts.test_utils.tel.tel_defines import WAIT_TIME_AFTER_MODE_CHANGE
43from acts.test_utils.tel.tel_defines import WFC_MODE_CELLULAR_PREFERRED
44from acts.test_utils.tel.tel_defines import WFC_MODE_WIFI_PREFERRED
45from acts.test_utils.tel.tel_defines import WAIT_TIME_CHANGE_MESSAGE_SUB_ID
46from acts.test_utils.tel.tel_defines import WAIT_TIME_CHANGE_VOICE_SUB_ID
47from acts.test_utils.tel.tel_defines import WAIT_TIME_FOR_CBRS_DATA_SWITCH
48from acts.test_utils.tel.tel_defines import CARRIER_SING
49from acts.test_utils.tel.tel_lookup_tables import is_rat_svd_capable
50from acts.test_utils.tel.tel_test_utils import STORY_LINE
51from acts.test_utils.tel.tel_test_utils import active_file_download_test
52from acts.test_utils.tel.tel_test_utils import is_phone_in_call
53from acts.test_utils.tel.tel_test_utils import call_setup_teardown
54from acts.test_utils.tel.tel_test_utils import ensure_network_generation_for_subscription
55from acts.test_utils.tel.tel_test_utils import ensure_wifi_connected
56from acts.test_utils.tel.tel_test_utils import extract_test_log
57from acts.test_utils.tel.tel_test_utils import force_connectivity_metrics_upload
58from acts.test_utils.tel.tel_test_utils import get_device_epoch_time
59from acts.test_utils.tel.tel_test_utils import get_telephony_signal_strength
60from acts.test_utils.tel.tel_test_utils import hangup_call
61from acts.test_utils.tel.tel_test_utils import hangup_call_by_adb
62from acts.test_utils.tel.tel_test_utils import initiate_call
63from acts.test_utils.tel.tel_test_utils import last_call_drop_reason
64from acts.test_utils.tel.tel_test_utils import run_multithread_func
65from acts.test_utils.tel.tel_test_utils import set_wfc_mode
66from acts.test_utils.tel.tel_test_utils import sms_send_receive_verify
67from acts.test_utils.tel.tel_test_utils import start_qxdm_loggers
68from acts.test_utils.tel.tel_test_utils import start_sdm_loggers
69from acts.test_utils.tel.tel_test_utils import start_adb_tcpdump
70from acts.test_utils.tel.tel_test_utils import synchronize_device_time
71from acts.test_utils.tel.tel_test_utils import mms_send_receive_verify
72from acts.test_utils.tel.tel_test_utils import set_preferred_network_mode_pref
73from acts.test_utils.tel.tel_test_utils import verify_internet_connection
74from acts.test_utils.tel.tel_test_utils import verify_internet_connection_by_ping
75from acts.test_utils.tel.tel_test_utils import verify_http_connection
76from acts.test_utils.tel.tel_test_utils import wait_for_call_id_clearing
77from acts.test_utils.tel.tel_test_utils import wait_for_data_connection
78from acts.test_utils.tel.tel_test_utils import wait_for_in_call_active
79from acts.test_utils.tel.tel_test_utils import is_current_data_on_cbrs
80from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_3g
81from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_2g
82from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_csfb
83from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_iwlan
84from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_volte
85from acts.test_utils.tel.tel_voice_utils import phone_setup_csfb
86from acts.test_utils.tel.tel_voice_utils import phone_setup_iwlan
87from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_3g
88from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_2g
89from acts.test_utils.tel.tel_voice_utils import phone_setup_volte
90from acts.test_utils.tel.tel_voice_utils import phone_idle_iwlan
91from acts.test_utils.tel.tel_voice_utils import phone_idle_volte
92from acts.test_utils.tel.tel_voice_utils import get_current_voice_rat
93from acts.test_utils.tel.tel_subscription_utils import get_subid_from_slot_index
94from acts.test_utils.tel.tel_subscription_utils import get_operatorname_from_slot_index
95from acts.test_utils.tel.tel_subscription_utils import get_carrierid_from_slot_index
96from acts.test_utils.tel.tel_subscription_utils import get_isopportunistic_from_slot_index
97from acts.test_utils.tel.tel_subscription_utils import set_subid_for_data
98from acts.test_utils.tel.tel_subscription_utils import set_subid_for_message
99from acts.test_utils.tel.tel_subscription_utils import set_subid_for_outgoing_call
100from acts.test_utils.tel.tel_subscription_utils import set_slways_allow_mms_data
101from acts.utils import get_current_epoch_time
102from acts.utils import rand_ascii_str
103
104EXCEPTION_TOLERANCE = 5
105BINDER_LOGS = ["/sys/kernel/debug/binder"]
106DEFAULT_FILE_DOWNLOADS = ["1MB", "5MB", "10MB", "20MB", "50MB"]
107
108
109class TelLiveStressTest(TelephonyBaseTest):
110    def setup_class(self):
111        super(TelLiveStressTest, self).setup_class()
112        self.dut = self.android_devices[0]
113        self.single_phone_test = self.user_params.get("single_phone_test",
114                                                      False)
115        # supported file download methods: chrome, sl4a, curl
116        self.file_download_method = self.user_params.get(
117            "file_download_method", "curl")
118        self.get_binder_logs = self.user_params.get("get_binder_logs", False)
119        if len(self.android_devices) == 1:
120            self.single_phone_test = True
121        if self.single_phone_test:
122            self.android_devices = self.android_devices[:1]
123            self.call_server_number = self.user_params.get(
124                "call_server_number", STORY_LINE)
125            if self.file_download_method == "sl4a":
126                # with single device, do not use sl4a file download
127                # due to stability issue
128                self.file_download_method = "curl"
129        else:
130            self.android_devices = self.android_devices[:2]
131        self.sdm_log = self.user_params.get("sdm_log", False)
132        for ad in self.android_devices:
133            setattr(ad, "sdm_log", self.sdm_log)
134            ad.adb.shell("setprop nfc.debug_enable 1")
135            if self.user_params.get("turn_on_tcpdump", False):
136                start_adb_tcpdump(ad, interface="any", mask="all")
137        self.user_params["telephony_auto_rerun"] = 0
138        self.phone_call_iteration = int(
139            self.user_params.get("phone_call_iteration", 500))
140        self.max_phone_call_duration = int(
141            self.user_params.get("max_phone_call_duration", 600))
142        self.min_sleep_time = int(self.user_params.get("min_sleep_time", 30))
143        self.max_sleep_time = int(self.user_params.get("max_sleep_time", 60))
144        self.max_run_time = int(self.user_params.get("max_run_time", 14400))
145        self.max_sms_length = int(self.user_params.get("max_sms_length", 1000))
146        self.max_mms_length = int(self.user_params.get("max_mms_length", 160))
147        self.min_sms_length = int(self.user_params.get("min_sms_length", 1))
148        self.min_mms_length = int(self.user_params.get("min_mms_length", 1))
149        self.min_phone_call_duration = int(
150            self.user_params.get("min_phone_call_duration", 10))
151        self.crash_check_interval = int(
152            self.user_params.get("crash_check_interval", 300))
153        self.cbrs_check_interval = int(
154            self.user_params.get("cbrs_check_interval", 100))
155        self.dut_incall = False
156        self.dsds_esim = self.user_params.get("dsds_esim", False)
157        self.cbrs_esim = self.user_params.get("cbrs_esim", False)
158        telephony_info = getattr(self.dut, "telephony", {})
159        self.dut_capabilities = telephony_info.get("capabilities", [])
160        self.dut_wfc_modes = telephony_info.get("wfc_modes", [])
161        self.gps_log_file = self.user_params.get("gps_log_file", None)
162        self.file_name_list = self.user_params.get("file_downloads", DEFAULT_FILE_DOWNLOADS)
163        return True
164
165    def setup_test(self):
166        super(TelLiveStressTest, self).setup_test()
167        self.result_info = collections.defaultdict(int)
168        self._init_perf_json()
169        self.internet_connection_check_method = verify_internet_connection
170
171    def on_fail(self, test_name, begin_time):
172        pass
173
174    def _take_bug_report(self, test_name, begin_time):
175        if self._skip_bug_report(test_name):
176            return
177        src_dir = context.get_current_context().get_full_output_path()
178        dst_dir = os.path.join(self.log_path, test_name)
179
180        # Extract test_run_info.txt, test_run_debug.txt
181        for file_name in ("test_run_info.txt", "test_run_debug.txt"):
182            extract_test_log(self.log, os.path.join(src_dir, file_name),
183                             os.path.join(dst_dir,
184                                          "%s_%s" % (test_name, file_name)),
185                             "\[Test Case\] %s " % test_name)
186        super()._take_bug_report(test_name, begin_time)
187
188    def _ad_take_extra_logs(self, ad, test_name, begin_time):
189        src_file = os.path.join(ad.device_log_path,
190                                'adblog_%s_debug.txt' % ad.serial)
191        dst_file = os.path.join(ad.device_log_path, test_name,
192                                "%s_%s.logcat" % (ad.serial, test_name))
193        extract_test_log(self.log, src_file, dst_file, test_name)
194        return super()._ad_take_extra_logs(ad, test_name, begin_time)
195
196    def _setup_wfc(self):
197        for ad in self.android_devices:
198            if not ensure_wifi_connected(
199                    self.log,
200                    ad,
201                    self.wifi_network_ssid,
202                    self.wifi_network_pass,
203                    retries=3):
204                ad.log.error("Bringing up Wifi connection fails.")
205                return False
206            ad.log.info("Phone WIFI is connected successfully.")
207            if not set_wfc_mode(self.log, ad, WFC_MODE_WIFI_PREFERRED):
208                ad.log.error("Phone failed to enable Wifi-Calling.")
209                return False
210            ad.log.info("Phone is set in Wifi-Calling successfully.")
211            if not phone_idle_iwlan(self.log, ad):
212                ad.log.error("Phone is not in WFC enabled state.")
213                return False
214            ad.log.info("Phone is in WFC enabled state.")
215        return True
216
217    def _setup_wfc_apm(self):
218        for ad in self.android_devices:
219            if not phone_setup_iwlan(
220                    self.log, ad, True, WFC_MODE_CELLULAR_PREFERRED,
221                    self.wifi_network_ssid, self.wifi_network_pass):
222                ad.log.error("Failed to setup WFC.")
223                return False
224        return True
225
226    def _setup_lte_volte_enabled(self):
227        for ad in self.android_devices:
228            if not phone_setup_volte(self.log, ad):
229                ad.log.error("Phone failed to enable VoLTE.")
230                return False
231            ad.log.info("Phone VOLTE is enabled successfully.")
232        return True
233
234    def _setup_lte_volte_disabled(self):
235        for ad in self.android_devices:
236            if not phone_setup_csfb(self.log, ad):
237                ad.log.error("Phone failed to setup CSFB.")
238                return False
239            ad.log.info("Phone VOLTE is disabled successfully.")
240        return True
241
242    def _setup_3g(self):
243        for ad in self.android_devices:
244            if not phone_setup_voice_3g(self.log, ad):
245                ad.log.error("Phone failed to setup 3g.")
246                return False
247            ad.log.info("Phone RAT 3G is enabled successfully.")
248        return True
249
250    def _setup_2g(self):
251        for ad in self.android_devices:
252            if not phone_setup_voice_2g(self.log, ad):
253                ad.log.error("Phone failed to setup 2g.")
254                return False
255            ad.log.info("RAT 2G is enabled successfully.")
256        return True
257
258    def _get_network_rat(self, slot_id):
259        rat = self.dut.adb.getprop("gsm.network.type")
260        if "," in rat:
261            if self.dsds_esim:
262                rat = rat.split(',')[slot_id]
263            else:
264                (rat1, rat2) = rat.split(',')
265                if rat1 == "Unknown":
266                    rat = rat2
267                else:
268                    rat = rat1
269        return rat
270
271    def _send_message(self, max_wait_time=2 * MAX_WAIT_TIME_SMS_RECEIVE):
272        slot_id_rx = None
273        if self.single_phone_test:
274            ads = [self.dut, self.dut]
275        else:
276            ads = self.android_devices[:]
277            random.shuffle(ads)
278        slot_id = random.randint(0,1)
279        if self.dsds_esim:
280            sub_id = get_subid_from_slot_index(self.log, ads[0], slot_id)
281            ads[0].log.info("Message - MO - slot_Id %d", slot_id)
282            set_subid_for_message(ads[0], sub_id)
283            time.sleep(WAIT_TIME_CHANGE_MESSAGE_SUB_ID)
284            slot_id_rx = random.randint(0,1)
285            ads[1].log.info("Message - MT - slot_id %d", slot_id_rx)
286        selection = random.randrange(0, 2)
287        message_type_map = {0: "SMS", 1: "MMS"}
288        max_length_map = {0: self.max_sms_length, 1: self.max_mms_length}
289        min_length_map = {0: self.min_sms_length, 1: self.min_mms_length}
290        length = random.randrange(min_length_map[selection],
291                                  max_length_map[selection] + 1)
292        message_func_map = {
293            0: sms_send_receive_verify,
294            1: mms_send_receive_verify
295        }
296        rat = self._get_network_rat(slot_id)
297        self.dut.log.info("Network in RAT %s", rat)
298        if self.dut_incall and not is_rat_svd_capable(rat.upper()):
299            self.dut.log.info("In call data not supported, test SMS only")
300            selection = 0
301        message_type = message_type_map[selection]
302        the_number = self.result_info["%s Total" % message_type] + 1
303        begin_time = get_device_epoch_time(self.dut)
304        test_name = "%s_No_%s_%s" % (self.test_name, the_number, message_type)
305        if self.sdm_log:
306            start_sdm_loggers(self.log, self.android_devices)
307        else:
308            start_qxdm_loggers(self.log, self.android_devices)
309        log_msg = "[Test Case] %s" % test_name
310        self.log.info("%s begin", log_msg)
311        for ad in self.android_devices:
312            if self.user_params.get("turn_on_tcpdump", False):
313                start_adb_tcpdump(ad, interface="any", mask="all")
314            if not getattr(ad, "messaging_droid", None):
315                ad.messaging_droid, ad.messaging_ed = ad.get_droid()
316                ad.messaging_ed.start()
317            else:
318                try:
319                    if not ad.messaging_droid.is_live:
320                        ad.messaging_droid, ad.messaging_ed = ad.get_droid()
321                        ad.messaging_ed.start()
322                    else:
323                        ad.messaging_ed.clear_all_events()
324                except Exception:
325                    ad.log.info("Create new sl4a session for messaging")
326                    ad.messaging_droid, ad.messaging_ed = ad.get_droid()
327                    ad.messaging_ed.start()
328            ad.messaging_droid.logI("[BEGIN]%s" % log_msg)
329
330        text = "%s:" % test_name
331        text_length = len(text)
332        if length < text_length:
333            text = text[:length]
334        else:
335            text += rand_ascii_str(length - text_length)
336        message_content_map = {0: [text], 1: [(test_name, text, None)]}
337
338        result = message_func_map[selection](self.log, ads[0], ads[1],
339                                             message_content_map[selection],
340                                             max_wait_time,
341                                             slot_id_rx=slot_id_rx)
342        self.log.info("%s end", log_msg)
343        for ad in self.android_devices:
344            ad.messaging_droid.logI("[END]%s" % log_msg)
345        if not result:
346            self.result_info["%s Total" % message_type] += 1
347            if message_type == "SMS":
348                self.log.error("%s fails", log_msg)
349                self.result_info["%s Failure" % message_type] += 1
350            else:
351                rat = self._get_network_rat(slot_id)
352                self.dut.log.info("Network in RAT %s", rat)
353                if self.dut_incall and not is_rat_svd_capable(rat.upper()):
354                    self.dut.log.info(
355                        "In call data not supported, MMS failure expected")
356                    self.result_info["Expected In-call MMS failure"] += 1
357                    return True
358                else:
359                    self.log.error("%s fails", log_msg)
360                    self.result_info["MMS Failure"] += 1
361            try:
362                self._take_bug_report(test_name, begin_time)
363            except Exception as e:
364                self.log.exception(e)
365            return False
366        else:
367            self.result_info["%s Total" % message_type] += 1
368            self.log.info("%s succeed", log_msg)
369            self.result_info["%s Success" % message_type] += 1
370            return True
371
372    def _make_phone_call(self, call_verification_func=None):
373        ads = self.android_devices[:]
374        slot_id_callee = None
375        if not self.single_phone_test:
376            random.shuffle(ads)
377        if self.dsds_esim:
378            slot_id = random.randint(0,1)
379            sub_id = get_subid_from_slot_index(self.log, ads[0], slot_id)
380            ads[0].log.info("Voice - MO - slot_Id %d", slot_id)
381            set_subid_for_outgoing_call(ads[0], sub_id)
382            time.sleep(WAIT_TIME_CHANGE_VOICE_SUB_ID)
383            slot_id_callee = random.randint(0,1)
384            ads[1].log.info("Voice - MT - slot_id %d", slot_id_callee)
385        the_number = self.result_info["Call Total"] + 1
386        duration = random.randrange(self.min_phone_call_duration,
387                                    self.max_phone_call_duration)
388        result = True
389        test_name = "%s_No_%s_phone_call" % (self.test_name, the_number)
390        log_msg = "[Test Case] %s" % test_name
391        self.log.info("%s for %s seconds begin", log_msg, duration)
392        begin_time = get_device_epoch_time(ads[0])
393        for ad in self.android_devices:
394            if self.user_params.get("turn_on_tcpdump", False):
395                start_adb_tcpdump(ad, interface="any", mask="all")
396            if not getattr(ad, "droid", None):
397                ad.droid, ad.ed = ad.get_droid()
398                ad.ed.start()
399            else:
400                try:
401                    if not ad.droid.is_live:
402                        ad.droid, ad.ed = ad.get_droid()
403                        ad.ed.start()
404                    else:
405                        ad.ed.clear_all_events()
406                except Exception:
407                    ad.log.info("Create new sl4a session for phone call")
408                    ad.droid, ad.ed = ad.get_droid()
409                    ad.ed.start()
410            ad.droid.logI("[BEGIN]%s" % log_msg)
411        if self.sdm_log:
412            for ad in ads:
413                ad.adb.shell("i2cset -fy 3 64 6 1 b", ignore_status=True)
414                ad.adb.shell("i2cset -fy 3 65 6 1 b", ignore_status=True)
415            start_sdm_loggers(self.log, self.android_devices)
416        else:
417            start_qxdm_loggers(self.log, self.android_devices)
418        if self.cbrs_esim:
419            self._cbrs_data_check_test(begin_time, expected_cbrs=True,
420                                       test_time="before")
421        failure_reasons = set()
422        self.dut_incall = True
423        if self.single_phone_test:
424            call_setup_result = initiate_call(
425                self.log,
426                self.dut,
427                self.call_server_number,
428                incall_ui_display=INCALL_UI_DISPLAY_BACKGROUND
429            ) and wait_for_in_call_active(self.dut, 60, 3)
430        else:
431            call_setup_result = call_setup_teardown(
432                self.log,
433                ads[0],
434                ads[1],
435                ad_hangup=None,
436                verify_caller_func=call_verification_func,
437                verify_callee_func=call_verification_func,
438                wait_time_in_call=0,
439                incall_ui_display=INCALL_UI_DISPLAY_BACKGROUND,
440                slot_id_callee=slot_id_callee)
441        if not call_setup_result:
442            get_telephony_signal_strength(ads[0])
443            if not self.single_phone_test:
444                get_telephony_signal_strength(ads[1])
445            call_logs = ads[0].search_logcat(
446                "ActivityManager: START u0 {act=android.intent.action.CALL",
447                begin_time)
448            messaging_logs = ads[0].search_logcat(
449                "com.google.android.apps.messaging/.ui.conversation.ConversationActivity",
450                begin_time)
451            if call_logs and messaging_logs:
452                if (messaging_logs[-1]["datetime_obj"] -
453                        call_logs[-1]["datetime_obj"]).seconds < 5:
454                    ads[0].log.info(
455                        "Call setup failure due to simultaneous activities")
456                    self.result_info[
457                        "Call Setup Failure With Simultaneous Activity"] += 1
458                    return True
459            self.log.error("%s: Setup Call failed.", log_msg)
460            failure_reasons.add("Setup")
461            result = False
462        else:
463            elapsed_time = 0
464            check_interval = 5
465            if self.sdm_log:
466                for ad in ads:
467                    ad.adb.shell("i2cset -fy 3 64 6 1 b", ignore_status=True)
468                    ad.adb.shell("i2cset -fy 3 65 6 1 b", ignore_status=True)
469            if self.cbrs_esim:
470                time.sleep(5)
471                self._cbrs_data_check_test(begin_time, expected_cbrs=False,
472                                           test_time="during")
473            while (elapsed_time < duration):
474                check_interval = min(check_interval, duration - elapsed_time)
475                time.sleep(check_interval)
476                elapsed_time += check_interval
477                time_message = "at <%s>/<%s> second." % (elapsed_time,
478                                                         duration)
479                for ad in ads:
480                    get_telephony_signal_strength(ad)
481                    if not call_verification_func(self.log, ad):
482                        ad.log.warning("Call is NOT in correct %s state at %s",
483                                       call_verification_func.__name__,
484                                       time_message)
485                        if call_verification_func.__name__ == "is_phone_in_call_iwlan":
486                            if is_phone_in_call(self.log, ad):
487                                if getattr(ad, "data_rat_state_error_count",
488                                           0) < 1:
489                                    setattr(ad, "data_rat_state_error_count",
490                                            1)
491                                    continue
492                        failure_reasons.add("Maintenance")
493                        last_call_drop_reason(ad, begin_time)
494                        hangup_call(self.log, ads[0])
495                        result = False
496                    else:
497                        ad.log.info("Call is in correct %s state at %s",
498                                    call_verification_func.__name__,
499                                    time_message)
500                if not result:
501                    break
502        if not hangup_call(self.log, ads[0]):
503            failure_reasons.add("Teardown")
504            result = False
505        for ad in ads:
506            if not wait_for_call_id_clearing(ad,
507                                             []) or ad.droid.telecomIsInCall():
508                ad.log.error("Fail to hang up call")
509                failure_reasons.add("Teardown")
510                result = False
511        self.result_info["Call Total"] += 1
512        for ad in self.android_devices:
513            try:
514                ad.droid.logI("[END]%s" % log_msg)
515            except:
516                pass
517        self.log.info("%s end", log_msg)
518        self.dut_incall = False
519        if self.cbrs_esim:
520            time.sleep(30)
521            self._cbrs_data_check_test(begin_time, expected_cbrs=True,
522                                       test_time="after")
523        if not result:
524            self.log.info("%s failed", log_msg)
525            if self.gps_log_file:
526                gps_info = job.run(
527                    "tail %s" % self.gps_log_file, ignore_status=True)
528                if gps_info.stdout:
529                    gps_log_path = os.path.join(self.log_path, test_name)
530                    os.makedirs(gps_log_path, exist_ok=True)
531                    job.run(
532                        "tail %s > %s" %
533                        (self.gps_log_file,
534                         os.path.join(gps_log_path, "gps_logs.txt")),
535                        ignore_status=True)
536                    self.log.info("gps log:\n%s", gps_info.stdout)
537                else:
538                    self.log.warning("Fail to get gps log %s",
539                                     self.user_params["gps_log_file"])
540            for reason in failure_reasons:
541                self.result_info["Call %s Failure" % reason] += 1
542            for ad in ads:
543                log_path = os.path.join(self.log_path, test_name,
544                                        "%s_binder_logs" % ad.serial)
545                os.makedirs(log_path, exist_ok=True)
546                ad.pull_files(BINDER_LOGS, log_path)
547            try:
548                self._take_bug_report(test_name, begin_time)
549            except Exception as e:
550                self.log.exception(e)
551            for ad in ads:
552                if ad.droid.telecomIsInCall():
553                    hangup_call_by_adb(ad)
554        else:
555            self.log.info("%s test succeed", log_msg)
556            self.result_info["Call Success"] += 1
557            if self.result_info["Call Total"] % 50 == 0:
558                for ad in ads:
559                    synchronize_device_time(ad)
560                    force_connectivity_metrics_upload(ad)
561                    if self.get_binder_logs:
562                        log_path = os.path.join(self.log_path, test_name,
563                                                "%s_binder_logs" % ad.serial)
564                        os.makedirs(log_path, exist_ok=True)
565                        ad.pull_files(BINDER_LOGS, log_path)
566        return result
567
568    def _prefnetwork_mode_change(self, sub_id):
569        # ModePref change to non-LTE
570        begin_time = get_device_epoch_time(self.dut)
571        if self.sdm_log:
572            start_sdm_loggers(self.log, self.android_devices)
573        else:
574            start_qxdm_loggers(self.log, self.android_devices)
575        self.result_info["Network Change Request Total"] += 1
576        test_name = "%s_network_change_iter_%s" % (
577            self.test_name, self.result_info["Network Change Request Total"])
578        log_msg = "[Test Case] %s" % test_name
579        self.log.info("%s begin", log_msg)
580        self.dut.droid.logI("[BEGIN]%s" % log_msg)
581        network_preference_list = [
582            NETWORK_MODE_TDSCDMA_GSM_WCDMA, NETWORK_MODE_WCDMA_ONLY,
583            NETWORK_MODE_GLOBAL, NETWORK_MODE_CDMA, NETWORK_MODE_GSM_ONLY
584        ]
585        network_preference = random.choice(network_preference_list)
586        set_preferred_network_mode_pref(self.log, self.dut, sub_id,
587                                        network_preference)
588        time.sleep(WAIT_TIME_AFTER_MODE_CHANGE)
589        self.dut.log.info("Current Voice RAT is %s",
590                          get_current_voice_rat(self.log, self.dut))
591
592        # ModePref change back to with LTE
593        if not phone_setup_volte(self.log, self.dut):
594            self.dut.log.error("Phone failed to enable VoLTE.")
595            self.result_info["VoLTE Setup Failure"] += 1
596            self.dut.droid.logI("%s end" % log_msg)
597            self.dut.log.info("[END]%s", log_msg)
598            try:
599                self._ad_take_extra_logs(self.dut, test_name, begin_time)
600                self._ad_take_bugreport(self.dut, test_name, begin_time)
601            except Exception as e:
602                self.log.exception(e)
603            return False
604        else:
605            self.result_info["VoLTE Setup Success"] += 1
606            return True
607
608    def _mobile_data_toggling(self, setup="volte"):
609        # ModePref change to non-LTE
610        begin_time = get_device_epoch_time(self.dut)
611        if self.sdm_log:
612            start_sdm_loggers(self.log, self.android_devices)
613        else:
614            start_qxdm_loggers(self.log, self.android_devices)
615        result = True
616        self.result_info["Data Toggling Request Total"] += 1
617        test_name = "%s_data_toggling_iter_%s" % (
618            self.test_name, self.result_info["Data Toggling Request Total"])
619        log_msg = "[Test Case] %s" % test_name
620        self.log.info("%s begin", log_msg)
621        self.dut.droid.logI("[BEGIN]%s" % log_msg)
622        self.dut.adb.shell("svc data disable")
623        time.sleep(WAIT_TIME_AFTER_MODE_CHANGE)
624        self.dut.adb.shell("svc data enable")
625        if not self._check_data():
626            result = False
627        elif setup == "volte" and not phone_idle_volte(self.log, self.dut):
628            result = False
629        self.dut.droid.logI("%s end" % log_msg)
630        self.dut.log.info("[END]%s", log_msg)
631        if not result:
632            self.result_info["Data Toggling Failure"] += 1
633            try:
634                self._ad_take_extra_logs(self.dut, test_name, begin_time)
635                self._ad_take_bugreport(self.dut, test_name, begin_time)
636            except Exception as e:
637                self.log.exception(e)
638            return False
639        else:
640            self.result_info["Data Toggling Success"] += 1
641            return True
642
643    def _get_result_message(self):
644        msg_list = [
645            "%s: %s" % (count, self.result_info[count])
646            for count in sorted(self.result_info.keys())
647        ]
648        return ", ".join(msg_list)
649
650    def _write_perf_json(self):
651        json_str = json.dumps(self.perf_data, indent=4, sort_keys=True)
652        with open(self.perf_file, 'w') as f:
653            f.write(json_str)
654
655    def _init_perf_json(self):
656        self.perf_file = os.path.join(self.log_path, "%s_perf_data_%s.json" %
657                                      (self.test_name, self.begin_time))
658        self.perf_data = self.dut.build_info.copy()
659        self.perf_data["build_fingerprint"] = self.dut.adb.getprop(
660            "ro.build.fingerprint")
661        self.perf_data["model"] = self.dut.model
662        self.perf_data["carrier"] = self.dut.adb.getprop(
663            "gsm.sim.operator.alpha")
664        self._write_perf_json()
665
666    def _update_perf_json(self):
667        for result_key, result_value in self.result_info.items():
668            self.perf_data[result_key] = result_value
669        self._write_perf_json()
670
671    def crash_check_test(self):
672        failure = 0
673        while time.time() < self.finishing_time:
674            try:
675                self.log.info(dict(self.result_info))
676                self._update_perf_json()
677                begin_time = get_device_epoch_time(self.dut)
678                run_time_in_seconds = (begin_time - self.begin_time) / 1000
679                test_name = "%s_crash_%s_seconds_after_start" % (
680                    self.test_name, run_time_in_seconds)
681                time.sleep(self.crash_check_interval)
682                for ad in self.android_devices:
683                    crash_report = ad.check_crash_report(
684                        test_name, begin_time, log_crash_report=True)
685                    if crash_report:
686                        ad.log.error("Find new crash reports %s", crash_report)
687                        failure += 1
688                        self.result_info["Crashes"] += len(crash_report)
689                        for crash in crash_report:
690                            if "ramdump_modem" in crash:
691                                self.result_info["Crashes-Modem"] += 1
692                        try:
693                            ad.take_bug_report(test_name, begin_time)
694                        except Exception as e:
695                            self.log.exception(e)
696            except Exception as e:
697                self.log.error("Exception error %s", str(e))
698                self.result_info["Exception Errors"] += 1
699            self.log.info("Crashes found: %s", failure)
700            if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE:
701                self.log.error("Too many exception errors, quit test")
702                return False
703        if failure:
704            return False
705        else:
706            return True
707
708    def _cbrs_data_check_test(self, begin_time, expected_cbrs=True,
709                              test_time="before"):
710        cbrs_fail_count = 0
711        the_number = self.result_info["CBRS Total"] + 1
712        test_name = "%s_cbrs_%s_call_No_%s" % (self.test_name,
713                                               test_time, the_number)
714        for ad in self.android_devices:
715            current_state = is_current_data_on_cbrs(ad, ad.cbrs)
716            if current_state == expected_cbrs:
717                self.result_info["CBRS-Check-Pass"] += 1
718            else:
719                self.result_info["CBRS-Check-Fail"] += 1
720                cbrs_fail_count += 1
721                try:
722                    self._ad_take_extra_logs(ad, test_name, begin_time)
723                    self._ad_take_bugreport(ad, test_name, begin_time)
724                except Exception as e:
725                    self.log.warning(e)
726        if cbrs_fail_count > 0:
727            ad.log.error("Found %d checks failed, expected cbrs %s",
728                         cbrs_fail_count, expected_cbrs)
729            cbrs_fail_count += 1
730        self.result_info["CBRS Total"] += 1
731        return True
732
733    def call_test(self, call_verification_func=None):
734        while time.time() < self.finishing_time:
735            time.sleep(
736                random.randrange(self.min_sleep_time, self.max_sleep_time))
737            try:
738                self._make_phone_call(call_verification_func)
739            except Exception as e:
740                self.log.exception("Exception error %s", str(e))
741                self.result_info["Exception Errors"] += 1
742            if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE:
743                self.log.error("Too many exception errors, quit test")
744                return False
745            self.log.info("%s", dict(self.result_info))
746        if any([
747                self.result_info["Call Setup Failure"],
748                self.result_info["Call Maintenance Failure"],
749                self.result_info["Call Teardown Failure"]
750        ]):
751            return False
752        else:
753            return True
754
755    def message_test(self, max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE):
756        while time.time() < self.finishing_time:
757            try:
758                self._send_message(max_wait_time=max_wait_time)
759            except Exception as e:
760                self.log.exception("Exception error %s", str(e))
761                self.result_info["Exception Errors"] += 1
762            self.log.info(dict(self.result_info))
763            if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE:
764                self.log.error("Too many exception errors, quit test")
765                return False
766            time.sleep(
767                random.randrange(self.min_sleep_time, self.max_sleep_time))
768        if self.result_info["SMS Failure"] or (
769                self.result_info["MMS Failure"] / self.result_info["MMS Total"]
770                > 0.3):
771            return False
772        else:
773            return True
774
775    def _data_download(self, file_names=[]):
776        begin_time = get_current_epoch_time()
777        slot_id = random.randint(0,1)
778        if self.dsds_esim:
779            sub_id = get_subid_from_slot_index(self.log, self.dut, slot_id)
780            self.dut.log.info("Data - slot_Id %d", slot_id)
781            set_subid_for_data(self.dut, sub_id)
782            self.dut.droid.telephonyToggleDataConnection(True)
783        if self.sdm_log:
784            start_sdm_loggers(self.log, self.android_devices)
785        else:
786            start_qxdm_loggers(self.log, self.android_devices)
787        self.dut.log.info(dict(self.result_info))
788        selection = random.randrange(0, len(file_names))
789        file_name = file_names[selection]
790        self.result_info["Internet Connection Check Total"] += 1
791
792        rat = self._get_network_rat(slot_id)
793        if not self.internet_connection_check_method(self.log, self.dut):
794            self.dut.log.info("Network in RAT %s", rat)
795            if self.dut_incall and not is_rat_svd_capable(rat.upper()):
796                self.result_info[
797                    "Expected Incall Internet Connection Check Failure"] += 1
798                return True
799            else:
800                self.result_info["Internet Connection Check Failure"] += 1
801                test_name = "%s_internet_connection_No_%s_failure" % (
802                    self.test_name,
803                    self.result_info["Internet Connection Check Failure"])
804                try:
805                    self._ad_take_extra_logs(self.dut, test_name, begin_time)
806                    self._ad_take_bugreport(self.dut, test_name, begin_time)
807                except Exception as e:
808                    self.log.exception(e)
809                return False
810        else:
811            self.result_info["Internet Connection Check Success"] += 1
812
813        self.result_info["File Download Total"] += 1
814        if not active_file_download_test(
815                self.log, self.dut, file_name,
816                method=self.file_download_method):
817            self.result_info["File Download Failure"] += 1
818            if self.result_info["File Download Failure"] == 1:
819                try:
820                    self._ad_take_extra_logs(
821                        self.dut, "%s_file_download_failure" % self.test_name,
822                        begin_time)
823                    self._ad_take_bugreport(
824                        self.dut, "%s_file_download_failure" % self.test_name,
825                        begin_time)
826                except Exception as e:
827                    self.log.exception(e)
828            return False
829        else:
830            self.result_info["File Download Success"] += 1
831            return True
832
833    def data_test(self):
834        while time.time() < self.finishing_time:
835            try:
836                self._data_download(self.file_name_list)
837            except Exception as e:
838                self.log.error("Exception error %s", str(e))
839                self.result_info["Exception Errors"] += 1
840            self.log.info("%s", dict(self.result_info))
841            if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE:
842                self.log.error("Too many exception errors, quit test")
843                return False
844            time.sleep(
845                random.randrange(self.min_sleep_time, self.max_sleep_time))
846        if self.result_info["Internet Connection Check Failure"]:
847            return False
848        else:
849            return True
850
851    def _check_data(self):
852        self.result_info["Data Connection Check Total"] += 1
853        if not wait_for_data_connection(self.log, self.dut, True):
854            self.result_info["Data Connection Setup Failure"] += 1
855            return False
856        if not self.internet_connection_check_method(self.log, self.dut):
857            rat = self.dut.adb.getprop("gsm.network.type")
858            self.dut.log.info("Network in RAT %s", rat)
859            self.result_info["Internet Connection Check Failure"] += 1
860            return False
861        return True
862
863    def _data_call_test(self, sub_id, generation):
864        self.dut.log.info(dict(self.result_info))
865        begin_time = get_device_epoch_time(self.dut)
866        if self.sdm_log:
867            start_sdm_loggers(self.log, self.android_devices)
868        else:
869            start_qxdm_loggers(self.log, self.android_devices)
870        self.result_info["Network Change Request Total"] += 1
871        test_name = "%s_network_change_test_iter_%s" % (
872            self.test_name, self.result_info["Network Change Request Total"])
873        log_msg = "[Test Case] %s" % test_name
874        self.log.info("%s begin", log_msg)
875        self.dut.droid.logI("[BEGIN]%s" % log_msg)
876        if not ensure_network_generation_for_subscription(
877                self.log, self.dut, sub_id,
878                generation) or not self._check_data():
879            self.result_info["Network Change Failure"] += 1
880            self.dut.droid.logI("%s end" % log_msg)
881            self.dut.log.info("[END]%s", log_msg)
882            try:
883                self._ad_take_extra_logs(self.dut, test_name, begin_time)
884                self._ad_take_bugreport(self.dut, test_name, begin_time)
885            except Exception as e:
886                self.log.warning(e)
887            return False
888        if not self._mobile_data_toggling(setup=None):
889            return False
890        return True
891
892    def data_call_stress_test(self):
893        result = True
894        sub_id = self.dut.droid.subscriptionGetDefaultSubId()
895        while time.time() < self.finishing_time:
896            for generation in (GEN_4G, GEN_3G):
897                try:
898                    if not self._data_call_test(sub_id, generation):
899                        result = False
900                except Exception as e:
901                    self.log.error("Exception error %s", str(e))
902                    self.result_info["Exception Errors"] += 1
903            if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE:
904                self.log.error("Too many exception errors, quit test")
905                return False
906        return result
907
908    def check_incall_data(self):
909        if verify_internet_connection_by_ping(self.log, self.dut):
910            self.internet_connection_check_method = verify_internet_connection_by_ping
911        elif verify_http_connection(self.log, self.dut):
912            self.internet_connection_check_method = verify_http_connection
913        else:
914            self.dut.log.error("Data test failed")
915            raise signals.TestFailure("Data check failed")
916        if self.single_phone_test:
917            if not initiate_call(
918                    self.log, self.dut,
919                    self.call_server_number) and wait_for_in_call_active(
920                        self.dut, 60, 3):
921                self._take_bug_report(self.test_name, self.begin_time)
922                raise signals.TestFailure("Unable to make phone call")
923        else:
924            if not call_setup_teardown(
925                    self.log, self.dut, self.android_devices[1],
926                    ad_hangup=None):
927                self._take_bug_report(self.test_name, self.begin_time)
928                raise signals.TestFailure("Unable to make phone call")
929        voice_rat = self.dut.droid.telephonyGetCurrentVoiceNetworkType()
930        data_rat = self.dut.droid.telephonyGetCurrentDataNetworkType()
931        self.dut.log.info("Voice in RAT %s, Data in RAT %s", voice_rat,
932                          data_rat)
933        try:
934            if "wfc" in self.test_name or is_rat_svd_capable(
935                    voice_rat.upper()) and is_rat_svd_capable(
936                        data_rat.upper()):
937                self.dut.log.info("Capable for simultaneous voice and data")
938
939                if not self.internet_connection_check_method(
940                        self.log, self.dut):
941                    self.dut.log.error("Incall data check failed")
942                    raise signals.TestFailure("Incall data check failed")
943                else:
944                    return True
945            else:
946                self.dut.log.info(
947                    "Not capable for simultaneous voice and data")
948                return False
949            hangup_call(self.log, self.dut)
950        finally:
951            for ad in self.android_devices:
952                if ad.droid.telecomIsInCall():
953                    hangup_call(self.log, ad)
954
955    def parallel_tests(self, setup_func=None, call_verification_func=None):
956        self.log.info(self._get_result_message())
957        if setup_func and not setup_func():
958            msg = "%s setup %s failed" % (self.test_name, setup_func.__name__)
959            self.log.error(msg)
960            self._take_bug_report("%s%s" % (self.test_name,
961                                            setup_func.__name__),
962                                  self.begin_time)
963            return False
964        if not call_verification_func:
965            call_verification_func = is_phone_in_call
966        self.finishing_time = time.time() + self.max_run_time
967        # CBRS setup
968        if self.cbrs_esim:
969            cbrs_sub_count = 0
970            for ad in self.android_devices:
971                if not getattr(ad, 'cbrs', {}):
972                    setattr(ad, 'cbrs', None)
973                for i in range(0, 2):
974                    sub_id = get_subid_from_slot_index(ad.log, ad, i)
975                    operator = get_operatorname_from_slot_index(ad, i)
976                    carrier_id = get_carrierid_from_slot_index(ad, i)
977                    is_opportunistic = get_isopportunistic_from_slot_index(ad, i)
978                    ad.log.info("Slot %d - Sub %s - %s - %d", i, sub_id, operator, carrier_id)
979                    if carrier_id == 2340 or (carrier_id == 1989 and is_opportunistic):
980                        ad.cbrs = sub_id
981                        cbrs_sub_count += 1
982            if cbrs_sub_count != 2:
983                self.log.error("Expecting - 2 CBRS subs, found - %d", cbrs_sub_count)
984                raise signals.TestAbortClass("Cannot find all expected CBRS subs")
985        # DSDS setup
986        if self.dsds_esim:
987            for ad in self.android_devices:
988                for i in range(0, 2):
989                    sub_id = get_subid_from_slot_index(ad.log, ad, i)
990                    set_slways_allow_mms_data(ad, sub_id)
991                    operator = get_operatorname_from_slot_index(ad, i)
992                    ad.log.info("Slot %d - Sub %s - %s", i, sub_id, operator)
993        # Actual test trigger
994        if not self.dsds_esim and self.check_incall_data():
995            self.log.info(
996                "==== Start parallel voice/message/data stress test ====")
997            self.perf_data["testing method"] = "parallel"
998            results = run_multithread_func(
999                self.log, [(self.call_test, [call_verification_func]),
1000                           (self.message_test, []), (self.data_test, []),
1001                           (self.crash_check_test, [])])
1002        else:
1003            self.log.info(
1004                "==== Start sequential voice/message/data stress test ====")
1005            self.perf_data["testing method"] = "sequential"
1006            results = run_multithread_func(
1007                self.log, [(self.sequential_tests, [call_verification_func]),
1008                           (self.crash_check_test, [])])
1009        result_message = self._get_result_message()
1010        self.log.info(result_message)
1011        self._update_perf_json()
1012        self.result_detail = result_message
1013        return all(results)
1014
1015    def sequential_tests(self, call_verification_func):
1016        funcs = [(self._make_phone_call, [call_verification_func]),
1017                 (self._send_message, []), (self._data_download, [["5MB"]])]
1018        while time.time() < self.finishing_time:
1019            selection = random.randrange(0, 3)
1020            try:
1021                funcs[selection][0](*funcs[selection][1])
1022            except Exception as e:
1023                self.log.error("Exception error %s", str(e))
1024                self.result_info["Exception Errors"] += 1
1025            self.log.info("%s", dict(self.result_info))
1026            if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE:
1027                self.log.error("Too many exception errors, quit test")
1028                return False
1029            time.sleep(
1030                random.randrange(self.min_sleep_time, self.max_sleep_time))
1031        if any([
1032                self.result_info["Call Setup Failure"],
1033                self.result_info["Call Maintenance Failure"],
1034                self.result_info["Call Teardown Failure"],
1035                self.result_info["SMS Failure"],
1036                self.result_info["MMS Failure"],
1037                self.result_info["Internet Connection Check Failure"]
1038        ]):
1039            return False
1040        return True
1041
1042    def volte_modechange_volte_test(self):
1043        sub_id = self.dut.droid.subscriptionGetDefaultSubId()
1044        result = True
1045        while time.time() < self.finishing_time:
1046            try:
1047                if self._prefnetwork_mode_change(sub_id):
1048                    run_multithread_func(
1049                        self.log,
1050                        [(self._data_download, [["5MB"]]),
1051                         (self._make_phone_call, [is_phone_in_call_volte]),
1052                         (self._send_message, [])])
1053                else:
1054                    result = False
1055                if self._mobile_data_toggling():
1056                    run_multithread_func(
1057                        self.log,
1058                        [(self._data_download, [["5MB"]]),
1059                         (self._make_phone_call, [is_phone_in_call_volte]),
1060                         (self._send_message, [])])
1061                else:
1062                    result = False
1063            except Exception as e:
1064                self.log.error("Exception error %s", str(e))
1065                self.result_info["Exception Errors"] += 1
1066            self.log.info(dict(self.result_info))
1067            if self.result_info["Exception Errors"] >= EXCEPTION_TOLERANCE:
1068                self.log.error("Too many exception errors, quit test")
1069                return False
1070        return result
1071
1072    def parallel_with_network_change_tests(self, setup_func=None):
1073        if setup_func and not setup_func():
1074            self.log.error("Test setup %s failed", setup_func.__name__)
1075            return False
1076        self.finishing_time = time.time() + self.max_run_time
1077        results = run_multithread_func(self.log,
1078                                       [(self.volte_modechange_volte_test, []),
1079                                        (self.crash_check_test, [])])
1080        result_message = self._get_result_message()
1081        self.log.info(result_message)
1082        self._update_perf_json()
1083        self.result_detail = result_message
1084        return all(results)
1085
1086    def connect_to_wifi(self):
1087        for ad in self.android_devices:
1088            if not ensure_wifi_connected(
1089                    self.log,
1090                    ad,
1091                    self.wifi_network_ssid,
1092                    self.wifi_network_pass,
1093                    retries=3):
1094                ad.log.error("Bringing up Wifi connection fails.")
1095                return False
1096        ad.log.info("Phone WIFI is connected successfully.")
1097        return True
1098
1099    """ Tests Begin """
1100
1101    @test_tracker_info(uuid="d035e5b9-476a-4e3d-b4e9-6fd86c51a68d")
1102    @TelephonyBaseTest.tel_test_wrap
1103    def test_default_parallel_stress(self):
1104        """ Default state stress test"""
1105        return self.parallel_tests()
1106
1107    @test_tracker_info(uuid="798a3c34-db75-4bcf-b8ef-e1116414a7fe")
1108    @TelephonyBaseTest.tel_test_wrap
1109    def test_default_parallel_stress_with_wifi(self):
1110        """ Default state stress test with Wifi enabled."""
1111        if self.connect_to_wifi():
1112            return self.parallel_tests()
1113
1114    @test_tracker_info(uuid="c21e1f17-3282-4f0b-b527-19f048798098")
1115    @TelephonyBaseTest.tel_test_wrap
1116    def test_lte_volte_parallel_stress(self):
1117        """ VoLTE on stress test"""
1118        return self.parallel_tests(
1119            setup_func=self._setup_lte_volte_enabled,
1120            call_verification_func=is_phone_in_call_volte)
1121
1122    @test_tracker_info(uuid="a317c23a-41e0-4ef8-af67-661451cfefcf")
1123    @TelephonyBaseTest.tel_test_wrap
1124    def test_csfb_parallel_stress(self):
1125        """ LTE non-VoLTE stress test"""
1126        return self.parallel_tests(
1127            setup_func=self._setup_lte_volte_disabled,
1128            call_verification_func=is_phone_in_call_csfb)
1129
1130    @test_tracker_info(uuid="fdb791bf-c414-4333-9fa3-cc18c9b3b234")
1131    @TelephonyBaseTest.tel_test_wrap
1132    def test_wfc_parallel_stress(self):
1133        """ Wifi calling APM mode off stress test"""
1134        if WFC_MODE_WIFI_PREFERRED not in self.dut_wfc_modes:
1135            raise signals.TestSkip("WFC_MODE_WIFI_PREFERRED is not supported")
1136        return self.parallel_tests(
1137            setup_func=self._setup_wfc,
1138            call_verification_func=is_phone_in_call_iwlan)
1139
1140    @test_tracker_info(uuid="e334c1b3-4378-49bb-bf57-1573fa1b23fa")
1141    @TelephonyBaseTest.tel_test_wrap
1142    def test_wfc_apm_parallel_stress(self):
1143        """ Wifi calling in APM mode on stress test"""
1144        return self.parallel_tests(
1145            setup_func=self._setup_wfc_apm,
1146            call_verification_func=is_phone_in_call_iwlan)
1147
1148    @test_tracker_info(uuid="4566eef6-55de-4ac8-87ee-58f2ef41a3e8")
1149    @TelephonyBaseTest.tel_test_wrap
1150    def test_3g_parallel_stress(self):
1151        """ 3G stress test"""
1152        return self.parallel_tests(
1153            setup_func=self._setup_3g,
1154            call_verification_func=is_phone_in_call_3g)
1155
1156    @test_tracker_info(uuid="f34f1a31-3948-4675-8698-372a83b8088d")
1157    @TelephonyBaseTest.tel_test_wrap
1158    def test_2g_parallel_stress(self):
1159        """ 2G call stress test"""
1160        return self.parallel_tests(
1161            setup_func=self._setup_2g,
1162            call_verification_func=is_phone_in_call_2g)
1163
1164    @test_tracker_info(uuid="af580fca-fea6-4ca5-b981-b8c710302d37")
1165    @TelephonyBaseTest.tel_test_wrap
1166    def test_volte_modeprefchange_parallel_stress(self):
1167        """ VoLTE Mode Pref call stress test"""
1168        return self.parallel_with_network_change_tests(
1169            setup_func=self._setup_lte_volte_enabled)
1170
1171    @test_tracker_info(uuid="10e34247-5fd3-4f87-81bf-3c17a6b71ab2")
1172    @TelephonyBaseTest.tel_test_wrap
1173    def test_data_call_stress(self):
1174        """ Default state stress test"""
1175        self.finishing_time = time.time() + self.max_run_time
1176        results = run_multithread_func(self.log,
1177                                       [(self.data_call_stress_test, []),
1178                                        (self.crash_check_test, [])])
1179        result_message = self._get_result_message()
1180        self.log.info(result_message)
1181        self._update_perf_json()
1182        self.result_detail = result_message
1183        return all(results)
1184
1185    """ Tests End """
1186