#!/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.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
from acts.test_utils.tel.tel_defines import DIRECTION_MOBILE_ORIGINATED
from acts.test_utils.tel.tel_defines import DIRECTION_MOBILE_TERMINATED
from acts.test_utils.tel.tel_defines import WAIT_TIME_BETWEEN_REG_AND_CALL
from acts.test_utils.tel.tel_defines import WAIT_TIME_IN_CALL
from acts.test_utils.tel.tel_defines import WAIT_TIME_FOR_CBRS_DATA_SWITCH
from acts.test_utils.tel.tel_test_utils import get_phone_number
from acts.test_utils.tel.tel_test_utils import hangup_call
from acts.test_utils.tel.tel_test_utils import hangup_call_by_adb
from acts.test_utils.tel.tel_test_utils import initiate_call
from acts.test_utils.tel.tel_test_utils import is_phone_not_in_call
from acts.test_utils.tel.tel_test_utils import wait_and_answer_call
from acts.test_utils.tel.tel_test_utils import is_phone_in_call
from acts.test_utils.tel.tel_test_utils import start_qxdm_loggers
from acts.test_utils.tel.tel_test_utils import test_data_browsing_success_using_sl4a
from acts.test_utils.tel.tel_test_utils import test_data_browsing_failure_using_sl4a
from acts.test_utils.tel.tel_test_utils import get_operator_name
from acts.test_utils.tel.tel_test_utils import is_current_data_on_cbrs
from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_3g
from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_2g
from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_csfb
from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_iwlan
from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_not_iwlan
from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_volte
from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_general
from acts.test_utils.tel.tel_voice_utils import phone_setup_volte_for_subscription
from acts.test_utils.tel.tel_voice_utils import phone_setup_cdma
from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_2g
from acts.test_utils.tel.tel_voice_utils import phone_idle_3g
from acts.test_utils.tel.tel_voice_utils import phone_idle_2g
from acts.test_utils.tel.tel_voice_utils import phone_idle_csfb
from acts.test_utils.tel.tel_voice_utils import phone_idle_iwlan
from acts.test_utils.tel.tel_voice_utils import phone_idle_not_iwlan
from acts.test_utils.tel.tel_voice_utils import phone_idle_volte
from acts.test_utils.tel.tel_subscription_utils import get_subid_from_slot_index
from acts.test_utils.tel.tel_subscription_utils import get_operatorname_from_slot_index
from acts.utils import get_current_epoch_time


class TelLiveCBRSTest(TelephonyBaseTest):
    def __init__(self, controllers):
        TelephonyBaseTest.__init__(self, controllers)
        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_data_subid = None


    def on_fail(self, test_name, begin_time):
        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

        # Fetch CBRS sub_id and default
        slot_dict = {0: {}, 1: {}}
        for slot in (0, 1):
            slot_dict[slot]['sub_id'] = get_subid_from_slot_index(
                ads[0].log, ads[0], slot)
            slot_dict[slot]['operator'] = get_operatorname_from_slot_index(
                ads[0], slot)
            if "Google" in slot_dict[slot]['operator']:
                self.cbrs_subid = slot_dict[slot]['sub_id']
            else:
                self.default_data_subid = slot_dict[slot]['sub_id']
            ads[0].log.info("Slot %d - Sub %s - %s", slot,
                            slot_dict[slot]['sub_id'],
                            slot_dict[slot]['operator'])

        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_data_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)


    """ 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)
        slot_dict = {0: {}, 1: {}}
        for slot in (0, 1):
            slot_dict[slot]['sub_id'] = get_subid_from_slot_index(
                ads[0].log, ads[0], slot)
            slot_dict[slot]['operator'] = get_operatorname_from_slot_index(
                ads[0], slot)
            if "Google" in slot_dict[slot]['operator']:
                self.cbrs_subid = slot_dict[slot]['sub_id']
            else:
                self.default_data_subid = slot_dict[slot]['sub_id']
            ads[0].log.info("Slot %d - Sub %s - %s", slot,
                            slot_dict[slot]['sub_id'],
                            slot_dict[slot]['operator'])
        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)