acts_tests/acts_contrib/test_utils/tel/tel_voice_conf_utils.py

#!/usr/bin/env python3
#
#   Copyright 2021 - 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.

import time

from acts import signals
from acts_contrib.test_utils.tel.tel_defines import CALL_CAPABILITY_MANAGE_CONFERENCE
from acts_contrib.test_utils.tel.tel_defines import CALL_PROPERTY_CONFERENCE
from acts_contrib.test_utils.tel.tel_defines import CALL_STATE_ACTIVE
from acts_contrib.test_utils.tel.tel_defines import CALL_STATE_HOLDING
from acts_contrib.test_utils.tel.tel_defines import PHONE_TYPE_GSM
from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_IN_CALL
from acts_contrib.test_utils.tel.tel_test_utils import call_setup_teardown
from acts_contrib.test_utils.tel.tel_test_utils import get_call_uri
from acts_contrib.test_utils.tel.tel_test_utils import hangup_call
from acts_contrib.test_utils.tel.tel_test_utils import multithread_func
from acts_contrib.test_utils.tel.tel_test_utils import num_active_calls
from acts_contrib.test_utils.tel.tel_test_utils import verify_incall_state
from acts_contrib.test_utils.tel.tel_voice_utils import get_cep_conference_call_id
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 is_phone_in_call_wcdma
from acts_contrib.test_utils.tel.tel_test_utils import is_uri_equivalent
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_setup_voice_3g
from acts_contrib.test_utils.tel.tel_voice_utils import phone_setup_volte
from acts_contrib.test_utils.tel.tel_voice_utils import swap_calls


def _three_phone_call_mo_add_mo(log, ads, phone_setups, verify_funcs):
    """Use 3 phones to make MO calls.

    Call from PhoneA to PhoneB, accept on PhoneB.
    Call from PhoneA to PhoneC, accept on PhoneC.

    Args:
        ads: list of ad object.
            The list should have three objects.
        phone_setups: list of phone setup functions.
            The list should have three objects.
        verify_funcs: list of phone call verify functions.
            The list should have three objects.

    Returns:
        If success, return 'call_AB' id in PhoneA.
        if fail, return None.
    """

    class _CallException(Exception):
        pass

    try:
        verify_func_a, verify_func_b, verify_func_c = verify_funcs
        tasks = []
        for ad, setup_func in zip(ads, phone_setups):
            if setup_func is not None:
                tasks.append((setup_func, (log, ad)))
        if tasks != [] and not multithread_func(log, tasks):
            log.error("Phone Failed to Set Up Properly.")
            raise _CallException("Setup failed.")
        for ad in ads:
            ad.droid.telecomCallClearCallList()
            if num_active_calls(log, ad) != 0:
                ad.log.error("Phone Call List is not empty.")
                raise _CallException("Clear call list failed.")

        log.info("Step1: Call From PhoneA to PhoneB.")
        if not call_setup_teardown(
                log,
                ads[0],
                ads[1],
                ad_hangup=None,
                verify_caller_func=verify_func_a,
                verify_callee_func=verify_func_b):
            raise _CallException("PhoneA call PhoneB failed.")

        calls = ads[0].droid.telecomCallGetCallIds()
        ads[0].log.info("Calls in PhoneA %s", calls)
        if num_active_calls(log, ads[0]) != 1:
            raise _CallException("Call list verify failed.")
        call_ab_id = calls[0]

        log.info("Step2: Call From PhoneA to PhoneC.")
        if not call_setup_teardown(
                log,
                ads[0],
                ads[2],
                ad_hangup=None,
                verify_caller_func=verify_func_a,
                verify_callee_func=verify_func_c):
            raise _CallException("PhoneA call PhoneC failed.")
        if not verify_incall_state(log, [ads[0], ads[1], ads[2]],
                                   True):
            raise _CallException("Not All phones are in-call.")

    except _CallException:
        return None

    return call_ab_id


