1#!/usr/bin/env python3.4
2#
3#   Copyright 2019 - 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 CBRS devices
18"""
19
20import time
21import collections
22
23from acts.test_decorators import test_tracker_info
24from acts_contrib.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
25from acts_contrib.test_utils.tel.tel_defines import DIRECTION_MOBILE_ORIGINATED
26from acts_contrib.test_utils.tel.tel_defines import DIRECTION_MOBILE_TERMINATED
27from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_BETWEEN_REG_AND_CALL
28from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_IN_CALL
29from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_FOR_CBRS_DATA_SWITCH
30from acts_contrib.test_utils.tel.tel_defines import EventActiveDataSubIdChanged
31from acts_contrib.test_utils.tel.tel_defines import NetworkCallbackAvailable
32from acts_contrib.test_utils.tel.tel_defines import EventNetworkCallback
33from acts_contrib.test_utils.tel.tel_test_utils import get_phone_number
34from acts_contrib.test_utils.tel.tel_test_utils import hangup_call
35from acts_contrib.test_utils.tel.tel_test_utils import hangup_call_by_adb
36from acts_contrib.test_utils.tel.tel_test_utils import initiate_call
37from acts_contrib.test_utils.tel.tel_test_utils import is_phone_not_in_call
38from acts_contrib.test_utils.tel.tel_test_utils import wait_and_answer_call
39from acts_contrib.test_utils.tel.tel_test_utils import is_phone_in_call
40from acts_contrib.test_utils.tel.tel_test_utils import start_qxdm_loggers
41from acts_contrib.test_utils.tel.tel_test_utils import load_scone_cat_simulate_data
42from acts_contrib.test_utils.tel.tel_test_utils import test_data_browsing_success_using_sl4a
43from acts_contrib.test_utils.tel.tel_test_utils import test_data_browsing_failure_using_sl4a
44from acts_contrib.test_utils.tel.tel_test_utils import get_operator_name
45from acts_contrib.test_utils.tel.tel_test_utils import is_current_data_on_cbrs
46from acts_contrib.test_utils.tel.tel_test_utils import toggle_airplane_mode
47from acts_contrib.test_utils.tel.tel_test_utils import STORY_LINE
48from acts_contrib.test_utils.tel.tel_test_utils import get_device_epoch_time
49from acts_contrib.test_utils.tel.tel_test_utils import start_qxdm_logger
50from acts_contrib.test_utils.tel.tel_test_utils import wifi_toggle_state
51from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_3g
52from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_2g
53from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_csfb
54from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_iwlan
55from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_not_iwlan
56from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_volte
57from acts_contrib.test_utils.tel.tel_voice_utils import phone_setup_voice_general
58from acts_contrib.test_utils.tel.tel_voice_utils import phone_setup_volte_for_subscription
59from acts_contrib.test_utils.tel.tel_voice_utils import phone_setup_cdma
60from acts_contrib.test_utils.tel.tel_voice_utils import phone_setup_voice_2g
61from acts_contrib.test_utils.tel.tel_voice_utils import phone_idle_3g
62from acts_contrib.test_utils.tel.tel_voice_utils import phone_idle_2g
63from acts_contrib.test_utils.tel.tel_voice_utils import phone_idle_csfb
64from acts_contrib.test_utils.tel.tel_voice_utils import phone_idle_iwlan
65from acts_contrib.test_utils.tel.tel_voice_utils import phone_idle_not_iwlan
66from acts_contrib.test_utils.tel.tel_voice_utils import phone_idle_volte
67from acts_contrib.test_utils.tel.tel_subscription_utils import get_subid_from_slot_index
68from acts_contrib.test_utils.tel.tel_subscription_utils import get_operatorname_from_slot_index
69from acts_contrib.test_utils.tel.tel_subscription_utils import get_cbrs_and_default_sub_id
70from acts.utils import get_current_epoch_time
71from queue import Empty
72
73WAIT_TIME_BETWEEN_ITERATION = 5
74WAIT_TIME_BETWEEN_HANDOVER = 10
75TIME_PERMITTED_FOR_CBRS_SWITCH = 2
76
77class TelLiveCBRSTest(TelephonyBaseTest):
78    def setup_class(self):
79        super().setup_class()
80        self.number_of_devices = 2
81        self.stress_test_number = self.get_stress_test_number()
82        self.message_lengths = (50, 160, 180)
83        self.long_message_lengths = (800, 1600)
84        self.cbrs_subid = None
85        self.default_subid = None
86        self.switch_time_dict = {}
87        self.stress_test_number = int(
88            self.user_params.get("stress_test_number", 5))
89        self.simulate_weak_signal = None
90        self.simulate_out_of_service = None
91        self.stop_simulation = None
92        self.simulate_out_of_coverage = None
93
94
95    def on_fail(self, test_name, begin_time):
96        if self.stop_simulation is not None:
97            load_scone_cat_simulate_data(self.android_devices[0],self.stop_simulation[0],1)
98        if test_name.startswith('test_stress'):
99            return
100        super().on_fail(test_name, begin_time)
101
102
103    def _cbrs_call_sequence(self, ads, mo_mt,
104                            cbrs_phone_setup_func,
105                            verify_cbrs_initial_idle_func,
106                            verify_data_initial_func,
107                            verify_cbrs_in_call_state_func,
108                            verify_data_in_call_func,
109                            incall_cbrs_setting_check_func,
110                            verify_data_final_func,
111                            verify_cbrs_final_func,
112                            expected_result):
113        """_cbrs_call_sequence
114
115        Args:
116            ads: list of android devices. This list should have 2 ad.
117            mo_mt: indicating this call sequence is MO or MT.
118                Valid input: DIRECTION_MOBILE_ORIGINATED and
119                DIRECTION_MOBILE_TERMINATED.
120
121        Returns:
122            if expected_result is True,
123                Return True if call sequence finish without exception.
124            if expected_result is string,
125                Return True if expected exception happened. Otherwise False.
126
127        """
128
129        class _CBRSCallSequenceException(Exception):
130            pass
131
132        if (len(ads) != 2) or (mo_mt not in [
133                DIRECTION_MOBILE_ORIGINATED, DIRECTION_MOBILE_TERMINATED
134        ]):
135            self.log.error("Invalid parameters.")
136            return False
137
138        self.cbrs_subid, self.default_subid = get_cbrs_and_default_sub_id(ads[0])
139
140        if mo_mt == DIRECTION_MOBILE_ORIGINATED:
141            ad_caller = ads[0]
142            ad_callee = ads[1]
143            caller_number = get_phone_number(self.log, ad_caller)
144            callee_number = get_phone_number(self.log, ad_callee)
145            mo_operator = get_operator_name(ads[0].log, ads[0])
146            mt_operator = get_operator_name(ads[1].log, ads[1])
147        else:
148            ad_caller = ads[1]
149            ad_callee = ads[0]
150            caller_number = get_phone_number(self.log, ad_caller)
151            callee_number = get_phone_number(self.log, ad_callee)
152            mt_operator = get_operator_name(ads[0].log, ads[0])
153            mo_operator = get_operator_name(ads[1].log, ads[1])
154
155        self.log.info("-->Begin cbrs_call_sequence: %s to %s<--",
156                      caller_number, callee_number)
157        self.log.info("--> %s to %s <--", mo_operator, mt_operator)
158
159        try:
160            # Setup
161            if cbrs_phone_setup_func and not cbrs_phone_setup_func():
162                raise _CBRSCallSequenceException("cbrs_phone_setup_func fail.")
163            if not phone_setup_voice_general(self.log, ads[1]):
164                raise _CBRSCallSequenceException(
165                    "phone_setup_voice_general fail.")
166            time.sleep(WAIT_TIME_BETWEEN_REG_AND_CALL)
167
168            # Ensure idle status correct
169            if verify_cbrs_initial_idle_func and not \
170                verify_cbrs_initial_idle_func():
171                raise _CBRSCallSequenceException(
172                    "verify_cbrs_initial_idle_func fail.")
173
174            # Ensure data checks are performed
175            if verify_data_initial_func and not \
176                verify_data_initial_func():
177                raise _CBRSCallSequenceException(
178                    "verify_data_initial_func fail.")
179
180            # Make MO/MT call.
181            if not initiate_call(self.log, ad_caller, callee_number):
182                raise _CBRSCallSequenceException("initiate_call fail.")
183            if not wait_and_answer_call(self.log, ad_callee, caller_number):
184                raise _CBRSCallSequenceException("wait_and_answer_call fail.")
185            time.sleep(WAIT_TIME_FOR_CBRS_DATA_SWITCH)
186
187            # Check state, wait 30 seconds, check again.
188            if (verify_cbrs_in_call_state_func and not
189                    verify_cbrs_in_call_state_func()):
190                raise _CBRSCallSequenceException(
191                    "verify_cbrs_in_call_state_func fail.")
192
193            if is_phone_not_in_call(self.log, ads[1]):
194                raise _CBRSCallSequenceException("PhoneB not in call.")
195
196            # Ensure data checks are performed
197            if verify_data_in_call_func and not \
198                verify_data_in_call_func():
199                raise _CBRSCallSequenceException(
200                    "verify_data_in_call_func fail.")
201
202            time.sleep(WAIT_TIME_IN_CALL)
203
204            if (verify_cbrs_in_call_state_func and not
205                    verify_cbrs_in_call_state_func()):
206                raise _CBRSCallSequenceException(
207                    "verify_cbrs_in_call_state_func fail after 30 seconds.")
208            if is_phone_not_in_call(self.log, ads[1]):
209                raise _CBRSCallSequenceException(
210                    "PhoneB not in call after 30 seconds.")
211
212            # in call change setting and check
213            if (incall_cbrs_setting_check_func and not
214                    incall_cbrs_setting_check_func()):
215                raise _CBRSCallSequenceException(
216                    "incall_cbrs_setting_check_func fail.")
217
218            # Hangup call
219            if is_phone_in_call(self.log, ads[0]):
220                if not hangup_call(self.log, ads[0]):
221                    raise _CBRSCallSequenceException("hangup_call fail.")
222            else:
223                if incall_cbrs_setting_check_func is None:
224                    raise _CBRSCallSequenceException("Unexpected call drop.")
225
226            time.sleep(WAIT_TIME_FOR_CBRS_DATA_SWITCH)
227
228            # Ensure data checks are performed
229            if verify_data_final_func and not \
230                verify_data_final_func():
231                raise _CBRSCallSequenceException(
232                    "verify_data_final_func fail.")
233
234            # Ensure data checks are performed
235            if verify_cbrs_final_func and not \
236                verify_cbrs_final_func():
237                raise _CBRSCallSequenceException(
238                    "verify_cbrs_final_func fail.")
239
240        except _CBRSCallSequenceException as e:
241            if str(e) == expected_result:
242                self.log.info("Expected exception: <%s>, return True.", e)
243                return True
244            else:
245                self.log.info("Unexpected exception: <%s>, return False.", e)
246                return False
247        finally:
248            for ad in ads:
249                if ad.droid.telecomIsInCall():
250                    hangup_call_by_adb(ad)
251        self.log.info("cbrs_call_sequence finished, return %s",
252                      expected_result is True)
253        return (expected_result is True)
254
255
256    def _phone_idle_iwlan(self):
257        return phone_idle_iwlan(self.log, self.android_devices[0])
258
259    def _phone_idle_not_iwlan(self):
260        return phone_idle_not_iwlan(self.log, self.android_devices[0])
261
262    def _phone_idle_volte(self):
263        return phone_idle_volte(self.log, self.android_devices[0])
264
265    def _phone_idle_csfb(self):
266        return phone_idle_csfb(self.log, self.android_devices[0])
267
268    def _phone_idle_3g(self):
269        return phone_idle_3g(self.log, self.android_devices[0])
270
271    def _phone_idle_2g(self):
272        return phone_idle_2g(self.log, self.android_devices[0])
273
274    def _is_phone_in_call_iwlan(self):
275        return is_phone_in_call_iwlan(self.log, self.android_devices[0])
276
277    def _is_phone_in_call_not_iwlan(self):
278        return is_phone_in_call_not_iwlan(self.log, self.android_devices[0])
279
280    def _is_phone_not_in_call(self):
281        if is_phone_in_call(self.log, self.android_devices[0]):
282            self.log.info("{} in call.".format(self.android_devices[0].serial))
283            return False
284        self.log.info("{} not in call.".format(self.android_devices[0].serial))
285        return True
286
287    def _is_phone_in_call_volte(self):
288        return is_phone_in_call_volte(self.log, self.android_devices[0])
289
290    def _is_phone_in_call_3g(self):
291        return is_phone_in_call_3g(self.log, self.android_devices[0])
292
293    def _is_phone_in_call_2g(self):
294        return is_phone_in_call_2g(self.log, self.android_devices[0])
295
296    def _is_phone_in_call_csfb(self):
297        return is_phone_in_call_csfb(self.log, self.android_devices[0])
298
299    def _is_phone_in_call(self):
300        return is_phone_in_call(self.log, self.android_devices[0])
301
302    def _phone_setup_voice_general(self):
303        return phone_setup_voice_general(self.log, self.android_devices[0])
304
305    def _phone_setup_volte(self):
306        return phone_setup_volte_for_subscription(self.log,
307                                                  self.android_devices[0],
308                                                  self.default_subid)
309
310    def _phone_setup_1x(self):
311        return phone_setup_cdma(self.log, self.android_devices[0])
312
313    def _phone_setup_2g(self):
314        return phone_setup_voice_2g(self.log, self.android_devices[0])
315
316
317    def _test_data_browsing_success_using_sl4a(self):
318        return test_data_browsing_success_using_sl4a(self.log,
319                                                     self.android_devices[0])
320
321    def _test_data_browsing_failure_using_sl4a(self):
322        return test_data_browsing_failure_using_sl4a(self.log,
323                                                     self.android_devices[0])
324
325    def _is_current_data_on_cbrs(self):
326        return is_current_data_on_cbrs(self.android_devices[0],
327                                       self.cbrs_subid)
328
329    def _is_current_data_on_default(self):
330        return not is_current_data_on_cbrs(self.android_devices[0],
331                                           self.cbrs_subid)
332
333    def _get_list_average(self, input_list):
334        total_sum = float(sum(input_list))
335        total_count = float(len(input_list))
336        if input_list == []:
337            return False
338        return float(total_sum / total_count)
339
340    def _test_CBRS_switch(self, ads,
341                           verify_data_is_on_cbrs,
342                           verify_data_initial_func,
343                           simulate_signal,
344                           verify_data_in_call_func,
345                           incall_cbrs_setting_check_func,
346                           stop_simulation,
347                           verify_data_final_func,
348                           verify_cbrs_final_func,
349                           expected_result  ):
350        """ Test CBRS Signal in Reduced Signal Strength
351        By Default Data will be on CBRS.
352        Fake Signal strength of CBRS to weak.
353        Verify the data is switched to LTE.
354        Increase the signal strength on CBRS
355        Verify data is back on CBRS.
356        """
357
358
359        class _CBRSCallSequenceException1(Exception):
360            pass
361
362        self.cbrs_subid, self.default_subid = get_cbrs_and_default_sub_id(ads[0])
363
364        # Check if Data is on CBRS
365        try:
366            if verify_data_is_on_cbrs and not \
367                verify_data_is_on_cbrs():
368                raise _CBRSCallSequenceException1(
369                    "verify_cbrs_initial_idle_func fail.")
370
371            # Check if Data works well on CBRS
372            if verify_data_initial_func and not \
373                verify_data_initial_func():
374                raise _CBRSCallSequenceException1(
375                    "verify_data_initial_func fail.")
376
377            # Reduce the Signal strength on CBRS
378            self.log.info("Simulating the Signal using Scone CAT")
379            load_scone_cat_simulate_data(ads[0],simulate_signal[0], 1)
380            time.sleep(15)
381
382            # Ensure data checks are performed and it works on Default Sub Id
383            if verify_data_in_call_func and not \
384                verify_data_in_call_func():
385                raise _CBRSCallSequenceException1("verify_data_on_default SubId fail.")
386
387            # in call change setting and check
388            if (incall_cbrs_setting_check_func and not \
389                incall_cbrs_setting_check_func()):
390                raise _CBRSCallSequenceException1(
391                    "incall_cbrs_setting_check_func fail.")
392
393            # Stop Stimulation and wait for CBRS signal to recover
394            load_scone_cat_simulate_data(ads[0],stop_simulation[0],1)
395            time.sleep(15)
396
397            # Verify Data is Back on CBRS
398            # Ensure data checks are performed
399            if verify_data_final_func and not \
400                verify_data_final_func():
401                raise _CBRSCallSequenceException1(
402                    "verify_data_final_func fail.")
403
404            # Ensure data checks are performed
405            if verify_cbrs_final_func and not \
406                verify_cbrs_final_func():
407                raise _CBRSCallSequenceException1(
408                    "verify_cbrs_final_func fail.")
409
410        except _CBRSCallSequenceException1 as e:
411            if str(e) == expected_result:
412                self.log.info("Expected exception: <%s>, return True.", e)
413                return True
414            else:
415                self.log.info("Unexpected exception: <%s>, return False.", e)
416                return False
417        finally:
418            for ad in ads:
419                if ad.droid.telecomIsInCall():
420                    hangup_call_by_adb(ad)
421        self.log.info("cbrs_data_switch finished, return %s",
422                      expected_result is True)
423        return (expected_result is True)
424
425
426    """ Tests Begin """
427
428
429    @test_tracker_info(uuid="f7a3db92-2f1b-4131-99bc-b323dbce812c")
430    @TelephonyBaseTest.tel_test_wrap
431    def test_cbrs_mo_voice_data_concurrency_lte(self):
432        """ Test CBRS Data with MO Voice Call on LTE
433
434        PhoneA should be on LTE with VoLTE enabled
435        Verify Data Browsing works fine on cbrs before call
436        Call from PhoneA to PhoneB, call should succeed
437        Verify Data Browsing works fine on pSIM during call
438        Terminate call
439        Verify Data Browsing works fine on cbrs after call
440
441        Returns:
442            True if pass; False if fail.
443        """
444        ads = [self.android_devices[0], self.android_devices[1]]
445        result = self._cbrs_call_sequence(
446            ads, DIRECTION_MOBILE_ORIGINATED,
447            self._phone_setup_volte, self._is_current_data_on_cbrs,
448            self._test_data_browsing_success_using_sl4a,
449            self._is_phone_in_call_volte,
450            self._is_current_data_on_default,
451            self._test_data_browsing_success_using_sl4a,
452            self._test_data_browsing_success_using_sl4a,
453            self._is_current_data_on_cbrs, True)
454
455        self.log.info("CBRS MO Result: %s", result)
456        return result
457
458
459    @test_tracker_info(uuid="17acce7a-de9c-4540-b2d3-2c98367a0b4e")
460    @TelephonyBaseTest.tel_test_wrap
461    def test_cbrs_mt_voice_data_concurrency_lte(self):
462        """ Test CBRS Data with MT Voice Call on LTE
463
464        PhoneA should be on LTE with VoLTE enabled
465        Verify Data Browsing works fine on cbrs before call
466        Call from PhoneB to PhoneA, call should succeed
467        Verify Data Browsing works fine on pSIM during call
468        Terminate call
469        Verify Data Browsing works fine on cbrs after call
470
471        Returns:
472            True if pass; False if fail.
473        """
474        ads = [self.android_devices[0], self.android_devices[1]]
475        result = self._cbrs_call_sequence(
476            ads, DIRECTION_MOBILE_TERMINATED,
477            self._phone_setup_volte, self._is_current_data_on_cbrs,
478            self._test_data_browsing_success_using_sl4a,
479            self._is_phone_in_call_volte,
480            self._is_current_data_on_default,
481            self._test_data_browsing_success_using_sl4a,
482            self._test_data_browsing_success_using_sl4a,
483            self._is_current_data_on_cbrs, True)
484
485        self.log.info("CBRS MT Result: %s", result)
486        return result
487
488    @test_tracker_info(uuid="dc2608fc-b99d-419b-8989-e1f8cdeb04da")
489    @TelephonyBaseTest.tel_test_wrap
490    def test_cbrs_mo_voice_data_concurrency_1x(self):
491        """ Test CBRS Data with MO Voice Call on 3G
492
493        PhoneA should be on UMTS
494        Verify Data Browsing works fine on cbrs before call
495        Call from PhoneA to PhoneB, call should succeed
496        Verify Data Browsing works fine on pSIM during call
497        Terminate call
498        Verify Data Browsing works fine on cbrs after call
499
500        Returns:
501            True if pass; False if fail.
502        """
503        ads = [self.android_devices[0], self.android_devices[1]]
504        result = self._cbrs_call_sequence(
505            ads, DIRECTION_MOBILE_ORIGINATED,
506            self._phone_setup_1x, self._is_current_data_on_cbrs,
507            self._test_data_browsing_success_using_sl4a,
508            self._is_phone_in_call_3g,
509            self._is_current_data_on_default,
510            self._test_data_browsing_failure_using_sl4a,
511            self._test_data_browsing_success_using_sl4a,
512            self._is_current_data_on_cbrs, True)
513
514        self.log.info("CBRS MO Result: %s", result)
515        return result
516
517
518    @test_tracker_info(uuid="cd3a6613-ca37-43c7-8364-7e4e627ca558")
519    @TelephonyBaseTest.tel_test_wrap
520    def test_cbrs_mt_voice_data_concurrency_1x(self):
521        """ Test CBRS Data with MT Voice Call on 3G
522
523        PhoneA should be on UMTS
524        Verify Data Browsing works fine on cbrs before call
525        Call from PhoneA to PhoneA, call should succeed
526        Verify Data Browsing works fine on pSIM during call
527        Terminate call
528        Verify Data Browsing works fine on cbrs after call
529
530        Returns:
531            True if pass; False if fail.
532        """
533        ads = [self.android_devices[0], self.android_devices[1]]
534        result = self._cbrs_call_sequence(
535            ads, DIRECTION_MOBILE_TERMINATED,
536            self._phone_setup_1x, self._is_current_data_on_cbrs,
537            self._test_data_browsing_success_using_sl4a,
538            self._is_phone_in_call_3g,
539            self._is_current_data_on_default,
540            self._test_data_browsing_failure_using_sl4a,
541            self._test_data_browsing_success_using_sl4a,
542            self._is_current_data_on_cbrs, True)
543
544        self.log.info("CBRS MT Result: %s", result)
545        return result
546
547
548    def _test_stress_cbrs(self, mo_mt):
549        """ Test CBRS/SSIM VoLTE Stress
550
551            mo_mt: indicating this call sequence is MO or MT.
552                Valid input: DIRECTION_MOBILE_ORIGINATED and
553                DIRECTION_MOBILE_TERMINATED.
554
555        Returns:
556            True if pass; False if fail.
557        """
558        if (mo_mt not in [DIRECTION_MOBILE_ORIGINATED,
559                          DIRECTION_MOBILE_TERMINATED]):
560            self.log.error("Invalid parameters.")
561            return False
562        ads = [self.android_devices[0], self.android_devices[1]]
563        total_iteration = self.stress_test_number
564        fail_count = collections.defaultdict(int)
565        self.cbrs_subid, self.default_subid = get_cbrs_and_default_sub_id(ads[0])
566        self.log.info("Total iteration = %d.", total_iteration)
567        current_iteration = 1
568        for i in range(1, total_iteration + 1):
569            msg = "Stress Call Test Iteration: <%s> / <%s>" % (
570                i, total_iteration)
571            begin_time = get_current_epoch_time()
572            self.log.info(msg)
573            start_qxdm_loggers(self.log, self.android_devices, begin_time)
574            iteration_result = self._cbrs_call_sequence(
575                ads, mo_mt,
576                self._phone_setup_volte,
577                self._is_current_data_on_cbrs,
578                self._test_data_browsing_success_using_sl4a,
579                self._is_phone_in_call_volte,
580                self._is_current_data_on_default,
581                self._test_data_browsing_success_using_sl4a,
582                self._test_data_browsing_success_using_sl4a,
583                self._is_current_data_on_cbrs, True)
584            self.log.info("Result: %s", iteration_result)
585            if iteration_result:
586                self.log.info(">----Iteration : %d/%d succeed.----<",
587                    i, total_iteration)
588            else:
589                fail_count["cbrs_fail"] += 1
590                self.log.error(">----Iteration : %d/%d failed.----<",
591                    i, total_iteration)
592                self._take_bug_report("%s_IterNo_%s" % (self.test_name, i),
593                                      begin_time)
594            current_iteration += 1
595        test_result = True
596        for failure, count in fail_count.items():
597            if count:
598                self.log.error("%s: %s %s failures in %s iterations",
599                               self.test_name, count, failure,
600                               total_iteration)
601                test_result = False
602        return test_result
603
604    @test_tracker_info(uuid="860dc00d-5be5-4cdd-aeb1-a89edfa83342")
605    @TelephonyBaseTest.tel_test_wrap
606    def test_stress_cbrs_mt_calls_lte(self):
607        """ Test SSIM to CBRS stress
608
609        Call from PhoneB to PhoneA
610        Perform CBRS Data checks
611        Repeat above steps
612
613        Returns:
614            True if pass; False if fail.
615        """
616        return self._test_stress_cbrs(DIRECTION_MOBILE_TERMINATED)
617
618    @test_tracker_info(uuid="54366c70-c9cb-4eed-bd1c-a37c83d5c0ae")
619    @TelephonyBaseTest.tel_test_wrap
620    def test_stress_cbrs_mo_calls_lte(self):
621        """ Test CBRS to SSIM stress
622
623        Call from PhoneA to PhoneB
624        Perform CBRS Data checks
625        Repeat above steps
626
627        Returns:
628            True if pass; False if fail.
629        """
630        return self._test_stress_cbrs(DIRECTION_MOBILE_ORIGINATED)
631
632
633    def _cbrs_default_data_switch_timing(self, ad, method, validation=False):
634        result = True
635        callback_key = None
636        if not getattr(ad, "cbrs_droid", None):
637            ad.cbrs_droid, ad.cbrs_ed = ad.get_droid()
638            ad.cbrs_ed.start()
639        else:
640            try:
641                if not ad.cbrs_droid.is_live:
642                    ad.cbrs_droid, ad.cbrs_ed = ad.get_droid()
643                    ad.cbrs_ed.start()
644                else:
645                    ad.cbrs_ed.clear_all_events()
646                ad.cbrs_droid.logI("Start test_stress_cbrsdataswitch test")
647            except Exception:
648                ad.log.info("Create new sl4a session for CBRS")
649                ad.cbrs_droid, ad.cbrs_ed = ad.get_droid()
650                ad.cbrs_ed.start()
651        if validation:
652            ad.cbrs_droid.telephonyStartTrackingActiveDataChange()
653        else:
654            callback_key = ad.cbrs_droid.connectivityRegisterDefaultNetworkCallback()
655            ad.cbrs_droid.connectivityNetworkCallbackStartListeningForEvent(
656                callback_key, NetworkCallbackAvailable)
657        time.sleep(WAIT_TIME_FOR_CBRS_DATA_SWITCH)
658        try:
659            ad.cbrs_ed.clear_events(EventActiveDataSubIdChanged)
660            ad.cbrs_ed.clear_events(EventNetworkCallback)
661            initiate_time_before = get_device_epoch_time(ad)
662            ad.log.debug("initiate_time_before: %d", initiate_time_before)
663            if method == "api":
664                ad.log.info("Setting DDS to default sub %s", self.default_subid)
665                ad.droid.telephonySetPreferredOpportunisticDataSubscription(
666                    2147483647, validation)
667            else:
668                ad.log.info("Making a Voice Call to %s", STORY_LINE)
669                ad.droid.telecomCallNumber(STORY_LINE, False)
670            try:
671                if validation:
672                    events = ad.cbrs_ed.pop_events("(%s)" %
673                        (EventActiveDataSubIdChanged),
674                        WAIT_TIME_FOR_CBRS_DATA_SWITCH)
675                    for event in events:
676                        ad.log.info("Got event %s", event["name"])
677                        if event["name"] == EventActiveDataSubIdChanged:
678                            if event.get("data") and \
679                                    event["data"] == self.default_subid:
680                                ad.log.info("%s has data switched to: %s sub",
681                                            event["name"], event["data"])
682                                initiate_time_after = event["time"]
683                            break
684                else:
685                    events = ad.cbrs_ed.pop_events("(%s)" %
686                       (EventNetworkCallback), WAIT_TIME_FOR_CBRS_DATA_SWITCH)
687                    for event in events:
688                        ad.log.info("Got event %s", event["name"])
689                        if event["name"] == EventNetworkCallback:
690                            if event.get("data") and event["data"].get("networkCallbackEvent"):
691                                ad.log.info("%s %s has data switched to: %s sub",
692                                            event["name"],
693                                            event["data"]["networkCallbackEvent"],
694                                            self.default_subid)
695                                initiate_time_after = event["time"]
696                            break
697            except Empty:
698                ad.log.error("No %s or %s event for DataSwitch received in %d seconds",
699                             EventActiveDataSubIdChanged, EventNetworkCallback,
700                             WAIT_TIME_FOR_CBRS_DATA_SWITCH)
701                return False
702            time_interval = (initiate_time_after - initiate_time_before) / 1000
703            self.switch_time_dict['cbrs_default_switch'].append(time_interval)
704            if time_interval > TIME_PERMITTED_FOR_CBRS_SWITCH:
705                ad.log.error("Time for CBRS->Default - %.2f secs", time_interval)
706                result = False
707            else:
708                ad.log.info("Time for CBRS->Default - %.2f secs", time_interval)
709            time.sleep(WAIT_TIME_BETWEEN_HANDOVER)
710            ad.cbrs_ed.clear_events(EventActiveDataSubIdChanged)
711            ad.cbrs_ed.clear_events(EventNetworkCallback)
712            hangup_time_before = get_device_epoch_time(ad)
713            ad.log.debug("hangup_time_before: %d", hangup_time_before)
714            if method == "api":
715                ad.log.info("Setting DDS to cbrs sub %s", self.cbrs_subid)
716                ad.droid.telephonySetPreferredOpportunisticDataSubscription(
717                    self.cbrs_subid, validation)
718            else:
719                ad.log.info("Ending Call")
720                ad.droid.telecomEndCall()
721            try:
722                if validation:
723                    events = ad.cbrs_ed.pop_events("(%s)" %
724                        (EventActiveDataSubIdChanged), WAIT_TIME_FOR_CBRS_DATA_SWITCH)
725                    for event in events:
726                        ad.log.info("Got event %s", event["name"])
727                        if event["name"] == EventActiveDataSubIdChanged:
728                            if event.get("data") and \
729                                    event["data"] == self.cbrs_subid:
730                                ad.log.info("%s has data switched to: %s sub",
731                                            event["name"], event["data"])
732                                hangup_time_after = event["time"]
733                            break
734                else:
735                    events = ad.cbrs_ed.pop_events("(%s)" %
736                        (EventNetworkCallback), WAIT_TIME_FOR_CBRS_DATA_SWITCH)
737                    for event in events:
738                        ad.log.info("Got event %s", event["name"])
739                        if event["name"] == EventNetworkCallback:
740                            if event.get("data") and event["data"].get("networkCallbackEvent"):
741                                ad.log.info("%s %s has data switched to: %s sub",
742                                            event["name"],
743                                            event["data"]["networkCallbackEvent"],
744                                            self.cbrs_subid)
745                                hangup_time_after = event["time"]
746                            break
747            except Empty:
748                ad.log.error("No %s event for DataSwitch received in %d seconds",
749                             EventActiveDataSubIdChanged,
750                             WAIT_TIME_FOR_CBRS_DATA_SWITCH)
751                return False
752            time_interval = (hangup_time_after - hangup_time_before) / 1000
753            self.switch_time_dict['default_cbrs_switch'].append(time_interval)
754            if time_interval > TIME_PERMITTED_FOR_CBRS_SWITCH:
755                ad.log.error("Time for Default->CBRS - %.2f secs", time_interval)
756                result = False
757            else:
758                ad.log.info("Time for Default->CBRS - %.2f secs", time_interval)
759        except Exception as e:
760            self.log.error("Exception error %s", e)
761            raise
762        finally:
763            if validation:
764                ad.cbrs_droid.telephonyStopTrackingActiveDataChange()
765            elif callback_key:
766                ad.cbrs_droid.connectivityNetworkCallbackStopListeningForEvent(
767                    callback_key, NetworkCallbackAvailable)
768        return result
769
770
771    def _test_stress_cbrsdataswitch_timing(self, ad, method, validation=False):
772        setattr(self, "number_of_devices", 1)
773        ad.adb.shell("pm disable com.google.android.apps.scone")
774        wifi_toggle_state(self.log, ad, True)
775        self.cbrs_subid, self.default_subid = get_cbrs_and_default_sub_id(ad)
776        toggle_airplane_mode(ad.log, ad, new_state=False, strict_checking=False)
777        if self._is_current_data_on_cbrs():
778            ad.log.info("Current Data is on CBRS, proceeding with test")
779        else:
780            ad.log.error("Current Data not on CBRS, forcing it now..")
781            ad.droid.telephonySetPreferredOpportunisticDataSubscription(
782                self.cbrs_subid, False)
783        ad.droid.telephonyUpdateAvailableNetworks(self.cbrs_subid)
784        total_iteration = self.stress_test_number
785        fail_count = collections.defaultdict(int)
786        self.switch_time_dict = {'cbrs_default_switch':[],
787                                 'default_cbrs_switch': []}
788        current_iteration = 1
789        for i in range(1, total_iteration + 1):
790            msg = "Stress CBRS Test Iteration: <%s> / <%s>" % (
791                i, total_iteration)
792            begin_time = get_current_epoch_time()
793            self.log.info(msg)
794            start_qxdm_logger(ad, begin_time)
795            iteration_result = self._cbrs_default_data_switch_timing(
796                ad, method, validation)
797            self.log.info("Result: %s", iteration_result)
798            if iteration_result:
799                self.log.info(">----Iteration : %d/%d succeed.----<",
800                    i, total_iteration)
801            else:
802                fail_count["cbrsdataswitch_fail"] += 1
803                self.log.error(">----Iteration : %d/%d failed.----<",
804                    i, total_iteration)
805                self._take_bug_report("%s_IterNo_%s" % (self.test_name, i),
806                                      begin_time)
807            current_iteration += 1
808            time.sleep(WAIT_TIME_BETWEEN_ITERATION)
809        test_result = True
810        for time_task, time_list in self.switch_time_dict.items():
811            ad.log.info("%s %s", time_task, time_list)
812            avg_time = self._get_list_average(time_list)
813            ad.log.info("Average %s for %d iterations = %.2f seconds",
814                                time_task, self.stress_test_number, avg_time)
815        for failure, count in fail_count.items():
816            if count:
817                self.log.error("%s: %s %s failures in %s iterations",
818                               self.test_name, count, failure,
819                               total_iteration)
820                test_result = False
821        ad.adb.shell("pm enable com.google.android.apps.scone")
822        return test_result
823
824
825    @test_tracker_info(uuid="c66e307b-8456-4a86-af43-d715597ce802")
826    @TelephonyBaseTest.tel_test_wrap
827    def test_stress_cbrsdataswitch_via_api_without_validation(self):
828        """ Test CBRS to Default Data Switch Stress Test
829
830        By default, data is expected to be on CBRS
831        Using TelephonyManagerAPI, switch data to default
832        Measure time to switch
833        Using TelephonyManagerAPI, switch data to cbrs
834        Measure time to switch
835        Switching to be done with validation set to False
836        Repeat above steps
837        Calculate average time to switch
838
839        Returns:
840            True if pass; False if fail.
841        """
842        ad = self.android_devices[0]
843        self._test_stress_cbrsdataswitch_timing(ad, "api", validation=False)
844
845
846    @test_tracker_info(uuid="a0ae7691-4890-4e94-8a01-b54ba5b6f489")
847    @TelephonyBaseTest.tel_test_wrap
848    def test_stress_cbrsdataswitch_via_api_with_validation(self):
849        """ Test CBRS to Default Data Switch Stress Test
850
851        By default, data is expected to be on CBRS
852        Using TelephonyManagerAPI, switch data to default
853        Measure time to switch
854        Using TelephonyManagerAPI, switch data to cbrs
855        Measure time to switch
856        Switching to be done with validation set to True
857        Repeat above steps
858        Calculate average time to switch
859
860        Returns:
861            True if pass; False if fail.
862        """
863        ad = self.android_devices[0]
864        self._test_stress_cbrsdataswitch_timing(ad, "api", validation=True)
865
866
867    @test_tracker_info(uuid="1e69b57a-f2f7-42c6-8389-2194356c599c")
868    @TelephonyBaseTest.tel_test_wrap
869    def test_stress_cbrsdataswitch_via_call(self):
870        """ Test CBRS to Default Data Switch Stress Test
871
872        By default, data is expected to be on CBRS
873        Initiate MO Phone, data will switch to default
874        Measure time to switch
875        Hangup MO Phone, data will switch to cbrs
876        Measure time to switch
877        Repeat above steps
878        Calculate average time to switch
879
880        Returns:
881            True if pass; False if fail.
882        """
883        ad = self.android_devices[0]
884        self._test_stress_cbrsdataswitch_timing(ad, "call")
885
886    @test_tracker_info(uuid="c2f67a50-1d06-4e3f-8935-5360720a55a3")
887    @TelephonyBaseTest.tel_test_wrap
888    def test_cbrs_forced_weak_signalstrength(self):
889        """ Test CBRS Data switch
890
891        Verify Data Browsing works fine on cbrs before weak signal.
892        Using CAT Scone apk, reduce the signal strength.
893        Verify the data switches from CBRS to Psim.
894        Verify Data Browsing works fine on pSIM.
895        Verify Data Browsing works fine on cbrs after simulation is stopped.
896
897        Returns:
898            True if pass; False if fail.
899        """
900        self.simulate_weak_signal = self.user_params.get("simulate_weak_signal")
901        self.stop_simulation = self.user_params.get("stop_simulate")
902        ads = [self.android_devices[0]]
903        result = self._test_CBRS_switch(
904            ads, self._is_current_data_on_cbrs,
905            self._test_data_browsing_success_using_sl4a,
906            self.simulate_weak_signal,
907            self._is_current_data_on_default,
908            self._test_data_browsing_success_using_sl4a,
909            self.stop_simulation,
910            self._test_data_browsing_success_using_sl4a,
911            self._is_current_data_on_cbrs, True)
912
913        self.log.info("CBRS data switch Result: %s", result)
914        return result
915
916    @test_tracker_info(uuid="066ff1e4-c3f7-4834-a86a-6ebd9fbbee06")
917    @TelephonyBaseTest.tel_test_wrap
918    def test_cbrs_forced_out_of_service(self):
919        """ Test CBRS Data switch
920
921        Verify Data Browsing works fine on cbrs before weak signal.
922        Using CAT Scone apk, reduce the signal strength
923            to mimic CBRS signal out of service
924        Verify the data switches from CBRS to Psim.
925        Verify Data Browsing works fine on pSIM.
926        Verify Data Browsing works fine on cbrs after simulation is stopped.
927
928        Returns:
929            True if pass; False if fail.
930        """
931        self.simulate_out_of_service = self.user_params.get("simulate_out_of_service")
932        self.stop_simulation = self.user_params.get("stop_simulate")
933        ads = [self.android_devices[0]]
934        result = self._test_CBRS_switch(
935            ads, self._is_current_data_on_cbrs,
936            self._test_data_browsing_success_using_sl4a,
937            self.simulate_out_of_service,
938            self._is_current_data_on_default,
939            self._test_data_browsing_success_using_sl4a,
940            self.stop_simulation,
941            self._test_data_browsing_success_using_sl4a,
942            self._is_current_data_on_cbrs, True)
943
944        self.log.info("CBRS data switch Result: %s", result)
945        return result
946
947    @test_tracker_info(uuid="8268f88f-c2e3-4000-8f05-97e34e028a92")
948    @TelephonyBaseTest.tel_test_wrap
949    def test_cbrs_forced_out_of_coverage(self):
950        """ Test CBRS Data switch
951
952        Verify Data Browsing works fine on cbrs before weak signal.
953        Using CAT Scone apk, reduce the signal strength
954            to mimic CBRS signal out of coverage.
955        Verify the data switches from CBRS to Psim.
956        Verify Data Browsing works fine on pSIM.
957        Verify Data Browsing works fine on cbrs after simulation is stopped.
958
959        Returns:
960            True if pass; False if fail.
961        """
962        self.simulate_out_of_coverage = self.user_params.get("simulate_out_of_coverage")
963        self.stop_simulation = self.user_params.get("stop_simulate")
964        ads = [self.android_devices[0]]
965        result = self._test_CBRS_switch(
966            ads, self._is_current_data_on_cbrs,
967            self._test_data_browsing_success_using_sl4a,
968            self.simulate_out_of_coverage,
969            self._is_current_data_on_default,
970            self._test_data_browsing_success_using_sl4a,
971            self.stop_simulation,
972            self._test_data_browsing_success_using_sl4a,
973            self._is_current_data_on_cbrs, True)
974
975        self.log.info("CBRS data switch Result: %s", result)
976        return result
977