| #!/usr/bin/env python3.4 |
| # |
| # Copyright 2019 - Google |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| """ |
| Test Script for CBRS devices |
| """ |
| |
| import time |
| import collections |
| |
| from acts.test_decorators import test_tracker_info |
| from acts_contrib.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest |
| from acts_contrib.test_utils.tel.tel_defines import DIRECTION_MOBILE_ORIGINATED |
| from acts_contrib.test_utils.tel.tel_defines import DIRECTION_MOBILE_TERMINATED |
| from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_BETWEEN_REG_AND_CALL |
| from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_IN_CALL |
| from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_FOR_CBRS_DATA_SWITCH |
| from acts_contrib.test_utils.tel.tel_defines import EventActiveDataSubIdChanged |
| from acts_contrib.test_utils.tel.tel_defines import NetworkCallbackAvailable |
| from acts_contrib.test_utils.tel.tel_defines import EventNetworkCallback |
| from acts_contrib.test_utils.tel.tel_logging_utils import start_qxdm_logger |
| from acts_contrib.test_utils.tel.tel_logging_utils import start_qxdm_loggers |
| from acts_contrib.test_utils.tel.tel_test_utils import get_phone_number |
| from acts_contrib.test_utils.tel.tel_test_utils import hangup_call |
| from acts_contrib.test_utils.tel.tel_test_utils import hangup_call_by_adb |
| from acts_contrib.test_utils.tel.tel_test_utils import initiate_call |
| from acts_contrib.test_utils.tel.tel_test_utils import is_phone_not_in_call |
| from acts_contrib.test_utils.tel.tel_test_utils import wait_and_answer_call |
| from acts_contrib.test_utils.tel.tel_test_utils import is_phone_in_call |
| from acts_contrib.test_utils.tel.tel_test_utils import load_scone_cat_simulate_data |
| from acts_contrib.test_utils.tel.tel_test_utils import test_data_browsing_success_using_sl4a |
| from acts_contrib.test_utils.tel.tel_test_utils import test_data_browsing_failure_using_sl4a |
| from acts_contrib.test_utils.tel.tel_test_utils import get_operator_name |
| from acts_contrib.test_utils.tel.tel_test_utils import is_current_data_on_cbrs |
| from acts_contrib.test_utils.tel.tel_test_utils import toggle_airplane_mode |
| from acts_contrib.test_utils.tel.tel_test_utils import STORY_LINE |
| from acts_contrib.test_utils.tel.tel_test_utils import get_device_epoch_time |
| from acts_contrib.test_utils.tel.tel_test_utils import wifi_toggle_state |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_3g |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_2g |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_csfb |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_iwlan |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_not_iwlan |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_volte |
| from acts_contrib.test_utils.tel.tel_voice_utils import phone_setup_voice_general |
| from acts_contrib.test_utils.tel.tel_voice_utils import phone_setup_volte_for_subscription |
| from acts_contrib.test_utils.tel.tel_voice_utils import phone_setup_cdma |
| from acts_contrib.test_utils.tel.tel_voice_utils import phone_setup_voice_2g |
| from acts_contrib.test_utils.tel.tel_voice_utils import phone_idle_3g |
| from acts_contrib.test_utils.tel.tel_voice_utils import phone_idle_2g |
| from acts_contrib.test_utils.tel.tel_voice_utils import phone_idle_csfb |
| from acts_contrib.test_utils.tel.tel_voice_utils import phone_idle_iwlan |
| from acts_contrib.test_utils.tel.tel_voice_utils import phone_idle_not_iwlan |
| from acts_contrib.test_utils.tel.tel_voice_utils import phone_idle_volte |
| from acts_contrib.test_utils.tel.tel_subscription_utils import get_cbrs_and_default_sub_id |
| from acts.utils import get_current_epoch_time |
| from queue import Empty |
| |
| WAIT_TIME_BETWEEN_ITERATION = 5 |
| WAIT_TIME_BETWEEN_HANDOVER = 10 |
| TIME_PERMITTED_FOR_CBRS_SWITCH = 2 |
| |
| class TelLiveCBRSTest(TelephonyBaseTest): |
| def setup_class(self): |
| super().setup_class() |
| self.number_of_devices = 2 |
| self.stress_test_number = self.get_stress_test_number() |
| self.message_lengths = (50, 160, 180) |
| self.long_message_lengths = (800, 1600) |
| self.cbrs_subid = None |
| self.default_subid = None |
| self.switch_time_dict = {} |
| self.stress_test_number = int( |
| self.user_params.get("stress_test_number", 5)) |
| self.simulate_weak_signal = None |
| self.simulate_out_of_service = None |
| self.stop_simulation = None |
| self.simulate_out_of_coverage = None |
| |
| |
| def on_fail(self, test_name, begin_time): |
| if self.stop_simulation is not None: |
| load_scone_cat_simulate_data(self.android_devices[0],self.stop_simulation[0],1) |
| if test_name.startswith('test_stress'): |
| return |
| super().on_fail(test_name, begin_time) |
| |
| |
| def _cbrs_call_sequence(self, ads, mo_mt, |
| cbrs_phone_setup_func, |
| verify_cbrs_initial_idle_func, |
| verify_data_initial_func, |
| verify_cbrs_in_call_state_func, |
| verify_data_in_call_func, |
| incall_cbrs_setting_check_func, |
| verify_data_final_func, |
| verify_cbrs_final_func, |
| expected_result): |
| """_cbrs_call_sequence |
| |
| Args: |
| ads: list of android devices. This list should have 2 ad. |
| mo_mt: indicating this call sequence is MO or MT. |
| Valid input: DIRECTION_MOBILE_ORIGINATED and |
| DIRECTION_MOBILE_TERMINATED. |
| |
| Returns: |
| if expected_result is True, |
| Return True if call sequence finish without exception. |
| if expected_result is string, |
| Return True if expected exception happened. Otherwise False. |
| |
| """ |
| |
| class _CBRSCallSequenceException(Exception): |
| pass |
| |
| if (len(ads) != 2) or (mo_mt not in [ |
| DIRECTION_MOBILE_ORIGINATED, DIRECTION_MOBILE_TERMINATED |
| ]): |
| self.log.error("Invalid parameters.") |
| return False |
| |
| self.cbrs_subid, self.default_subid = get_cbrs_and_default_sub_id(ads[0]) |
| |
| if mo_mt == DIRECTION_MOBILE_ORIGINATED: |
| ad_caller = ads[0] |
| ad_callee = ads[1] |
| caller_number = get_phone_number(self.log, ad_caller) |
| callee_number = get_phone_number(self.log, ad_callee) |
| mo_operator = get_operator_name(ads[0].log, ads[0]) |
| mt_operator = get_operator_name(ads[1].log, ads[1]) |
| else: |
| ad_caller = ads[1] |
| ad_callee = ads[0] |
| caller_number = get_phone_number(self.log, ad_caller) |
| callee_number = get_phone_number(self.log, ad_callee) |
| mt_operator = get_operator_name(ads[0].log, ads[0]) |
| mo_operator = get_operator_name(ads[1].log, ads[1]) |
| |
| self.log.info("-->Begin cbrs_call_sequence: %s to %s<--", |
| caller_number, callee_number) |
| self.log.info("--> %s to %s <--", mo_operator, mt_operator) |
| |
| try: |
| # Setup |
| if cbrs_phone_setup_func and not cbrs_phone_setup_func(): |
| raise _CBRSCallSequenceException("cbrs_phone_setup_func fail.") |
| if not phone_setup_voice_general(self.log, ads[1]): |
| raise _CBRSCallSequenceException( |
| "phone_setup_voice_general fail.") |
| time.sleep(WAIT_TIME_BETWEEN_REG_AND_CALL) |
| |
| # Ensure idle status correct |
| if verify_cbrs_initial_idle_func and not \ |
| verify_cbrs_initial_idle_func(): |
| raise _CBRSCallSequenceException( |
| "verify_cbrs_initial_idle_func fail.") |
| |
| # Ensure data checks are performed |
| if verify_data_initial_func and not \ |
| verify_data_initial_func(): |
| raise _CBRSCallSequenceException( |
| "verify_data_initial_func fail.") |
| |
| # Make MO/MT call. |
| if not initiate_call(self.log, ad_caller, callee_number): |
| raise _CBRSCallSequenceException("initiate_call fail.") |
| if not wait_and_answer_call(self.log, ad_callee, caller_number): |
| raise _CBRSCallSequenceException("wait_and_answer_call fail.") |
| time.sleep(WAIT_TIME_FOR_CBRS_DATA_SWITCH) |
| |
| # Check state, wait 30 seconds, check again. |
| if (verify_cbrs_in_call_state_func and not |
| verify_cbrs_in_call_state_func()): |
| raise _CBRSCallSequenceException( |
| "verify_cbrs_in_call_state_func fail.") |
| |
| if is_phone_not_in_call(self.log, ads[1]): |
| raise _CBRSCallSequenceException("PhoneB not in call.") |
| |
| # Ensure data checks are performed |
| if verify_data_in_call_func and not \ |
| verify_data_in_call_func(): |
| raise _CBRSCallSequenceException( |
| "verify_data_in_call_func fail.") |
| |
| time.sleep(WAIT_TIME_IN_CALL) |
| |
| if (verify_cbrs_in_call_state_func and not |
| verify_cbrs_in_call_state_func()): |
| raise _CBRSCallSequenceException( |
| "verify_cbrs_in_call_state_func fail after 30 seconds.") |
| if is_phone_not_in_call(self.log, ads[1]): |
| raise _CBRSCallSequenceException( |
| "PhoneB not in call after 30 seconds.") |
| |
| # in call change setting and check |
| if (incall_cbrs_setting_check_func and not |
| incall_cbrs_setting_check_func()): |
| raise _CBRSCallSequenceException( |
| "incall_cbrs_setting_check_func fail.") |
| |
| # Hangup call |
| if is_phone_in_call(self.log, ads[0]): |
| if not hangup_call(self.log, ads[0]): |
| raise _CBRSCallSequenceException("hangup_call fail.") |
| else: |
| if incall_cbrs_setting_check_func is None: |
| raise _CBRSCallSequenceException("Unexpected call drop.") |
| |
| time.sleep(WAIT_TIME_FOR_CBRS_DATA_SWITCH) |
| |
| # Ensure data checks are performed |
| if verify_data_final_func and not \ |
| verify_data_final_func(): |
| raise _CBRSCallSequenceException( |
| "verify_data_final_func fail.") |
| |
| # Ensure data checks are performed |
| if verify_cbrs_final_func and not \ |
| verify_cbrs_final_func(): |
| raise _CBRSCallSequenceException( |
| "verify_cbrs_final_func fail.") |
| |
| except _CBRSCallSequenceException as e: |
| if str(e) == expected_result: |
| self.log.info("Expected exception: <%s>, return True.", e) |
| return True |
| else: |
| self.log.info("Unexpected exception: <%s>, return False.", e) |
| return False |
| finally: |
| for ad in ads: |
| if ad.droid.telecomIsInCall(): |
| hangup_call_by_adb(ad) |
| self.log.info("cbrs_call_sequence finished, return %s", |
| expected_result is True) |
| return (expected_result is True) |
| |
| |
| def _phone_idle_iwlan(self): |
| return phone_idle_iwlan(self.log, self.android_devices[0]) |
| |
| def _phone_idle_not_iwlan(self): |
| return phone_idle_not_iwlan(self.log, self.android_devices[0]) |
| |
| def _phone_idle_volte(self): |
| return phone_idle_volte(self.log, self.android_devices[0]) |
| |
| def _phone_idle_csfb(self): |
| return phone_idle_csfb(self.log, self.android_devices[0]) |
| |
| def _phone_idle_3g(self): |
| return phone_idle_3g(self.log, self.android_devices[0]) |
| |
| def _phone_idle_2g(self): |
| return phone_idle_2g(self.log, self.android_devices[0]) |
| |
| def _is_phone_in_call_iwlan(self): |
| return is_phone_in_call_iwlan(self.log, self.android_devices[0]) |
| |
| def _is_phone_in_call_not_iwlan(self): |
| return is_phone_in_call_not_iwlan(self.log, self.android_devices[0]) |
| |
| def _is_phone_not_in_call(self): |
| if is_phone_in_call(self.log, self.android_devices[0]): |
| self.log.info("{} in call.".format(self.android_devices[0].serial)) |
| return False |
| self.log.info("{} not in call.".format(self.android_devices[0].serial)) |
| return True |
| |
| def _is_phone_in_call_volte(self): |
| return is_phone_in_call_volte(self.log, self.android_devices[0]) |
| |
| def _is_phone_in_call_3g(self): |
| return is_phone_in_call_3g(self.log, self.android_devices[0]) |
| |
| def _is_phone_in_call_2g(self): |
| return is_phone_in_call_2g(self.log, self.android_devices[0]) |
| |
| def _is_phone_in_call_csfb(self): |
| return is_phone_in_call_csfb(self.log, self.android_devices[0]) |
| |
| def _is_phone_in_call(self): |
| return is_phone_in_call(self.log, self.android_devices[0]) |
| |
| def _phone_setup_voice_general(self): |
| return phone_setup_voice_general(self.log, self.android_devices[0]) |
| |
| def _phone_setup_volte(self): |
| return phone_setup_volte_for_subscription(self.log, |
| self.android_devices[0], |
| self.default_subid) |
| |
| def _phone_setup_1x(self): |
| return phone_setup_cdma(self.log, self.android_devices[0]) |
| |
| def _phone_setup_2g(self): |
| return phone_setup_voice_2g(self.log, self.android_devices[0]) |
| |
| |
| def _test_data_browsing_success_using_sl4a(self): |
| return test_data_browsing_success_using_sl4a(self.log, |
| self.android_devices[0]) |
| |
| def _test_data_browsing_failure_using_sl4a(self): |
| return test_data_browsing_failure_using_sl4a(self.log, |
| self.android_devices[0]) |
| |
| def _is_current_data_on_cbrs(self): |
| return is_current_data_on_cbrs(self.android_devices[0], |
| self.cbrs_subid) |
| |
| def _is_current_data_on_default(self): |
| return not is_current_data_on_cbrs(self.android_devices[0], |
| self.cbrs_subid) |
| |
| def _get_list_average(self, input_list): |
| total_sum = float(sum(input_list)) |
| total_count = float(len(input_list)) |
| if input_list == []: |
| return False |
| return float(total_sum / total_count) |
| |
| def _test_CBRS_switch(self, ads, |
| verify_data_is_on_cbrs, |
| verify_data_initial_func, |
| simulate_signal, |
| verify_data_in_call_func, |
| incall_cbrs_setting_check_func, |
| stop_simulation, |
| verify_data_final_func, |
| verify_cbrs_final_func, |
| expected_result ): |
| """ Test CBRS Signal in Reduced Signal Strength |
| By Default Data will be on CBRS. |
| Fake Signal strength of CBRS to weak. |
| Verify the data is switched to LTE. |
| Increase the signal strength on CBRS |
| Verify data is back on CBRS. |
| """ |
| |
| |
| class _CBRSCallSequenceException1(Exception): |
| pass |
| |
| self.cbrs_subid, self.default_subid = get_cbrs_and_default_sub_id(ads[0]) |
| |
| # Check if Data is on CBRS |
| try: |
| if verify_data_is_on_cbrs and not \ |
| verify_data_is_on_cbrs(): |
| raise _CBRSCallSequenceException1( |
| "verify_cbrs_initial_idle_func fail.") |
| |
| # Check if Data works well on CBRS |
| if verify_data_initial_func and not \ |
| verify_data_initial_func(): |
| raise _CBRSCallSequenceException1( |
| "verify_data_initial_func fail.") |
| |
| # Reduce the Signal strength on CBRS |
| self.log.info("Simulating the Signal using Scone CAT") |
| load_scone_cat_simulate_data(ads[0],simulate_signal[0], 1) |
| time.sleep(15) |
| |
| # Ensure data checks are performed and it works on Default Sub Id |
| if verify_data_in_call_func and not \ |
| verify_data_in_call_func(): |
| raise _CBRSCallSequenceException1("verify_data_on_default SubId fail.") |
| |
| # in call change setting and check |
| if (incall_cbrs_setting_check_func and not \ |
| incall_cbrs_setting_check_func()): |
| raise _CBRSCallSequenceException1( |
| "incall_cbrs_setting_check_func fail.") |
| |
| # Stop Stimulation and wait for CBRS signal to recover |
| load_scone_cat_simulate_data(ads[0],stop_simulation[0],1) |
| time.sleep(15) |
| |
| # Verify Data is Back on CBRS |
| # Ensure data checks are performed |
| if verify_data_final_func and not \ |
| verify_data_final_func(): |
| raise _CBRSCallSequenceException1( |
| "verify_data_final_func fail.") |
| |
| # Ensure data checks are performed |
| if verify_cbrs_final_func and not \ |
| verify_cbrs_final_func(): |
| raise _CBRSCallSequenceException1( |
| "verify_cbrs_final_func fail.") |
| |
| except _CBRSCallSequenceException1 as e: |
| if str(e) == expected_result: |
| self.log.info("Expected exception: <%s>, return True.", e) |
| return True |
| else: |
| self.log.info("Unexpected exception: <%s>, return False.", e) |
| return False |
| finally: |
| for ad in ads: |
| if ad.droid.telecomIsInCall(): |
| hangup_call_by_adb(ad) |
| self.log.info("cbrs_data_switch finished, return %s", |
| expected_result is True) |
| return (expected_result is True) |
| |
| |
| """ Tests Begin """ |
| |
| |
| @test_tracker_info(uuid="f7a3db92-2f1b-4131-99bc-b323dbce812c") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_cbrs_mo_voice_data_concurrency_lte(self): |
| """ Test CBRS Data with MO Voice Call on LTE |
| |
| PhoneA should be on LTE with VoLTE enabled |
| Verify Data Browsing works fine on cbrs before call |
| Call from PhoneA to PhoneB, call should succeed |
| Verify Data Browsing works fine on pSIM during call |
| Terminate call |
| Verify Data Browsing works fine on cbrs after call |
| |
| Returns: |
| True if pass; False if fail. |
| """ |
| ads = [self.android_devices[0], self.android_devices[1]] |
| result = self._cbrs_call_sequence( |
| ads, DIRECTION_MOBILE_ORIGINATED, |
| self._phone_setup_volte, self._is_current_data_on_cbrs, |
| self._test_data_browsing_success_using_sl4a, |
| self._is_phone_in_call_volte, |
| self._is_current_data_on_default, |
| self._test_data_browsing_success_using_sl4a, |
| self._test_data_browsing_success_using_sl4a, |
| self._is_current_data_on_cbrs, True) |
| |
| self.log.info("CBRS MO Result: %s", result) |
| return result |
| |
| |
| @test_tracker_info(uuid="17acce7a-de9c-4540-b2d3-2c98367a0b4e") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_cbrs_mt_voice_data_concurrency_lte(self): |
| """ Test CBRS Data with MT Voice Call on LTE |
| |
| PhoneA should be on LTE with VoLTE enabled |
| Verify Data Browsing works fine on cbrs before call |
| Call from PhoneB to PhoneA, call should succeed |
| Verify Data Browsing works fine on pSIM during call |
| Terminate call |
| Verify Data Browsing works fine on cbrs after call |
| |
| Returns: |
| True if pass; False if fail. |
| """ |
| ads = [self.android_devices[0], self.android_devices[1]] |
| result = self._cbrs_call_sequence( |
| ads, DIRECTION_MOBILE_TERMINATED, |
| self._phone_setup_volte, self._is_current_data_on_cbrs, |
| self._test_data_browsing_success_using_sl4a, |
| self._is_phone_in_call_volte, |
| self._is_current_data_on_default, |
| self._test_data_browsing_success_using_sl4a, |
| self._test_data_browsing_success_using_sl4a, |
| self._is_current_data_on_cbrs, True) |
| |
| self.log.info("CBRS MT Result: %s", result) |
| return result |
| |
| @test_tracker_info(uuid="dc2608fc-b99d-419b-8989-e1f8cdeb04da") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_cbrs_mo_voice_data_concurrency_1x(self): |
| """ Test CBRS Data with MO Voice Call on 3G |
| |
| PhoneA should be on UMTS |
| Verify Data Browsing works fine on cbrs before call |
| Call from PhoneA to PhoneB, call should succeed |
| Verify Data Browsing works fine on pSIM during call |
| Terminate call |
| Verify Data Browsing works fine on cbrs after call |
| |
| Returns: |
| True if pass; False if fail. |
| """ |
| ads = [self.android_devices[0], self.android_devices[1]] |
| result = self._cbrs_call_sequence( |
| ads, DIRECTION_MOBILE_ORIGINATED, |
| self._phone_setup_1x, self._is_current_data_on_cbrs, |
| self._test_data_browsing_success_using_sl4a, |
| self._is_phone_in_call_3g, |
| self._is_current_data_on_default, |
| self._test_data_browsing_failure_using_sl4a, |
| self._test_data_browsing_success_using_sl4a, |
| self._is_current_data_on_cbrs, True) |
| |
| self.log.info("CBRS MO Result: %s", result) |
| return result |
| |
| |
| @test_tracker_info(uuid="cd3a6613-ca37-43c7-8364-7e4e627ca558") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_cbrs_mt_voice_data_concurrency_1x(self): |
| """ Test CBRS Data with MT Voice Call on 3G |
| |
| PhoneA should be on UMTS |
| Verify Data Browsing works fine on cbrs before call |
| Call from PhoneA to PhoneA, call should succeed |
| Verify Data Browsing works fine on pSIM during call |
| Terminate call |
| Verify Data Browsing works fine on cbrs after call |
| |
| Returns: |
| True if pass; False if fail. |
| """ |
| ads = [self.android_devices[0], self.android_devices[1]] |
| result = self._cbrs_call_sequence( |
| ads, DIRECTION_MOBILE_TERMINATED, |
| self._phone_setup_1x, self._is_current_data_on_cbrs, |
| self._test_data_browsing_success_using_sl4a, |
| self._is_phone_in_call_3g, |
| self._is_current_data_on_default, |
| self._test_data_browsing_failure_using_sl4a, |
| self._test_data_browsing_success_using_sl4a, |
| self._is_current_data_on_cbrs, True) |
| |
| self.log.info("CBRS MT Result: %s", result) |
| return result |
| |
| |
| def _test_stress_cbrs(self, mo_mt): |
| """ Test CBRS/SSIM VoLTE Stress |
| |
| mo_mt: indicating this call sequence is MO or MT. |
| Valid input: DIRECTION_MOBILE_ORIGINATED and |
| DIRECTION_MOBILE_TERMINATED. |
| |
| Returns: |
| True if pass; False if fail. |
| """ |
| if (mo_mt not in [DIRECTION_MOBILE_ORIGINATED, |
| DIRECTION_MOBILE_TERMINATED]): |
| self.log.error("Invalid parameters.") |
| return False |
| ads = [self.android_devices[0], self.android_devices[1]] |
| total_iteration = self.stress_test_number |
| fail_count = collections.defaultdict(int) |
| self.cbrs_subid, self.default_subid = get_cbrs_and_default_sub_id(ads[0]) |
| self.log.info("Total iteration = %d.", total_iteration) |
| current_iteration = 1 |
| for i in range(1, total_iteration + 1): |
| msg = "Stress Call Test Iteration: <%s> / <%s>" % ( |
| i, total_iteration) |
| begin_time = get_current_epoch_time() |
| self.log.info(msg) |
| start_qxdm_loggers(self.log, self.android_devices, begin_time) |
| iteration_result = self._cbrs_call_sequence( |
| ads, mo_mt, |
| self._phone_setup_volte, |
| self._is_current_data_on_cbrs, |
| self._test_data_browsing_success_using_sl4a, |
| self._is_phone_in_call_volte, |
| self._is_current_data_on_default, |
| self._test_data_browsing_success_using_sl4a, |
| self._test_data_browsing_success_using_sl4a, |
| self._is_current_data_on_cbrs, True) |
| self.log.info("Result: %s", iteration_result) |
| if iteration_result: |
| self.log.info(">----Iteration : %d/%d succeed.----<", |
| i, total_iteration) |
| else: |
| fail_count["cbrs_fail"] += 1 |
| self.log.error(">----Iteration : %d/%d failed.----<", |
| i, total_iteration) |
| self._take_bug_report("%s_IterNo_%s" % (self.test_name, i), |
| begin_time) |
| current_iteration += 1 |
| test_result = True |
| for failure, count in fail_count.items(): |
| if count: |
| self.log.error("%s: %s %s failures in %s iterations", |
| self.test_name, count, failure, |
| total_iteration) |
| test_result = False |
| return test_result |
| |
| @test_tracker_info(uuid="860dc00d-5be5-4cdd-aeb1-a89edfa83342") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_stress_cbrs_mt_calls_lte(self): |
| """ Test SSIM to CBRS stress |
| |
| Call from PhoneB to PhoneA |
| Perform CBRS Data checks |
| Repeat above steps |
| |
| Returns: |
| True if pass; False if fail. |
| """ |
| return self._test_stress_cbrs(DIRECTION_MOBILE_TERMINATED) |
| |
| @test_tracker_info(uuid="54366c70-c9cb-4eed-bd1c-a37c83d5c0ae") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_stress_cbrs_mo_calls_lte(self): |
| """ Test CBRS to SSIM stress |
| |
| Call from PhoneA to PhoneB |
| Perform CBRS Data checks |
| Repeat above steps |
| |
| Returns: |
| True if pass; False if fail. |
| """ |
| return self._test_stress_cbrs(DIRECTION_MOBILE_ORIGINATED) |
| |
| |
| def _cbrs_default_data_switch_timing(self, ad, method, validation=False): |
| result = True |
| callback_key = None |
| if not getattr(ad, "cbrs_droid", None): |
| ad.cbrs_droid, ad.cbrs_ed = ad.get_droid() |
| ad.cbrs_ed.start() |
| else: |
| try: |
| if not ad.cbrs_droid.is_live: |
| ad.cbrs_droid, ad.cbrs_ed = ad.get_droid() |
| ad.cbrs_ed.start() |
| else: |
| ad.cbrs_ed.clear_all_events() |
| ad.cbrs_droid.logI("Start test_stress_cbrsdataswitch test") |
| except Exception: |
| ad.log.info("Create new sl4a session for CBRS") |
| ad.cbrs_droid, ad.cbrs_ed = ad.get_droid() |
| ad.cbrs_ed.start() |
| if validation: |
| ad.cbrs_droid.telephonyStartTrackingActiveDataChange() |
| else: |
| callback_key = ad.cbrs_droid.connectivityRegisterDefaultNetworkCallback() |
| ad.cbrs_droid.connectivityNetworkCallbackStartListeningForEvent( |
| callback_key, NetworkCallbackAvailable) |
| time.sleep(WAIT_TIME_FOR_CBRS_DATA_SWITCH) |
| try: |
| ad.cbrs_ed.clear_events(EventActiveDataSubIdChanged) |
| ad.cbrs_ed.clear_events(EventNetworkCallback) |
| initiate_time_before = get_device_epoch_time(ad) |
| ad.log.debug("initiate_time_before: %d", initiate_time_before) |
| if method == "api": |
| ad.log.info("Setting DDS to default sub %s", self.default_subid) |
| ad.droid.telephonySetPreferredOpportunisticDataSubscription( |
| 2147483647, validation) |
| else: |
| ad.log.info("Making a Voice Call to %s", STORY_LINE) |
| ad.droid.telecomCallNumber(STORY_LINE, False) |
| try: |
| if validation: |
| events = ad.cbrs_ed.pop_events("(%s)" % |
| (EventActiveDataSubIdChanged), |
| WAIT_TIME_FOR_CBRS_DATA_SWITCH) |
| for event in events: |
| ad.log.info("Got event %s", event["name"]) |
| if event["name"] == EventActiveDataSubIdChanged: |
| if event.get("data") and \ |
| event["data"] == self.default_subid: |
| ad.log.info("%s has data switched to: %s sub", |
| event["name"], event["data"]) |
| initiate_time_after = event["time"] |
| break |
| else: |
| events = ad.cbrs_ed.pop_events("(%s)" % |
| (EventNetworkCallback), WAIT_TIME_FOR_CBRS_DATA_SWITCH) |
| for event in events: |
| ad.log.info("Got event %s", event["name"]) |
| if event["name"] == EventNetworkCallback: |
| if event.get("data") and event["data"].get("networkCallbackEvent"): |
| ad.log.info("%s %s has data switched to: %s sub", |
| event["name"], |
| event["data"]["networkCallbackEvent"], |
| self.default_subid) |
| initiate_time_after = event["time"] |
| break |
| except Empty: |
| ad.log.error("No %s or %s event for DataSwitch received in %d seconds", |
| EventActiveDataSubIdChanged, EventNetworkCallback, |
| WAIT_TIME_FOR_CBRS_DATA_SWITCH) |
| return False |
| time_interval = (initiate_time_after - initiate_time_before) / 1000 |
| self.switch_time_dict['cbrs_default_switch'].append(time_interval) |
| if time_interval > TIME_PERMITTED_FOR_CBRS_SWITCH: |
| ad.log.error("Time for CBRS->Default - %.2f secs", time_interval) |
| result = False |
| else: |
| ad.log.info("Time for CBRS->Default - %.2f secs", time_interval) |
| time.sleep(WAIT_TIME_BETWEEN_HANDOVER) |
| ad.cbrs_ed.clear_events(EventActiveDataSubIdChanged) |
| ad.cbrs_ed.clear_events(EventNetworkCallback) |
| hangup_time_before = get_device_epoch_time(ad) |
| ad.log.debug("hangup_time_before: %d", hangup_time_before) |
| if method == "api": |
| ad.log.info("Setting DDS to cbrs sub %s", self.cbrs_subid) |
| ad.droid.telephonySetPreferredOpportunisticDataSubscription( |
| self.cbrs_subid, validation) |
| else: |
| ad.log.info("Ending Call") |
| ad.droid.telecomEndCall() |
| try: |
| if validation: |
| events = ad.cbrs_ed.pop_events("(%s)" % |
| (EventActiveDataSubIdChanged), WAIT_TIME_FOR_CBRS_DATA_SWITCH) |
| for event in events: |
| ad.log.info("Got event %s", event["name"]) |
| if event["name"] == EventActiveDataSubIdChanged: |
| if event.get("data") and \ |
| event["data"] == self.cbrs_subid: |
| ad.log.info("%s has data switched to: %s sub", |
| event["name"], event["data"]) |
| hangup_time_after = event["time"] |
| break |
| else: |
| events = ad.cbrs_ed.pop_events("(%s)" % |
| (EventNetworkCallback), WAIT_TIME_FOR_CBRS_DATA_SWITCH) |
| for event in events: |
| ad.log.info("Got event %s", event["name"]) |
| if event["name"] == EventNetworkCallback: |
| if event.get("data") and event["data"].get("networkCallbackEvent"): |
| ad.log.info("%s %s has data switched to: %s sub", |
| event["name"], |
| event["data"]["networkCallbackEvent"], |
| self.cbrs_subid) |
| hangup_time_after = event["time"] |
| break |
| except Empty: |
| ad.log.error("No %s event for DataSwitch received in %d seconds", |
| EventActiveDataSubIdChanged, |
| WAIT_TIME_FOR_CBRS_DATA_SWITCH) |
| return False |
| time_interval = (hangup_time_after - hangup_time_before) / 1000 |
| self.switch_time_dict['default_cbrs_switch'].append(time_interval) |
| if time_interval > TIME_PERMITTED_FOR_CBRS_SWITCH: |
| ad.log.error("Time for Default->CBRS - %.2f secs", time_interval) |
| result = False |
| else: |
| ad.log.info("Time for Default->CBRS - %.2f secs", time_interval) |
| except Exception as e: |
| self.log.error("Exception error %s", e) |
| raise |
| finally: |
| if validation: |
| ad.cbrs_droid.telephonyStopTrackingActiveDataChange() |
| elif callback_key: |
| ad.cbrs_droid.connectivityNetworkCallbackStopListeningForEvent( |
| callback_key, NetworkCallbackAvailable) |
| return result |
| |
| |
| def _test_stress_cbrsdataswitch_timing(self, ad, method, validation=False): |
| setattr(self, "number_of_devices", 1) |
| ad.adb.shell("pm disable com.google.android.apps.scone") |
| wifi_toggle_state(self.log, ad, True) |
| self.cbrs_subid, self.default_subid = get_cbrs_and_default_sub_id(ad) |
| toggle_airplane_mode(ad.log, ad, new_state=False, strict_checking=False) |
| if self._is_current_data_on_cbrs(): |
| ad.log.info("Current Data is on CBRS, proceeding with test") |
| else: |
| ad.log.error("Current Data not on CBRS, forcing it now..") |
| ad.droid.telephonySetPreferredOpportunisticDataSubscription( |
| self.cbrs_subid, False) |
| ad.droid.telephonyUpdateAvailableNetworks(self.cbrs_subid) |
| total_iteration = self.stress_test_number |
| fail_count = collections.defaultdict(int) |
| self.switch_time_dict = {'cbrs_default_switch':[], |
| 'default_cbrs_switch': []} |
| current_iteration = 1 |
| for i in range(1, total_iteration + 1): |
| msg = "Stress CBRS Test Iteration: <%s> / <%s>" % ( |
| i, total_iteration) |
| begin_time = get_current_epoch_time() |
| self.log.info(msg) |
| start_qxdm_logger(ad, begin_time) |
| iteration_result = self._cbrs_default_data_switch_timing( |
| ad, method, validation) |
| self.log.info("Result: %s", iteration_result) |
| if iteration_result: |
| self.log.info(">----Iteration : %d/%d succeed.----<", |
| i, total_iteration) |
| else: |
| fail_count["cbrsdataswitch_fail"] += 1 |
| self.log.error(">----Iteration : %d/%d failed.----<", |
| i, total_iteration) |
| self._take_bug_report("%s_IterNo_%s" % (self.test_name, i), |
| begin_time) |
| current_iteration += 1 |
| time.sleep(WAIT_TIME_BETWEEN_ITERATION) |
| test_result = True |
| for time_task, time_list in self.switch_time_dict.items(): |
| ad.log.info("%s %s", time_task, time_list) |
| avg_time = self._get_list_average(time_list) |
| ad.log.info("Average %s for %d iterations = %.2f seconds", |
| time_task, self.stress_test_number, avg_time) |
| for failure, count in fail_count.items(): |
| if count: |
| self.log.error("%s: %s %s failures in %s iterations", |
| self.test_name, count, failure, |
| total_iteration) |
| test_result = False |
| ad.adb.shell("pm enable com.google.android.apps.scone") |
| return test_result |
| |
| |
| @test_tracker_info(uuid="c66e307b-8456-4a86-af43-d715597ce802") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_stress_cbrsdataswitch_via_api_without_validation(self): |
| """ Test CBRS to Default Data Switch Stress Test |
| |
| By default, data is expected to be on CBRS |
| Using TelephonyManagerAPI, switch data to default |
| Measure time to switch |
| Using TelephonyManagerAPI, switch data to cbrs |
| Measure time to switch |
| Switching to be done with validation set to False |
| Repeat above steps |
| Calculate average time to switch |
| |
| Returns: |
| True if pass; False if fail. |
| """ |
| ad = self.android_devices[0] |
| self._test_stress_cbrsdataswitch_timing(ad, "api", validation=False) |
| |
| |
| @test_tracker_info(uuid="a0ae7691-4890-4e94-8a01-b54ba5b6f489") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_stress_cbrsdataswitch_via_api_with_validation(self): |
| """ Test CBRS to Default Data Switch Stress Test |
| |
| By default, data is expected to be on CBRS |
| Using TelephonyManagerAPI, switch data to default |
| Measure time to switch |
| Using TelephonyManagerAPI, switch data to cbrs |
| Measure time to switch |
| Switching to be done with validation set to True |
| Repeat above steps |
| Calculate average time to switch |
| |
| Returns: |
| True if pass; False if fail. |
| """ |
| ad = self.android_devices[0] |
| self._test_stress_cbrsdataswitch_timing(ad, "api", validation=True) |
| |
| |
| @test_tracker_info(uuid="1e69b57a-f2f7-42c6-8389-2194356c599c") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_stress_cbrsdataswitch_via_call(self): |
| """ Test CBRS to Default Data Switch Stress Test |
| |
| By default, data is expected to be on CBRS |
| Initiate MO Phone, data will switch to default |
| Measure time to switch |
| Hangup MO Phone, data will switch to cbrs |
| Measure time to switch |
| Repeat above steps |
| Calculate average time to switch |
| |
| Returns: |
| True if pass; False if fail. |
| """ |
| ad = self.android_devices[0] |
| self._test_stress_cbrsdataswitch_timing(ad, "call") |
| |
| @test_tracker_info(uuid="c2f67a50-1d06-4e3f-8935-5360720a55a3") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_cbrs_forced_weak_signalstrength(self): |
| """ Test CBRS Data switch |
| |
| Verify Data Browsing works fine on cbrs before weak signal. |
| Using CAT Scone apk, reduce the signal strength. |
| Verify the data switches from CBRS to Psim. |
| Verify Data Browsing works fine on pSIM. |
| Verify Data Browsing works fine on cbrs after simulation is stopped. |
| |
| Returns: |
| True if pass; False if fail. |
| """ |
| self.simulate_weak_signal = self.user_params.get("simulate_weak_signal") |
| self.stop_simulation = self.user_params.get("stop_simulate") |
| ads = [self.android_devices[0]] |
| result = self._test_CBRS_switch( |
| ads, self._is_current_data_on_cbrs, |
| self._test_data_browsing_success_using_sl4a, |
| self.simulate_weak_signal, |
| self._is_current_data_on_default, |
| self._test_data_browsing_success_using_sl4a, |
| self.stop_simulation, |
| self._test_data_browsing_success_using_sl4a, |
| self._is_current_data_on_cbrs, True) |
| |
| self.log.info("CBRS data switch Result: %s", result) |
| return result |
| |
| @test_tracker_info(uuid="066ff1e4-c3f7-4834-a86a-6ebd9fbbee06") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_cbrs_forced_out_of_service(self): |
| """ Test CBRS Data switch |
| |
| Verify Data Browsing works fine on cbrs before weak signal. |
| Using CAT Scone apk, reduce the signal strength |
| to mimic CBRS signal out of service |
| Verify the data switches from CBRS to Psim. |
| Verify Data Browsing works fine on pSIM. |
| Verify Data Browsing works fine on cbrs after simulation is stopped. |
| |
| Returns: |
| True if pass; False if fail. |
| """ |
| self.simulate_out_of_service = self.user_params.get("simulate_out_of_service") |
| self.stop_simulation = self.user_params.get("stop_simulate") |
| ads = [self.android_devices[0]] |
| result = self._test_CBRS_switch( |
| ads, self._is_current_data_on_cbrs, |
| self._test_data_browsing_success_using_sl4a, |
| self.simulate_out_of_service, |
| self._is_current_data_on_default, |
| self._test_data_browsing_success_using_sl4a, |
| self.stop_simulation, |
| self._test_data_browsing_success_using_sl4a, |
| self._is_current_data_on_cbrs, True) |
| |
| self.log.info("CBRS data switch Result: %s", result) |
| return result |
| |
| @test_tracker_info(uuid="8268f88f-c2e3-4000-8f05-97e34e028a92") |
| @TelephonyBaseTest.tel_test_wrap |
| def test_cbrs_forced_out_of_coverage(self): |
| """ Test CBRS Data switch |
| |
| Verify Data Browsing works fine on cbrs before weak signal. |
| Using CAT Scone apk, reduce the signal strength |
| to mimic CBRS signal out of coverage. |
| Verify the data switches from CBRS to Psim. |
| Verify Data Browsing works fine on pSIM. |
| Verify Data Browsing works fine on cbrs after simulation is stopped. |
| |
| Returns: |
| True if pass; False if fail. |
| """ |
| self.simulate_out_of_coverage = self.user_params.get("simulate_out_of_coverage") |
| self.stop_simulation = self.user_params.get("stop_simulate") |
| ads = [self.android_devices[0]] |
| result = self._test_CBRS_switch( |
| ads, self._is_current_data_on_cbrs, |
| self._test_data_browsing_success_using_sl4a, |
| self.simulate_out_of_coverage, |
| self._is_current_data_on_default, |
| self._test_data_browsing_success_using_sl4a, |
| self.stop_simulation, |
| self._test_data_browsing_success_using_sl4a, |
| self._is_current_data_on_cbrs, True) |
| |
| self.log.info("CBRS data switch Result: %s", result) |
| return result |