def _test_call_mo_mo_add_swap_x(log,
                                ads,
                                num_swaps,
                                phone_setup_a=None,
                                phone_setup_b=None,
                                phone_setup_c=None,
                                verify_phone_a_network_subscription=None,
                                verify_phone_b_network_subscription=None,
                                verify_phone_c_network_subscription=None):
    """Test swap feature in VoLTE call.

    PhoneA call PhoneB , accept on PhoneB.
    PhoneA call PhoneC , accept on PhoneC.
    Swap active call on PhoneA.(N times)

    Args:
        num_swaps: do swap for 'num_swaps' times.
            This value can be 0 (no swap operation).

    Returns:
        call_ab_id, call_ac_id if succeed;
        None, None if failed.

    """
    if ((phone_setup_a == phone_setup_voice_3g) or (phone_setup_a == phone_setup_voice_2g)):
        # make sure PhoneA is GSM phone before proceed.
        if (ads[0].droid.telephonyGetPhoneType() != PHONE_TYPE_GSM):
            raise signals.TestSkip("not GSM phone, abort swap test.")

    if (((phone_setup_b == phone_setup_voice_3g)
        and (phone_setup_c == phone_setup_voice_3g)) or
        ((phone_setup_b == phone_setup_voice_2g)
        and (phone_setup_c == phone_setup_voice_2g))):
        # make sure PhoneB and PhoneC are GSM phone before proceed.
        for ad in [ads[1], ads[2]]:
            if (ad.droid.telephonyGetPhoneType() != PHONE_TYPE_GSM):
                raise signals.TestSkip("not GSM phone, abort swap test.")

    call_ab_id = _three_phone_call_mo_add_mo(log,
        [ads[0], ads[1], ads[2]],
        [phone_setup_a, phone_setup_b, phone_setup_c], [
            verify_phone_a_network_subscription, verify_phone_b_network_subscription,
            verify_phone_c_network_subscription
        ])
    if call_ab_id is None:
        log.error("Failed to get call_ab_id")
        return None, None

    calls = ads[0].droid.telecomCallGetCallIds()
    ads[0].log.info("Calls in PhoneA %s", calls)
    if num_active_calls(log, ads[0]) != 2:
        return None, None
    if calls[0] == call_ab_id:
        call_ac_id = calls[1]
    else:
        call_ac_id = calls[0]

    if num_swaps > 0:
        log.info("Step3: Begin Swap x%s test.", num_swaps)
        if not swap_calls(log, ads, call_ab_id, call_ac_id,
                          num_swaps):
            log.error("Swap test failed.")
            return None, None

    return call_ab_id, call_ac_id


def _merge_ims_conference_call(log, ads, call_ab_id, call_ac_id):
    """Merge IMS conference call for both cases of CEP enabled and disabled.

    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneB.
    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneC.
    Merge calls to conference on PhoneA.

    Args:
        call_ab_id: call id for call_AB on PhoneA.
        call_ac_id: call id for call_AC on PhoneA.

    Returns:
        call_id for conference
    """

    log.info("Step4: Merge to Conf Call and verify Conf Call.")
    ads[0].droid.telecomCallJoinCallsInConf(call_ab_id, call_ac_id)
    time.sleep(WAIT_TIME_IN_CALL)
    calls = ads[0].droid.telecomCallGetCallIds()
    ads[0].log.info("Calls in PhoneA %s", calls)

    call_conf_id = None
    if num_active_calls(log, ads[0]) != 1:
        ads[0].log.info("Total number of call ids is not 1.")
        call_conf_id = get_cep_conference_call_id(ads[0])
        if call_conf_id is not None:
            log.info("New conference call id is found. CEP enabled.")
            calls.remove(call_conf_id)
            if (set(ads[0].droid.telecomCallGetCallChildren(
                call_conf_id)) != set(calls)):
                ads[0].log.error(
                    "Children list %s for conference call is not correct.",
                    ads[0].droid.telecomCallGetCallChildren(call_conf_id))
                return None

            if (CALL_PROPERTY_CONFERENCE not in ads[0]
                    .droid.telecomCallGetProperties(call_conf_id)):
                ads[0].log.error("Conf call id % properties wrong: %s", call_conf_id,
                        ads[0].droid.telecomCallGetProperties(call_conf_id))
                return None

                if (CALL_CAPABILITY_MANAGE_CONFERENCE not in ads[0]
                        .droid.telecomCallGetCapabilities(call_conf_id)):
                    ads[0].log.error(
                        "Conf call id %s capabilities wrong: %s", call_conf_id,
                        ads[0].droid.telecomCallGetCapabilities(call_conf_id))
                    return None

                if (call_ab_id in calls) or (call_ac_id in calls):
                    log.error("Previous call ids should not in new call"
                    " list after merge.")
                    return None
        else:
            for call_id in calls:
                if call_id != call_ab_id and call_id != call_ac_id:
                    call_conf_id = call_id
                    log.info("CEP not enabled.")

        if not call_conf_id:
            log.error("Merge call fail, no new conference call id.")
            raise signals.TestFailure(
                "Calls were not merged. Failed to merge calls.",
                extras={"fail_reason": "Calls were not merged."
                    " Failed to merge calls."})
        if not verify_incall_state(log, [ads[0], ads[1], ads[2]], True):
            return False

        # Check if Conf Call is currently active
        if ads[0].droid.telecomCallGetCallState(
                call_conf_id) != CALL_STATE_ACTIVE:
            ads[0].log.error(
                "Call_ID: %s, state: %s, expected: STATE_ACTIVE", call_conf_id,
                ads[0].droid.telecomCallGetCallState(call_conf_id))
            return None

        return call_conf_id


