1#!/usr/bin/env python3
2#
3#   Copyright 2020 - 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
17import random
18import time
19from acts import signals
20from acts.test_decorators import test_tracker_info
21from acts_contrib.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
22from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_ANDROID_STATE_SETTLING
23from acts_contrib.test_utils.tel.tel_subscription_utils \
24    import set_message_subid
25from acts_contrib.test_utils.tel.tel_subscription_utils \
26    import get_subid_on_same_network_of_host_ad
27from acts_contrib.test_utils.tel.tel_test_utils import ensure_phones_idle
28from acts_contrib.test_utils.tel.tel_test_utils import multithread_func
29from acts_contrib.test_utils.tel.tel_test_utils import sms_send_receive_verify
30from acts_contrib.test_utils.tel.tel_voice_utils \
31    import phone_setup_volte_for_subscription
32from acts_contrib.test_utils.tel.tel_voice_utils \
33    import phone_setup_csfb_for_subscription
34from acts_contrib.test_utils.tel.tel_voice_utils import phone_setup_on_rat
35from acts.utils import rand_ascii_str
36
37class TelLiveStressSmsTest(TelephonyBaseTest):
38    def setup_class(self):
39        TelephonyBaseTest.setup_class(self)
40        self.short_sms_cs_4g_cycle = \
41            self.user_params.get("short_sms_cs_4g_cycle", 1)
42        self.long_sms_cs_4g_cycle = \
43            self.user_params.get("long_sms_cs_4g_cycle", 1)
44        self.short_sms_ims_cycle = \
45            self.user_params.get("short_sms_ims_cycle", 1)
46        self.long_sms_ims_cycle = \
47            self.user_params.get("long_sms_ims_cycle", 1)
48
49    def teardown_test(self):
50        ensure_phones_idle(self.log, self.android_devices)
51
52    def sms_test(self, ads, type='short', rat='volte'):
53        if type == 'short':
54            sms_length = random.randint(50, 180)
55            sms_body = rand_ascii_str(sms_length)
56        else:
57            sms_length = random.randint(800, 1600)
58            sms_body = rand_ascii_str(sms_length)
59
60        mo_sub_id, mt_sub_id, _ = get_subid_on_same_network_of_host_ad(
61            ads, host_sub_id=None, type="sms")
62        set_message_subid(ads[1], mt_sub_id)
63
64        if rat == 'volte':
65            phone_setup_func = phone_setup_volte_for_subscription
66        else:
67            phone_setup_func = phone_setup_csfb_for_subscription
68
69        tasks = [(phone_setup_func, (self.log, ads[0], mo_sub_id)),
70                (phone_setup_func, (self.log, ads[1], mt_sub_id))]
71        if not multithread_func(self.log, tasks):
72            self.log.error("Phone Failed to Set Up Properly.")
73            return False
74
75        time.sleep(WAIT_TIME_ANDROID_STATE_SETTLING)
76
77        if not sms_send_receive_verify(self.log, ads[0], ads[1], [sms_body]):
78            ads[0].log.warning("SMS of length %s test failed", sms_length)
79            return False
80
81        ads[0].log.info("SMS of length %s test succeeded", sms_length)
82        time.sleep(random.randint(5, 10))
83        return True
84
85    def calculte_fail_rate(self, result_dict={}):
86        result_list = []
87        for key in result_dict:
88            if result_dict[key]:
89                result_list.append(True)
90            else:
91                result_list.append(False)
92
93        fail_rate = result_list.count(False)/len(result_list)
94        self.log.info('Test result: %s', result_dict)
95        return fail_rate
96
97    @test_tracker_info(uuid="e54d15bf-db9b-4389-bfe8-c9c5dad7ef33")
98    @TelephonyBaseTest.tel_test_wrap
99    def test_stress_short_sms_ims(self):
100        cycle = self.short_sms_ims_cycle
101        result_dict = {}
102        for attempt in range(cycle):
103            self.log.info(
104                "============> Stress short SMS over IMS test cycle %s <============",
105                attempt+1)
106            result = self.sms_test(self.android_devices, 'short', 'volte')
107            result_dict['cycle_%s' % str(attempt+1)] = result
108
109        fail_rate = self.calculte_fail_rate(result_dict)
110        self.log.info('Fail rate of short SMS over IMS: %s', fail_rate)
111
112    @test_tracker_info(uuid="45858a3c-0c37-4b3d-af4d-60f374d0a2e5")
113    @TelephonyBaseTest.tel_test_wrap
114    def test_stress_long_sms_ims(self):
115        cycle = self.long_sms_ims_cycle
116        result_dict = {}
117        for attempt in range(cycle):
118            self.log.info(
119                "============> Stress long SMS over IMS test cycle %s <============",
120                attempt+1)
121            result = self.sms_test(self.android_devices, 'long', 'volte')
122            result_dict['cycle_%s' % str(attempt+1)] = result
123
124        fail_rate = self.calculte_fail_rate(result_dict)
125        self.log.info('Fail rate of long SMS over IMS: %s', fail_rate)
126
127    @test_tracker_info(uuid="319aec8f-2e04-4420-9f8a-388c092ddd39")
128    @TelephonyBaseTest.tel_test_wrap
129    def test_stress_short_sms_cs_4g(self):
130        cycle = self.short_sms_cs_4g_cycle
131        result_dict = {}
132        for attempt in range(cycle):
133            self.log.info(
134                "============> Stress short CS SMS on LTE test cycle %s <============",
135                attempt+1)
136            result = self.sms_test(self.android_devices, 'short', 'csfb')
137            result_dict['cycle_%s' % str(attempt+1)] = result
138
139        fail_rate = self.calculte_fail_rate(result_dict)
140        self.log.info('Fail rate of short CS SMS on LTE: %s', fail_rate)
141
142    @test_tracker_info(uuid="aa8a2e7a-1ca7-4406-801d-f8e1923d0695")
143    @TelephonyBaseTest.tel_test_wrap
144    def test_stress_long_sms_cs_4g(self):
145        cycle = self.long_sms_cs_4g_cycle
146        result_dict = {}
147        for attempt in range(cycle):
148            self.log.info(
149                "============> Stress long CS SMS on LTE test cycle %s <============",
150                attempt+1)
151            result = self.sms_test(self.android_devices, 'long', 'csfb')
152            result_dict['cycle_%s' % str(attempt+1)] = result
153
154        fail_rate = self.calculte_fail_rate(result_dict)
155        self.log.info('Fail rate of long CS SMS on LTE: %s', fail_rate)