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