def _hangup_call(log, ad, device_description='Device'):
    if not hangup_call(log, ad):
        ad.log.error("Failed to hang up on %s", device_description)
        return False
    return True


def _test_ims_conference_merge_drop_second_call_from_participant(
        log, ads, call_ab_id, call_ac_id):
    """Test conference merge and drop in IMS (VoLTE or WiFi Calling) call.
    (CEP enabled).

    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneB.
    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneC.
    Merge calls to conference on PhoneA (CEP enabled IMS conference).
    Hangup on PhoneC, check call continues between AB.
    Hangup on PhoneB, check A ends.

    Args:
        call_ab_id: call id for call_AB on PhoneA.
        call_ac_id: call id for call_AC on PhoneA.

    Returns:
        True if succeed;
        False if failed.
    """

    call_conf_id = _merge_ims_conference_call(log, ads, call_ab_id, call_ac_id)
    if call_conf_id is None:
        return False

    log.info("Step5: End call on PhoneC and verify call continues.")
    if not _hangup_call(log, ads[2], "PhoneC"):
        return False
    time.sleep(WAIT_TIME_IN_CALL)
    calls = ads[0].droid.telecomCallGetCallIds()
    ads[0].log.info("Calls in PhoneA %s", calls)
    if not verify_incall_state(log, [ads[0], ads[1]], True):
        return False
    if not verify_incall_state(log, [ads[2]], False):
        return False

    log.info("Step6: End call on PhoneB and verify PhoneA end.")
    if not _hangup_call(log, ads[1], "PhoneB"):
        return False
    time.sleep(WAIT_TIME_IN_CALL)
    if not verify_incall_state(log, [ads[0], ads[1], ads[2]], False):
        return False
    return True


def _test_ims_conference_merge_drop_second_call_from_host(
        log, ads, call_ab_id, call_ac_id):
    """Test conference merge and drop in IMS (VoLTE or WiFi Calling) call.
    (CEP enabled).

    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneB.
    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneC.
    Merge calls to conference on PhoneA (CEP enabled IMS conference).
    On PhoneA, disconnect call between A-C, verify PhoneA PhoneB still in call.
    On PhoneA, disconnect call between A-B, verify PhoneA PhoneB disconnected.

    Args:
        call_ab_id: call id for call_AB on PhoneA.
        call_ac_id: call id for call_AC on PhoneA.

    Returns:
        True if succeed;
        False if failed.
    """
    call_ab_uri = get_call_uri(ads[0], call_ab_id)
    call_ac_uri = get_call_uri(ads[0], call_ac_id)

    call_conf_id = _merge_ims_conference_call(log, ads, call_ab_id, call_ac_id)
    if call_conf_id is None:
        return False

    calls = ads[0].droid.telecomCallGetCallIds()
    calls.remove(call_conf_id)

    log.info("Step5: Disconnect call A-C and verify call continues.")
    call_to_disconnect = None
    for call in calls:
        if is_uri_equivalent(call_ac_uri, get_call_uri(ads[0], call)):
            call_to_disconnect = call
            calls.remove(call_to_disconnect)
            break
    if call_to_disconnect is None:
        log.error("Can NOT find call on host represents A-C.")
        return False
    else:
        ads[0].droid.telecomCallDisconnect(call_to_disconnect)
    time.sleep(WAIT_TIME_IN_CALL)
    if not verify_incall_state(log, [ads[0], ads[1]], True):
        return False
    if not verify_incall_state(log, [ads[2]], False):
        return False

    log.info(
        "Step6: Disconnect call A-B and verify PhoneA PhoneB end.")
    calls = ads[0].droid.telecomCallGetCallIds()
    call_to_disconnect = None
    for call in calls:
        if is_uri_equivalent(call_ab_uri, get_call_uri(ads[0], call)):
            call_to_disconnect = call
            calls.remove(call_to_disconnect)
            break
    if call_to_disconnect is None:
        log.error("Can NOT find call on host represents A-B.")
        return False
    else:
        ads[0].droid.telecomCallDisconnect(call_to_disconnect)
    time.sleep(WAIT_TIME_IN_CALL)
    if not verify_incall_state(log, [ads[0], ads[1], ads[2]], False):
        return False
    return True


