1#!/usr/bin/env python3
2#
3#   Copyright 2022 - 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 time
18from datetime import datetime, timedelta
19
20from acts import signals
21from acts.test_decorators import test_tracker_info
22from acts_contrib.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
23from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_ANDROID_STATE_SETTLING
24from acts_contrib.test_utils.tel.tel_defines import WFC_MODE_WIFI_PREFERRED
25from acts_contrib.test_utils.tel.tel_data_utils import activate_and_verify_cellular_data
26from acts_contrib.test_utils.tel.tel_data_utils import active_file_download_test
27from acts_contrib.test_utils.tel.tel_data_utils import deactivate_and_verify_cellular_data
28from acts_contrib.test_utils.tel.tel_ims_utils import toggle_wfc
29from acts_contrib.test_utils.tel.tel_ims_utils import wait_for_wfc_enabled
30from acts_contrib.test_utils.tel.tel_ims_utils import wait_for_wfc_disabled
31from acts_contrib.test_utils.tel.tel_parse_utils import print_nested_dict
32from acts_contrib.test_utils.tel.tel_parse_utils import parse_setup_data_call
33from acts_contrib.test_utils.tel.tel_parse_utils import parse_deactivate_data_call
34from acts_contrib.test_utils.tel.tel_parse_utils import parse_setup_data_call_on_iwlan
35from acts_contrib.test_utils.tel.tel_parse_utils import parse_deactivate_data_call_on_iwlan
36from acts_contrib.test_utils.tel.tel_phone_setup_utils import phone_setup_4g_for_subscription
37from acts_contrib.test_utils.tel.tel_phone_setup_utils import phone_setup_iwlan
38from acts_contrib.test_utils.tel.tel_subscription_utils import get_default_data_sub_id
39from acts_contrib.test_utils.tel.tel_subscription_utils import set_dds_on_slot_0
40from acts_contrib.test_utils.tel.tel_subscription_utils import set_dds_on_slot_1
41from acts_contrib.test_utils.tel.tel_subscription_utils import get_slot_index_from_data_sub_id
42from acts.utils import get_current_epoch_time
43from acts.libs.utils.multithread import multithread_func
44
45CALCULATE_EVERY_N_CYCLES = 10
46
47
48class TelLiveRilDataKpiTest(TelephonyBaseTest):
49    def setup_class(self):
50        TelephonyBaseTest.setup_class(self)
51        self.cycle_cellular_data_cycle = self.user_params.get(
52            "cycle_cellular_data_cycle", 1)
53        self.cycle_wfc_cycle = self.user_params.get("cycle_wfc_cycle", 1)
54        self.dds_switch_test_cycle = self.user_params.get(
55            "dds_switch_test_cycle", 1)
56        self.http_download_duration = self.user_params.get(
57            "http_download_duration", 3600)
58
59    def cycle_cellular_data(self, ad):
60        """ Toggle off and then toggle on again cellular data.
61
62        Args:
63            ad: Android object
64
65        Returns:
66            True if cellular data is cycled successfully. Otherwise False.
67        """
68        if not deactivate_and_verify_cellular_data(self.log, ad):
69            return False
70
71        if not activate_and_verify_cellular_data(self.log, ad):
72            return False
73
74        return True
75
76    def cycle_wfc(self, ad):
77        """ Toggle off and then toggle on again WFC.
78
79        Args:
80            ad: Android object
81
82        Returns:
83            True if WFC is cycled successfully. Otherwise False.
84        """
85        if not toggle_wfc(self.log, ad, new_state=False):
86            return False
87
88        if not wait_for_wfc_disabled(self.log, ad):
89            return False
90
91        if not toggle_wfc(self.log, ad, new_state=True):
92            return False
93
94        if not wait_for_wfc_enabled(self.log, ad):
95            return False
96
97        return True
98
99    def switch_dds(self, ad):
100        """Switch DDS to the other sub ID.
101
102        Args:
103            ad: Android object
104
105        Returns:
106            True if DDS is switched successfully. Otherwise False.
107        """
108        current_dds_slot = get_slot_index_from_data_sub_id(ad)
109
110        if current_dds_slot == 0:
111            if set_dds_on_slot_1(ad):
112                return True
113        else:
114            if set_dds_on_slot_0(ad):
115                return True
116
117        return False
118
119    @test_tracker_info(uuid="27424b59-efa9-47c3-89b4-4b5415003a58")
120    @TelephonyBaseTest.tel_test_wrap
121    def test_cycle_cellular_data_4g(self):
122        """Cycle cellular data on LTE to measure data call setup time,
123            deactivate time and LTE validation time.
124
125        Test steps:
126            1. Set up UE on LTE and ensure cellular data is connected.
127            2. Cycle cellular data.
128            3. Parse logcat to calculate data call setup time, deactivate time
129                and LTE validation time.
130        """
131        ad = self.android_devices[0]
132
133        cycle = self.cycle_cellular_data_cycle
134
135        tasks = [(
136            phone_setup_4g_for_subscription,
137            (self.log, ad, get_default_data_sub_id(ad)))]
138        if not multithread_func(self.log, tasks):
139            self.log.error("Phone Failed to Set Up Properly.")
140            return False
141
142        time.sleep(WAIT_TIME_ANDROID_STATE_SETTLING)
143
144        cycle_cellular_data_summary = []
145        for attempt in range(cycle):
146            ad.log.info(
147                '======> Cycling cellular data %s/%s <======',
148                attempt+1, cycle)
149            res = self.cycle_cellular_data(ad)
150            cycle_cellular_data_summary.append(res)
151            if not res:
152                self._take_bug_report(
153                    self.test_name, begin_time=get_current_epoch_time())
154
155            if (attempt+1) % CALCULATE_EVERY_N_CYCLES == 0 or attempt == cycle - 1:
156                (
157                    res,
158                    lst,
159                    avg_data_call_setup_time,
160                    avg_validation_time_on_lte) = parse_setup_data_call(ad)
161
162                ad.log.info('====== Setup data call list ======')
163                print_nested_dict(ad, res)
164
165                ad.log.info('====== Data call setup time list ======')
166                for item in lst:
167                    print_nested_dict(ad, item)
168                    ad.log.info('------------------')
169
170                (
171                    res,
172                    lst,
173                    avg_deactivate_data_call_time) = parse_deactivate_data_call(ad)
174
175                ad.log.info('====== Deactivate data call list ======')
176                print_nested_dict(ad, res)
177
178                ad.log.info('====== Data call deactivate time list ======')
179                for item in lst:
180                    print_nested_dict(ad, item)
181                    ad.log.info('------------------')
182
183                ad.log.info(
184                    'Average data call setup time on LTE: %.2f sec.',
185                    avg_data_call_setup_time)
186                ad.log.info(
187                    'Average validation time on LTE: %.2f sec.',
188                    avg_validation_time_on_lte)
189                ad.log.info(
190                    'Average deactivate data call time on LTE: %.2f sec.',
191                    avg_deactivate_data_call_time)
192
193                try:
194                    fail_rate = cycle_cellular_data_summary.count(False)/len(
195                            cycle_cellular_data_summary)
196                    self.log.info(
197                        'Fail rate of cycling cellular data on LTE: %s/%s (%.2f)',
198                        cycle_cellular_data_summary.count(False),
199                        len(cycle_cellular_data_summary),
200                        fail_rate)
201                except Exception as e:
202                    self.log.error(
203                        'Fail rate of cycling cellular data on LTE: ERROR (%s)',
204                        e)
205
206    @test_tracker_info(uuid="9f4ab929-176d-4f26-8e14-12bd6c25e80a")
207    @TelephonyBaseTest.tel_test_wrap
208    def test_cycle_wfc(self):
209        """Cycle WFC to measure data call setup time and deactivate time on
210            iwlan.
211
212        Test steps:
213            1. Set up UE on iwlan and ensure WFC is registered in Wi-Fi-preferred
214                mode.
215            2. Cycle WFC.
216            3. Parse logcat to calculate data call setup time and deactivate time
217                on iwlan.
218        """
219        ad = self.android_devices[0]
220
221        cycle = self.cycle_wfc_cycle
222
223        tasks = [(phone_setup_iwlan, (
224            self.log,
225            ad,
226            False,
227            WFC_MODE_WIFI_PREFERRED,
228            self.wifi_network_ssid,
229            self.wifi_network_pass))]
230        if not multithread_func(self.log, tasks):
231            self.log.error("Phone Failed to Set Up Properly.")
232            return False
233
234        time.sleep(WAIT_TIME_ANDROID_STATE_SETTLING)
235
236        cycle_wfc_summary = []
237        for attempt in range(cycle):
238            ad.log.info(
239                '==================> Cycling WFC %s/%s <==================',
240                attempt+1, cycle)
241            res = self.cycle_wfc(ad)
242            cycle_wfc_summary.append(res)
243            if not res:
244                self._take_bug_report(
245                    self.test_name, begin_time=get_current_epoch_time())
246
247            if (attempt+1) % CALCULATE_EVERY_N_CYCLES == 0 or attempt == cycle - 1:
248                (
249                    res,
250                    lst,
251                    avg_data_call_setup_time) = parse_setup_data_call_on_iwlan(ad)
252
253                ad.log.info('====== Setup data call list ======')
254                print_nested_dict(ad, res)
255
256                ad.log.info('====== Data call setup time list ======')
257                for item in lst:
258                    print_nested_dict(ad, item)
259                    ad.log.info('------------------')
260
261                (
262                    res,
263                    lst,
264                    avg_deactivate_data_call_time) = parse_deactivate_data_call_on_iwlan(ad)
265
266                ad.log.info('====== Deactivate data call list ======')
267                print_nested_dict(ad, res)
268
269                ad.log.info('====== Data call deactivate time list ======')
270                for item in lst:
271                    print_nested_dict(ad, item)
272                    ad.log.info('------------------')
273
274                ad.log.info(
275                    'Average WFC data call setup time: %.2f sec.',
276                    avg_data_call_setup_time)
277                ad.log.info(
278                    'Average WFC deactivate data call time: %.2f sec.',
279                    avg_deactivate_data_call_time)
280
281                try:
282                    fail_rate = cycle_wfc_summary.count(False)/len(
283                        cycle_wfc_summary)
284                    self.log.info(
285                        'Fail rate of cycling WFC: %s/%s (%.2f)',
286                        cycle_wfc_summary.count(False),
287                        len(cycle_wfc_summary),
288                        fail_rate)
289                except Exception as e:
290                    self.log.error('Fail rate of cycling WFC: ERROR (%s)', e)
291
292    @test_tracker_info(uuid="77388597-d764-4db3-be6f-656e56dc253a")
293    @TelephonyBaseTest.tel_test_wrap
294    def test_dds_switch(self):
295        """ Switch DDS to measure DDS switch time and LTE validation time.
296
297        Test steps:
298            1. Switch DDS.
299            2. Parse logcat to calculate DDS switch time and LTE validation time.
300        """
301        ad = self.android_devices[0]
302        cycle = self.dds_switch_test_cycle
303
304        if not getattr(ad, 'dsds', False):
305            raise signals.TestSkip("UE is in single mode. Test will be skipped.")
306
307        dds_switch_summary = []
308        for attempt in range(cycle):
309            self.log.info(
310                '======> DDS switch on LTE %s/%s <======',
311                attempt+1,
312                cycle)
313            if self.switch_dds(ad):
314                dds_switch_summary.append(True)
315            else:
316                dds_switch_summary.append(False)
317                self._take_bug_report(
318                    self.test_name, begin_time=get_current_epoch_time())
319
320            if (attempt+1) % CALCULATE_EVERY_N_CYCLES == 0 or attempt == cycle - 1:
321                (
322                    res,
323                    lst,
324                    avg_data_call_setup_time,
325                    avg_validation_time_on_lte) = parse_setup_data_call(
326                        ad, dds_switch=True)
327
328                ad.log.info('====== Setup data call list ======')
329                print_nested_dict(ad, res)
330
331                ad.log.info('====== Data call setup time list ======')
332                for item in lst:
333                    print_nested_dict(ad, item)
334                    ad.log.info('------------------')
335
336                try:
337                    ad.log.info(
338                        'Average data call setup time on LTE: %.2f sec.',
339                        avg_data_call_setup_time)
340                except Exception as e:
341                    ad.log.error(
342                        'Average data call setup time on LTE: ERROR (%s)', e)
343
344                try:
345                    ad.log.info(
346                        'Average validation time on LTE: %.2f sec.',
347                        avg_validation_time_on_lte)
348                except Exception as e:
349                    ad.log.error('Average validation tim on LTE: ERROR (%s)', e)
350
351                try:
352                    fail_rate = dds_switch_summary.count(False)/len(dds_switch_summary)
353                    self.log.info(
354                        'Fail rate of cycling cellular data on LTE: %s/%s (%.2f)',
355                        dds_switch_summary.count(False),
356                        len(dds_switch_summary),
357                        fail_rate)
358                except Exception as e:
359                    self.log.error(
360                        'Fail rate of cycling cellular data on LTE: ERROR (%s)',
361                        e)
362
363    @test_tracker_info(uuid="ac0b6541-d900-4413-8ccb-839ae998804e")
364    @TelephonyBaseTest.tel_test_wrap
365    def test_http_download(self, method='sl4a'):
366        """HTTP download large file for a long time to ensure there is no issue
367            related to the stability.
368
369        Test steps:
370            1. HTTP download a large file (e.g., 512MB) for a long time
371
372        Returns:
373            False if the download is interrupted. Otherwise True.
374        """
375        ad = self.android_devices[0]
376
377        duration = self.http_download_duration
378
379        start_time = datetime.now()
380
381        result = True
382        while datetime.now() - start_time <= timedelta(seconds=duration):
383            if not active_file_download_test(
384                self.log, ad, file_name='512MB', method=method):
385                result = False
386                self._take_bug_report(
387                    self.test_name, begin_time=get_current_epoch_time())
388        return result