def _test_ims_conference_merge_drop_first_call_from_participant(
        log, ads, call_ab_id, call_ac_id):
    """Test conference merge and drop in IMS (VoLTE or WiFi Calling) call.
    (CEP enabled).

    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneB.
    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneC.
    Merge calls to conference on PhoneA (CEP enabled IMS conference).
    Hangup on PhoneB, check call continues between AC.
    Hangup on PhoneC, check A ends.

    Args:
        call_ab_id: call id for call_AB on PhoneA.
        call_ac_id: call id for call_AC on PhoneA.

    Returns:
        True if succeed;
        False if failed.
    """
    call_conf_id = _merge_ims_conference_call(log, ads, call_ab_id, call_ac_id)
    if call_conf_id is None:
        return False

    log.info("Step5: End call on PhoneB and verify call continues.")
    if not _hangup_call(log, ads[1], "PhoneB"):
        return False
    time.sleep(WAIT_TIME_IN_CALL)
    if not verify_incall_state(log, [ads[0], ads[2]], True):
        return False
    if not verify_incall_state(log, [ads[1]], False):
        return False

    log.info("Step6: End call on PhoneC and verify PhoneA end.")
    if not _hangup_call(log, ads[2], "PhoneC"):
        return False
    time.sleep(WAIT_TIME_IN_CALL)
    if not verify_incall_state(log, [ads[0], ads[1], ads[2]], False):
        return False
    return True


def _test_ims_conference_merge_drop_first_call_from_host(
        log, ads, call_ab_id, call_ac_id):
    """Test conference merge and drop in IMS (VoLTE or WiFi Calling) call.
    (CEP enabled).

    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneB.
    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneC.
    Merge calls to conference on PhoneA (CEP enabled IMS conference).
    On PhoneA, disconnect call between A-B, verify PhoneA PhoneC still in call.
    On PhoneA, disconnect call between A-C, verify PhoneA PhoneC disconnected.

    Args:
        call_ab_id: call id for call_AB on PhoneA.
        call_ac_id: call id for call_AC on PhoneA.

    Returns:
        True if succeed;
        False if failed.
    """
    call_ab_uri = get_call_uri(ads[0], call_ab_id)
    call_ac_uri = get_call_uri(ads[0], call_ac_id)

    call_conf_id = _merge_ims_conference_call(log, ads, call_ab_id, call_ac_id)
    if call_conf_id is None:
        return False

    calls = ads[0].droid.telecomCallGetCallIds()
    calls.remove(call_conf_id)

    log.info("Step5: Disconnect call A-B and verify call continues.")
    call_to_disconnect = None
    for call in calls:
        if is_uri_equivalent(call_ab_uri, get_call_uri(ads[0], call)):
            call_to_disconnect = call
            calls.remove(call_to_disconnect)
            break
    if call_to_disconnect is None:
        log.error("Can NOT find call on host represents A-B.")
        return False
    else:
        ads[0].droid.telecomCallDisconnect(call_to_disconnect)
    time.sleep(WAIT_TIME_IN_CALL)
    if not verify_incall_state(log, [ads[0], ads[2]], True):
        return False
    if not verify_incall_state(log, [ads[1]], False):
        return False

    log.info(
        "Step6: Disconnect call A-C and verify PhoneA PhoneC end.")
    calls = ads[0].droid.telecomCallGetCallIds()
    call_to_disconnect = None
    for call in calls:
        if is_uri_equivalent(call_ac_uri, get_call_uri(ads[0], call)):
            call_to_disconnect = call
            calls.remove(call_to_disconnect)
            break
    if call_to_disconnect is None:
        log.error("Can NOT find call on host represents A-C.")
        return False
    else:
        ads[0].droid.telecomCallDisconnect(call_to_disconnect)
    time.sleep(WAIT_TIME_IN_CALL)
    if not verify_incall_state(log, [ads[0], ads[1], ads[2]], False):
        return False
    return True


def _three_phone_call_mo_add_mt(log, ads, phone_setups, verify_funcs):
    """Use 3 phones to make MO call and MT call.

    Call from PhoneA to PhoneB, accept on PhoneB.
    Call from PhoneC to PhoneA, accept on PhoneA.

    Args:
        ads: list of ad object.
            The list should have three objects.
        phone_setups: list of phone setup functions.
            The list should have three objects.
        verify_funcs: list of phone call verify functions.
            The list should have three objects.

    Returns:
        If success, return 'call_AB' id in PhoneA.
        if fail, return None.
    """

    class _CallException(Exception):
        pass

    try:
        verify_func_a, verify_func_b, verify_func_c = verify_funcs
        tasks = []
        for ad, setup_func in zip(ads, phone_setups):
            if setup_func is not None:
                tasks.append((setup_func, (log, ad)))
        if tasks != [] and not multithread_func(log, tasks):
            log.error("Phone Failed to Set Up Properly.")
            raise _CallException("Setup failed.")
        for ad in ads:
            ad.droid.telecomCallClearCallList()
            if num_active_calls(log, ad) != 0:
                ad.log.error("Phone Call List is not empty.")
                raise _CallException("Clear call list failed.")

        log.info("Step1: Call From PhoneA to PhoneB.")
        if not call_setup_teardown(
                log,
                ads[0],
                ads[1],
                ad_hangup=None,
                verify_caller_func=verify_func_a,
                verify_callee_func=verify_func_b):
            raise _CallException("PhoneA call PhoneB failed.")

        calls = ads[0].droid.telecomCallGetCallIds()
        ads[0].log.info("Calls in PhoneA %s", calls)
        if num_active_calls(log, ads[0]) != 1:
            raise _CallException("Call list verify failed.")
        call_ab_id = calls[0]

        log.info("Step2: Call From PhoneC to PhoneA.")
        if not call_setup_teardown(
                log,
                ads[2],
                ads[0],
                ad_hangup=None,
                verify_caller_func=verify_func_c,
                verify_callee_func=verify_func_a):
            raise _CallException("PhoneA call PhoneC failed.")
        if not verify_incall_state(log, [ads[0], ads[1], ads[2]],
                                   True):
            raise _CallException("Not All phones are in-call.")

    except _CallException:
        return None
    return call_ab_id


def _test_call_mo_mt_add_swap_x(log,
                                ads,
                                num_swaps,
                                phone_setup_a=None,
                                phone_setup_b=None,
                                phone_setup_c=None,
                                verify_phone_a_network_subscription=None,
                                verify_phone_b_network_subscription=None,
                                verify_phone_c_network_subscription=None):
    """Test swap feature in VoLTE call.

    PhoneA call PhoneB, accept on PhoneB.
    PhoneC call PhoneA, accept on PhoneA.
    Swap active call on PhoneA. (N times)

    Args:
        num_swaps: do swap for 'num_swaps' times.
            This value can be 0 (no swap operation).

    Returns:
        call_ab_id, call_ac_id if succeed;
        None, None if failed.

    """
    if ((phone_setup_a == phone_setup_voice_3g) or (phone_setup_a == phone_setup_voice_2g)):
        # make sure PhoneA is GSM phone before proceed.
        if (ads[0].droid.telephonyGetPhoneType() != PHONE_TYPE_GSM):
            raise signals.TestSkip("not GSM phone, abort swap test.")

    if (((phone_setup_b == phone_setup_voice_3g)
        and (phone_setup_c == phone_setup_voice_3g)) or
        ((phone_setup_b == phone_setup_voice_2g)
        and (phone_setup_c == phone_setup_voice_2g))):
        # make sure PhoneB and PhoneC are GSM phone before proceed.
        for ad in [ads[1], ads[2]]:
            if (ad.droid.telephonyGetPhoneType() != PHONE_TYPE_GSM):
                raise signals.TestSkip("not GSM phone, abort swap test.")

    call_ab_id = _three_phone_call_mo_add_mt(log,
        [ads[0], ads[1], ads[2]],
        [phone_setup_a, phone_setup_b, phone_setup_c], [
            verify_phone_a_network_subscription, verify_phone_b_network_subscription,
            verify_phone_c_network_subscription
            ])
    if call_ab_id is None:
        log.error("Failed to get call_ab_id")
        return None, None

    calls = ads[0].droid.telecomCallGetCallIds()
    ads[0].log.info("Calls in PhoneA %s", calls)
    if num_active_calls(log, ads[0]) != 2:
        return None, None
    if calls[0] == call_ab_id:
        call_ac_id = calls[1]
    else:
        call_ac_id = calls[0]

    if num_swaps > 0:
        log.info("Step3: Begin Swap x%s test.", num_swaps)
        if not swap_calls(log, ads, call_ab_id, call_ac_id,
                          num_swaps):
            log.error("Swap test failed.")
            return None, None

    return call_ab_id, call_ac_id


def _three_phone_call_mt_add_mt(log, ads, phone_setups, verify_funcs):
    """Use 3 phones to make MT call and MT call.

    Call from PhoneB to PhoneA, accept on PhoneA.
    Call from PhoneC to PhoneA, accept on PhoneA.

    Args:
        ads: list of ad object.
            The list should have three objects.
        phone_setups: list of phone setup functions.
            The list should have three objects.
        verify_funcs: list of phone call verify functions.
            The list should have three objects.

    Returns:
        If success, return 'call_AB' id in PhoneA.
        if fail, return None.
    """

    class _CallException(Exception):
        pass

    try:
        verify_func_a, verify_func_b, verify_func_c = verify_funcs
        tasks = []
        for ad, setup_func in zip(ads, phone_setups):
            if setup_func is not None:
                tasks.append((setup_func, (log, ad)))
        if tasks != [] and not multithread_func(log, tasks):
            log.error("Phone Failed to Set Up Properly.")
            raise _CallException("Setup failed.")
        for ad in ads:
            ad.droid.telecomCallClearCallList()
            if num_active_calls(log, ad) != 0:
                ad.log.error("Phone Call List is not empty.")
                raise _CallException("Clear call list failed.")

        log.info("Step1: Call From PhoneB to PhoneA.")
        if not call_setup_teardown(
                log,
                ads[1],
                ads[0],
                ad_hangup=None,
                verify_caller_func=verify_func_b,
                verify_callee_func=verify_func_a):
            raise _CallException("PhoneB call PhoneA failed.")

        calls = ads[0].droid.telecomCallGetCallIds()
        ads[0].log.info("Calls in PhoneA %s", calls)
        if num_active_calls(log, ads[0]) != 1:
            raise _CallException("Call list verify failed.")
        call_ab_id = calls[0]

        log.info("Step2: Call From PhoneC to PhoneA.")
        if not call_setup_teardown(
                log,
                ads[2],
                ads[0],
                ad_hangup=None,
                verify_caller_func=verify_func_c,
                verify_callee_func=verify_func_a):
            raise _CallException("PhoneA call PhoneC failed.")
        if not verify_incall_state(log, [ads[0], ads[1], ads[2]],
                                  True):
            raise _CallException("Not All phones are in-call.")

    except _CallException:
        return None

    return call_ab_id


def _test_call_mt_mt_add_swap_x(log,
                                ads,
                                num_swaps,
                                phone_setup_a=None,
                                phone_setup_b=None,
                                phone_setup_c=None,
                                verify_phone_a_network_subscription=None,
                                verify_phone_b_network_subscription=None,
                                verify_phone_c_network_subscription=None):
    """Test swap feature in VoLTE call.

    PhoneB call PhoneA, accept on PhoneA.
    PhoneC call PhoneA, accept on PhoneA.
    Swap active call on PhoneA. (N times)

    Args:
        num_swaps: do swap for 'num_swaps' times.
            This value can be 0 (no swap operation).

    Returns:
        call_ab_id, call_ac_id if succeed;
        None, None if failed.

    """
    if ((phone_setup_a == phone_setup_voice_3g) or (phone_setup_a == phone_setup_voice_2g)):
        # make sure PhoneA is GSM phone before proceeSSd.
        if (ads[0].droid.telephonyGetPhoneType() != PHONE_TYPE_GSM):
            raise signals.TestSkip("not GSM phone, abort swap test.")

    if (((phone_setup_b == phone_setup_voice_3g)
        and (phone_setup_c == phone_setup_voice_3g)) or
        ((phone_setup_b == phone_setup_voice_2g)
        and (phone_setup_c == phone_setup_voice_2g))):
        # make sure PhoneB and PhoneC are GSM phone before proceed.
        for ad in [ads[1], ads[2]]:
            if (ad.droid.telephonyGetPhoneType() != PHONE_TYPE_GSM):
                raise signals.TestSkip("not GSM phone, abort swap test.")

    call_ab_id = _three_phone_call_mt_add_mt(log,
        [ads[0], ads[1], ads[2]],
        [phone_setup_a, phone_setup_b, phone_setup_c], [
            verify_phone_a_network_subscription, verify_phone_b_network_subscription,
            verify_phone_c_network_subscription
        ])
    if call_ab_id is None:
        log.error("Failed to get call_ab_id")
        return None, None

    calls = ads[0].droid.telecomCallGetCallIds()
    ads[0].log.info("Calls in PhoneA %s", calls)
    if num_active_calls(log, ads[0]) != 2:
        return None, None
    if calls[0] == call_ab_id:
        call_ac_id = calls[1]
    else:
        call_ac_id = calls[0]

    if num_swaps > 0:
        log.info("Step3: Begin Swap x%s test.", num_swaps)
        if not swap_calls(log, ads, call_ab_id, call_ac_id,
                          num_swaps):
            log.error("Swap test failed.")
            return None, None

    return call_ab_id, call_ac_id


def _three_phone_hangup_call_verify_call_state(
        log, ad_hangup, ad_verify, call_id, call_state, ads_active):
    """Private Test utility for swap test.

    Hangup on 'ad_hangup'.
    Verify 'call_id' on 'ad_verify' is in expected 'call_state'
    Verify each ad in ads_active are 'in-call'.

    Args:
        ad_hangup: android object to hangup call.
        ad_verify: android object to verify call id state.
        call_id: call id in 'ad_verify'.
        call_state: expected state for 'call_id'.
            'call_state' is either CALL_STATE_HOLDING or CALL_STATE_ACTIVE.
        ads_active: list of android object.
            Each one of them should be 'in-call' after 'hangup' operation.

    Returns:
        True if no error happened. Otherwise False.

    """

    ad_hangup.log.info("Hangup, verify call continues.")
    if not _hangup_call(log, ad_hangup):
        ad_hangup.log.error("Phone fails to hang up")
        return False
    time.sleep(WAIT_TIME_IN_CALL)

    if ad_verify.droid.telecomCallGetCallState(call_id) != call_state:
        ad_verify.log.error(
            "Call_id: %s, state: %s, expected: %s", call_id,
            ad_verify.droid.telecomCallGetCallState(call_id), call_state)
        return False
    ad_verify.log.info("Call in expected %s state", call_state)

    if not verify_incall_state(log, ads_active, True):
        ads_active.log.error("Phone not in call state")
        return False
    if not verify_incall_state(log, [ad_hangup], False):
        ad_hangup.log.error("Phone not in hangup state")
        return False

    return True


def _get_expected_call_state(ad):
    if "vzw" in [
            sub["operator"]
            for sub in ad.telephony["subscription"].values()
    ]:
        return CALL_STATE_ACTIVE
    return CALL_STATE_HOLDING