1#!/usr/bin/env python3
2#
3#   Copyright 2021 - 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
17import time
18
19from acts import signals
20from acts.libs.utils.multithread import multithread_func
21from acts_contrib.test_utils.tel.tel_defines import CALL_CAPABILITY_MANAGE_CONFERENCE
22from acts_contrib.test_utils.tel.tel_defines import CALL_PROPERTY_CONFERENCE
23from acts_contrib.test_utils.tel.tel_defines import CALL_STATE_ACTIVE
24from acts_contrib.test_utils.tel.tel_defines import CALL_STATE_HOLDING
25from acts_contrib.test_utils.tel.tel_defines import PHONE_TYPE_GSM
26from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_IN_CALL
27from acts_contrib.test_utils.tel.tel_phone_setup_utils import phone_setup_voice_2g
28from acts_contrib.test_utils.tel.tel_phone_setup_utils import phone_setup_voice_3g
29from acts_contrib.test_utils.tel.tel_subscription_utils import get_incoming_voice_sub_id
30from acts_contrib.test_utils.tel.tel_test_utils import get_call_uri
31from acts_contrib.test_utils.tel.tel_test_utils import num_active_calls
32from acts_contrib.test_utils.tel.tel_test_utils import verify_incall_state
33from acts_contrib.test_utils.tel.tel_test_utils import is_uri_equivalent
34from acts_contrib.test_utils.tel.tel_voice_utils import call_setup_teardown
35from acts_contrib.test_utils.tel.tel_voice_utils import get_cep_conference_call_id
36from acts_contrib.test_utils.tel.tel_voice_utils import hangup_call
37from acts_contrib.test_utils.tel.tel_voice_utils import initiate_call
38from acts_contrib.test_utils.tel.tel_voice_utils import swap_calls
39from acts_contrib.test_utils.tel.tel_voice_utils import wait_and_reject_call_for_subscription
40
41
42def _three_phone_call_mo_add_mo(log, ads, phone_setups, verify_funcs):
43    """Use 3 phones to make MO calls.
44
45    Call from PhoneA to PhoneB, accept on PhoneB.
46    Call from PhoneA to PhoneC, accept on PhoneC.
47
48    Args:
49        ads: list of ad object.
50            The list should have three objects.
51        phone_setups: list of phone setup functions.
52            The list should have three objects.
53        verify_funcs: list of phone call verify functions.
54            The list should have three objects.
55
56    Returns:
57        If success, return 'call_AB' id in PhoneA.
58        if fail, return None.
59    """
60
61    class _CallException(Exception):
62        pass
63
64    try:
65        verify_func_a, verify_func_b, verify_func_c = verify_funcs
66        tasks = []
67        for ad, setup_func in zip(ads, phone_setups):
68            if setup_func is not None:
69                tasks.append((setup_func, (log, ad)))
70        if tasks != [] and not multithread_func(log, tasks):
71            log.error("Phone Failed to Set Up Properly.")
72            raise _CallException("Setup failed.")
73        for ad in ads:
74            ad.droid.telecomCallClearCallList()
75            if num_active_calls(log, ad) != 0:
76                ad.log.error("Phone Call List is not empty.")
77                raise _CallException("Clear call list failed.")
78
79        log.info("Step1: Call From PhoneA to PhoneB.")
80        if not call_setup_teardown(
81                log,
82                ads[0],
83                ads[1],
84                ad_hangup=None,
85                verify_caller_func=verify_func_a,
86                verify_callee_func=verify_func_b):
87            raise _CallException("PhoneA call PhoneB failed.")
88
89        calls = ads[0].droid.telecomCallGetCallIds()
90        ads[0].log.info("Calls in PhoneA %s", calls)
91        if num_active_calls(log, ads[0]) != 1:
92            raise _CallException("Call list verify failed.")
93        call_ab_id = calls[0]
94
95        log.info("Step2: Call From PhoneA to PhoneC.")
96        if not call_setup_teardown(
97                log,
98                ads[0],
99                ads[2],
100                ad_hangup=None,
101                verify_caller_func=verify_func_a,
102                verify_callee_func=verify_func_c):
103            raise _CallException("PhoneA call PhoneC failed.")
104        if not verify_incall_state(log, [ads[0], ads[1], ads[2]],
105                                   True):
106            raise _CallException("Not All phones are in-call.")
107
108    except _CallException:
109        return None
110
111    return call_ab_id
112
113
114def _test_call_mo_mo_add_swap_x(log,
115                                ads,
116                                num_swaps,
117                                phone_setup_a=None,
118                                phone_setup_b=None,
119                                phone_setup_c=None,
120                                verify_phone_a_network_subscription=None,
121                                verify_phone_b_network_subscription=None,
122                                verify_phone_c_network_subscription=None):
123    """Test swap feature in VoLTE call.
124
125    PhoneA call PhoneB , accept on PhoneB.
126    PhoneA call PhoneC , accept on PhoneC.
127    Swap active call on PhoneA.(N times)
128
129    Args:
130        num_swaps: do swap for 'num_swaps' times.
131            This value can be 0 (no swap operation).
132
133    Returns:
134        call_ab_id, call_ac_id if succeed;
135        None, None if failed.
136
137    """
138    if ((phone_setup_a == phone_setup_voice_3g) or (phone_setup_a == phone_setup_voice_2g)):
139        # make sure PhoneA is GSM phone before proceed.
140        if (ads[0].droid.telephonyGetPhoneType() != PHONE_TYPE_GSM):
141            raise signals.TestSkip("not GSM phone, abort swap test.")
142
143    if (((phone_setup_b == phone_setup_voice_3g)
144        and (phone_setup_c == phone_setup_voice_3g)) or
145        ((phone_setup_b == phone_setup_voice_2g)
146        and (phone_setup_c == phone_setup_voice_2g))):
147        # make sure PhoneB and PhoneC are GSM phone before proceed.
148        for ad in [ads[1], ads[2]]:
149            if (ad.droid.telephonyGetPhoneType() != PHONE_TYPE_GSM):
150                raise signals.TestSkip("not GSM phone, abort swap test.")
151
152    call_ab_id = _three_phone_call_mo_add_mo(log,
153        [ads[0], ads[1], ads[2]],
154        [phone_setup_a, phone_setup_b, phone_setup_c], [
155            verify_phone_a_network_subscription, verify_phone_b_network_subscription,
156            verify_phone_c_network_subscription
157        ])
158    if call_ab_id is None:
159        log.error("Failed to get call_ab_id")
160        return None, None
161
162    calls = ads[0].droid.telecomCallGetCallIds()
163    ads[0].log.info("Calls in PhoneA %s", calls)
164    if num_active_calls(log, ads[0]) != 2:
165        return None, None
166    if calls[0] == call_ab_id:
167        call_ac_id = calls[1]
168    else:
169        call_ac_id = calls[0]
170
171    if num_swaps > 0:
172        log.info("Step3: Begin Swap x%s test.", num_swaps)
173        if not swap_calls(log, ads, call_ab_id, call_ac_id,
174                          num_swaps):
175            log.error("Swap test failed.")
176            return None, None
177
178    return call_ab_id, call_ac_id
179
180
181def _merge_ims_conference_call(log, ads, call_ab_id, call_ac_id):
182    """Merge IMS conference call for both cases of CEP enabled and disabled.
183
184    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneB.
185    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneC.
186    Merge calls to conference on PhoneA.
187
188    Args:
189        call_ab_id: call id for call_AB on PhoneA.
190        call_ac_id: call id for call_AC on PhoneA.
191
192    Returns:
193        call_id for conference
194    """
195
196    log.info("Step4: Merge to Conf Call and verify Conf Call.")
197    ads[0].droid.telecomCallJoinCallsInConf(call_ab_id, call_ac_id)
198    time.sleep(WAIT_TIME_IN_CALL)
199    calls = ads[0].droid.telecomCallGetCallIds()
200    ads[0].log.info("Calls in PhoneA %s", calls)
201
202    call_conf_id = None
203    if num_active_calls(log, ads[0]) != 1:
204        ads[0].log.info("Total number of call ids is not 1.")
205        call_conf_id = get_cep_conference_call_id(ads[0])
206        if call_conf_id is not None:
207            log.info("New conference call id is found. CEP enabled.")
208            calls.remove(call_conf_id)
209            if (set(ads[0].droid.telecomCallGetCallChildren(
210                call_conf_id)) != set(calls)):
211                ads[0].log.error(
212                    "Children list %s for conference call is not correct.",
213                    ads[0].droid.telecomCallGetCallChildren(call_conf_id))
214                return None
215
216            if (CALL_PROPERTY_CONFERENCE not in ads[0]
217                    .droid.telecomCallGetProperties(call_conf_id)):
218                ads[0].log.error("Conf call id % properties wrong: %s", call_conf_id,
219                        ads[0].droid.telecomCallGetProperties(call_conf_id))
220                return None
221
222            if (CALL_CAPABILITY_MANAGE_CONFERENCE not in ads[0]
223                    .droid.telecomCallGetCapabilities(call_conf_id)):
224                ads[0].log.error(
225                    "Conf call id %s capabilities wrong: %s", call_conf_id,
226                    ads[0].droid.telecomCallGetCapabilities(call_conf_id))
227                return None
228
229            if (call_ab_id in calls) or (call_ac_id in calls):
230                log.error("Previous call ids should not in new call"
231                " list after merge.")
232                return None
233    else:
234        for call_id in calls:
235            if call_id != call_ab_id and call_id != call_ac_id:
236                call_conf_id = call_id
237                log.info("CEP not enabled.")
238
239    if not call_conf_id:
240        log.error("Merge call fail, no new conference call id.")
241        raise signals.TestFailure(
242            "Calls were not merged. Failed to merge calls.",
243            extras={"fail_reason": "Calls were not merged."
244                " Failed to merge calls."})
245    if not verify_incall_state(log, [ads[0], ads[1], ads[2]], True):
246        return False
247
248    # Check if Conf Call is currently active
249    if ads[0].droid.telecomCallGetCallState(
250            call_conf_id) != CALL_STATE_ACTIVE:
251        ads[0].log.error(
252            "Call_ID: %s, state: %s, expected: STATE_ACTIVE", call_conf_id,
253            ads[0].droid.telecomCallGetCallState(call_conf_id))
254        return None
255
256    return call_conf_id
257
258
259def _hangup_call(log, ad, device_description='Device'):
260    if not hangup_call(log, ad):
261        ad.log.error("Failed to hang up on %s", device_description)
262        return False
263    return True
264
265
266def _test_ims_conference_merge_drop_second_call_from_participant(
267        log, ads, call_ab_id, call_ac_id):
268    """Test conference merge and drop in IMS (VoLTE or WiFi Calling) call.
269    (supporting both cases of CEP enabled and disabled).
270
271    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneB.
272    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneC.
273    Merge calls to conference on PhoneA (CEP enabled IMS conference).
274    Hangup on PhoneC, check call continues between AB.
275    Hangup on PhoneB, check A ends.
276
277    Args:
278        call_ab_id: call id for call_AB on PhoneA.
279        call_ac_id: call id for call_AC on PhoneA.
280
281    Returns:
282        True if succeed;
283        False if failed.
284    """
285
286    call_conf_id = _merge_ims_conference_call(log, ads, call_ab_id, call_ac_id)
287    if call_conf_id is None:
288        return False
289
290    log.info("Step5: End call on PhoneC and verify call continues.")
291    if not _hangup_call(log, ads[2], "PhoneC"):
292        return False
293    time.sleep(WAIT_TIME_IN_CALL)
294    calls = ads[0].droid.telecomCallGetCallIds()
295    ads[0].log.info("Calls in PhoneA %s", calls)
296    if not verify_incall_state(log, [ads[0], ads[1]], True):
297        return False
298    if not verify_incall_state(log, [ads[2]], False):
299        return False
300
301    log.info("Step6: End call on PhoneB and verify PhoneA end.")
302    if not _hangup_call(log, ads[1], "PhoneB"):
303        return False
304    time.sleep(WAIT_TIME_IN_CALL)
305    if not verify_incall_state(log, [ads[0], ads[1], ads[2]], False):
306        return False
307    return True
308
309
310def _test_ims_conference_merge_drop_second_call_from_host(
311        log, ads, call_ab_id, call_ac_id):
312    """Test conference merge and drop in IMS (VoLTE or WiFi Calling) call.
313    (CEP enabled).
314
315    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneB.
316    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneC.
317    Merge calls to conference on PhoneA (CEP enabled IMS conference).
318    On PhoneA, disconnect call between A-C, verify PhoneA PhoneB still in call.
319    On PhoneA, disconnect call between A-B, verify PhoneA PhoneB disconnected.
320
321    Args:
322        call_ab_id: call id for call_AB on PhoneA.
323        call_ac_id: call id for call_AC on PhoneA.
324
325    Returns:
326        True if succeed;
327        False if failed.
328    """
329    call_ab_uri = get_call_uri(ads[0], call_ab_id)
330    call_ac_uri = get_call_uri(ads[0], call_ac_id)
331
332    call_conf_id = _merge_ims_conference_call(log, ads, call_ab_id, call_ac_id)
333    if call_conf_id is None:
334        return False
335
336    calls = ads[0].droid.telecomCallGetCallIds()
337    calls.remove(call_conf_id)
338
339    if not calls:
340        raise signals.TestSkip('CEP is not supported. The test will be skipped.')
341
342    log.info("Step5: Disconnect call A-C and verify call continues.")
343    call_to_disconnect = None
344    for call in calls:
345        new_uri = get_call_uri(ads[0], call)
346        if not new_uri:
347            ads[0].log.warning('New URI should NOT be None.')
348            raise signals.TestSkip('Invalid URI is found on the host. The test will be skipped.')
349        else:
350            ads[0].log.info('URI for call ID %s: %s', call, new_uri)
351        if is_uri_equivalent(call_ac_uri, new_uri):
352            call_to_disconnect = call
353            calls.remove(call_to_disconnect)
354            break
355    if call_to_disconnect is None:
356        log.error("Can NOT find call on host represents A-C.")
357        return False
358    else:
359        ads[0].droid.telecomCallDisconnect(call_to_disconnect)
360    time.sleep(WAIT_TIME_IN_CALL)
361    if not verify_incall_state(log, [ads[0], ads[1]], True):
362        return False
363    if not verify_incall_state(log, [ads[2]], False):
364        return False
365
366    log.info(
367        "Step6: Disconnect call A-B and verify PhoneA PhoneB end.")
368    calls = ads[0].droid.telecomCallGetCallIds()
369    call_to_disconnect = None
370    for call in calls:
371        new_uri = get_call_uri(ads[0], call)
372        if not new_uri:
373            ads[0].log.warning('New URI should NOT be None.')
374            raise signals.TestSkip('Invalid URI is found on the host. The test will be skipped.')
375        else:
376            ads[0].log.info('URI for call ID %s: %s', call, new_uri)
377        if is_uri_equivalent(call_ab_uri, new_uri):
378            call_to_disconnect = call
379            calls.remove(call_to_disconnect)
380            break
381    if call_to_disconnect is None:
382        log.error("Can NOT find call on host represents A-B.")
383        return False
384    else:
385        ads[0].droid.telecomCallDisconnect(call_to_disconnect)
386    time.sleep(WAIT_TIME_IN_CALL)
387    if not verify_incall_state(log, [ads[0], ads[1], ads[2]], False):
388        return False
389    return True
390
391
392def _test_ims_conference_merge_drop_first_call_from_participant(
393        log, ads, call_ab_id, call_ac_id):
394    """Test conference merge and drop in IMS (VoLTE or WiFi Calling) call.
395    (CEP enabled).
396
397    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneB.
398    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneC.
399    Merge calls to conference on PhoneA (CEP enabled IMS conference).
400    Hangup on PhoneB, check call continues between AC.
401    Hangup on PhoneC, check A ends.
402
403    Args:
404        call_ab_id: call id for call_AB on PhoneA.
405        call_ac_id: call id for call_AC on PhoneA.
406
407    Returns:
408        True if succeed;
409        False if failed.
410    """
411    call_conf_id = _merge_ims_conference_call(log, ads, call_ab_id, call_ac_id)
412    if call_conf_id is None:
413        return False
414
415    log.info("Step5: End call on PhoneB and verify call continues.")
416    if not _hangup_call(log, ads[1], "PhoneB"):
417        return False
418    time.sleep(WAIT_TIME_IN_CALL)
419    if not verify_incall_state(log, [ads[0], ads[2]], True):
420        return False
421    if not verify_incall_state(log, [ads[1]], False):
422        return False
423
424    log.info("Step6: End call on PhoneC and verify PhoneA end.")
425    if not _hangup_call(log, ads[2], "PhoneC"):
426        return False
427    time.sleep(WAIT_TIME_IN_CALL)
428    if not verify_incall_state(log, [ads[0], ads[1], ads[2]], False):
429        return False
430    return True
431
432
433def _test_ims_conference_merge_drop_first_call_from_host(
434        log, ads, call_ab_id, call_ac_id):
435    """Test conference merge and drop in IMS (VoLTE or WiFi Calling) call.
436    (CEP enabled).
437
438    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneB.
439    PhoneA in IMS (VoLTE or WiFi Calling) call with PhoneC.
440    Merge calls to conference on PhoneA (CEP enabled IMS conference).
441    On PhoneA, disconnect call between A-B, verify PhoneA PhoneC still in call.
442    On PhoneA, disconnect call between A-C, verify PhoneA PhoneC disconnected.
443
444    Args:
445        call_ab_id: call id for call_AB on PhoneA.
446        call_ac_id: call id for call_AC on PhoneA.
447
448    Returns:
449        True if succeed;
450        False if failed.
451    """
452    call_ab_uri = get_call_uri(ads[0], call_ab_id)
453    call_ac_uri = get_call_uri(ads[0], call_ac_id)
454
455    call_conf_id = _merge_ims_conference_call(log, ads, call_ab_id, call_ac_id)
456    if call_conf_id is None:
457        return False
458
459    calls = ads[0].droid.telecomCallGetCallIds()
460    calls.remove(call_conf_id)
461
462    if not calls:
463        raise signals.TestSkip('CEP is not supported. The test will be skipped.')
464
465    log.info("Step5: Disconnect call A-B and verify call continues.")
466    call_to_disconnect = None
467    for call in calls:
468        new_uri = get_call_uri(ads[0], call)
469        if not new_uri:
470            ads[0].log.warning('New URI should NOT be None.')
471            raise signals.TestSkip('Invalid URI is found on the host. The test will be skipped.')
472        else:
473            ads[0].log.info('URI for call ID %s: %s', call, new_uri)
474        if is_uri_equivalent(call_ab_uri, new_uri):
475            call_to_disconnect = call
476            calls.remove(call_to_disconnect)
477            break
478    if call_to_disconnect is None:
479        log.error("Can NOT find call on host represents A-B.")
480        return False
481    else:
482        ads[0].droid.telecomCallDisconnect(call_to_disconnect)
483    time.sleep(WAIT_TIME_IN_CALL)
484    if not verify_incall_state(log, [ads[0], ads[2]], True):
485        return False
486    if not verify_incall_state(log, [ads[1]], False):
487        return False
488
489    log.info(
490        "Step6: Disconnect call A-C and verify PhoneA PhoneC end.")
491    calls = ads[0].droid.telecomCallGetCallIds()
492    call_to_disconnect = None
493    for call in calls:
494        new_uri = get_call_uri(ads[0], call)
495        if not new_uri:
496            ads[0].log.warning('New URI should NOT be None.')
497            raise signals.TestSkip('Invalid URI is found on the host. The test will be skipped.')
498        else:
499            ads[0].log.info('URI for call ID %s: %s', call, new_uri)
500        if is_uri_equivalent(call_ac_uri, new_uri):
501            call_to_disconnect = call
502            calls.remove(call_to_disconnect)
503            break
504    if call_to_disconnect is None:
505        log.error("Can NOT find call on host represents A-C.")
506        return False
507    else:
508        ads[0].droid.telecomCallDisconnect(call_to_disconnect)
509    time.sleep(WAIT_TIME_IN_CALL)
510    if not verify_incall_state(log, [ads[0], ads[1], ads[2]], False):
511        return False
512    return True
513
514
515def _three_phone_call_mo_add_mt(
516    log,
517    ads,
518    phone_setups,
519    verify_funcs,
520    reject_once=False):
521    """Use 3 phones to make MO call and MT call.
522
523    Call from PhoneA to PhoneB, accept on PhoneB.
524    Call from PhoneC to PhoneA, accept on PhoneA.
525
526    Args:
527        ads: list of ad object.
528            The list should have three objects.
529        phone_setups: list of phone setup functions.
530            The list should have three objects.
531        verify_funcs: list of phone call verify functions.
532            The list should have three objects.
533        reject_once: True for rejecting the second call once.
534
535    Returns:
536        If success, return 'call_AB' id in PhoneA.
537        if fail, return None.
538    """
539
540    class _CallException(Exception):
541        pass
542
543    try:
544        verify_func_a, verify_func_b, verify_func_c = verify_funcs
545        tasks = []
546        for ad, setup_func in zip(ads, phone_setups):
547            if setup_func is not None:
548                tasks.append((setup_func, (log, ad)))
549        if tasks != [] and not multithread_func(log, tasks):
550            log.error("Phone Failed to Set Up Properly.")
551            raise _CallException("Setup failed.")
552        for ad in ads:
553            ad.droid.telecomCallClearCallList()
554            if num_active_calls(log, ad) != 0:
555                ad.log.error("Phone Call List is not empty.")
556                raise _CallException("Clear call list failed.")
557
558        log.info("Step1: Call From PhoneA to PhoneB.")
559        if not call_setup_teardown(
560                log,
561                ads[0],
562                ads[1],
563                ad_hangup=None,
564                verify_caller_func=verify_func_a,
565                verify_callee_func=verify_func_b):
566            raise _CallException("PhoneA call PhoneB failed.")
567
568        calls = ads[0].droid.telecomCallGetCallIds()
569        ads[0].log.info("Calls in PhoneA %s", calls)
570        if num_active_calls(log, ads[0]) != 1:
571            raise _CallException("Call list verify failed.")
572        call_ab_id = calls[0]
573
574        log.info("Step2: Call From PhoneC to PhoneA.")
575        if reject_once:
576            log.info("Step2-1: Reject incoming call once.")
577            if not initiate_call(
578                log,
579                ads[2],
580                ads[0].telephony['subscription'][get_incoming_voice_sub_id(
581                    ads[0])]['phone_num']):
582                ads[2].log.error("Initiate call failed.")
583                raise _CallException("Failed to initiate call.")
584
585            if not wait_and_reject_call_for_subscription(
586                    log,
587                    ads[0],
588                    get_incoming_voice_sub_id(ads[0]),
589                    incoming_number= \
590                        ads[2].telephony['subscription'][
591                            get_incoming_voice_sub_id(
592                                ads[2])]['phone_num']):
593                ads[0].log.error("Reject call fail.")
594                raise _CallException("Failed to reject call.")
595
596            _hangup_call(log, ads[2], "PhoneC")
597            time.sleep(15)
598
599        if not call_setup_teardown(
600                log,
601                ads[2],
602                ads[0],
603                ad_hangup=None,
604                verify_caller_func=verify_func_c,
605                verify_callee_func=verify_func_a):
606            raise _CallException("PhoneA call PhoneC failed.")
607        if not verify_incall_state(log, [ads[0], ads[1], ads[2]],
608                                   True):
609            raise _CallException("Not All phones are in-call.")
610
611    except _CallException:
612        return None
613    return call_ab_id
614
615
616def _test_call_mo_mt_add_swap_x(log,
617                                ads,
618                                num_swaps,
619                                phone_setup_a=None,
620                                phone_setup_b=None,
621                                phone_setup_c=None,
622                                verify_phone_a_network_subscription=None,
623                                verify_phone_b_network_subscription=None,
624                                verify_phone_c_network_subscription=None):
625    """Test swap feature in VoLTE call.
626
627    PhoneA call PhoneB, accept on PhoneB.
628    PhoneC call PhoneA, accept on PhoneA.
629    Swap active call on PhoneA. (N times)
630
631    Args:
632        num_swaps: do swap for 'num_swaps' times.
633            This value can be 0 (no swap operation).
634
635    Returns:
636        call_ab_id, call_ac_id if succeed;
637        None, None if failed.
638
639    """
640    if ((phone_setup_a == phone_setup_voice_3g) or (phone_setup_a == phone_setup_voice_2g)):
641        # make sure PhoneA is GSM phone before proceed.
642        if (ads[0].droid.telephonyGetPhoneType() != PHONE_TYPE_GSM):
643            raise signals.TestSkip("not GSM phone, abort swap test.")
644
645    if (((phone_setup_b == phone_setup_voice_3g)
646        and (phone_setup_c == phone_setup_voice_3g)) or
647        ((phone_setup_b == phone_setup_voice_2g)
648        and (phone_setup_c == phone_setup_voice_2g))):
649        # make sure PhoneB and PhoneC are GSM phone before proceed.
650        for ad in [ads[1], ads[2]]:
651            if (ad.droid.telephonyGetPhoneType() != PHONE_TYPE_GSM):
652                raise signals.TestSkip("not GSM phone, abort swap test.")
653
654    call_ab_id = _three_phone_call_mo_add_mt(log,
655        [ads[0], ads[1], ads[2]],
656        [phone_setup_a, phone_setup_b, phone_setup_c], [
657            verify_phone_a_network_subscription, verify_phone_b_network_subscription,
658            verify_phone_c_network_subscription
659            ])
660    if call_ab_id is None:
661        log.error("Failed to get call_ab_id")
662        return None, None
663
664    calls = ads[0].droid.telecomCallGetCallIds()
665    ads[0].log.info("Calls in PhoneA %s", calls)
666    if num_active_calls(log, ads[0]) != 2:
667        return None, None
668    if calls[0] == call_ab_id:
669        call_ac_id = calls[1]
670    else:
671        call_ac_id = calls[0]
672
673    if num_swaps > 0:
674        log.info("Step3: Begin Swap x%s test.", num_swaps)
675        if not swap_calls(log, ads, call_ab_id, call_ac_id,
676                          num_swaps):
677            log.error("Swap test failed.")
678            return None, None
679
680    return call_ab_id, call_ac_id
681
682
683def _three_phone_call_mt_add_mt(log, ads, phone_setups, verify_funcs):
684    """Use 3 phones to make MT call and MT call.
685
686    Call from PhoneB to PhoneA, accept on PhoneA.
687    Call from PhoneC to PhoneA, accept on PhoneA.
688
689    Args:
690        ads: list of ad object.
691            The list should have three objects.
692        phone_setups: list of phone setup functions.
693            The list should have three objects.
694        verify_funcs: list of phone call verify functions.
695            The list should have three objects.
696
697    Returns:
698        If success, return 'call_AB' id in PhoneA.
699        if fail, return None.
700    """
701
702    class _CallException(Exception):
703        pass
704
705    try:
706        verify_func_a, verify_func_b, verify_func_c = verify_funcs
707        tasks = []
708        for ad, setup_func in zip(ads, phone_setups):
709            if setup_func is not None:
710                tasks.append((setup_func, (log, ad)))
711        if tasks != [] and not multithread_func(log, tasks):
712            log.error("Phone Failed to Set Up Properly.")
713            raise _CallException("Setup failed.")
714        for ad in ads:
715            ad.droid.telecomCallClearCallList()
716            if num_active_calls(log, ad) != 0:
717                ad.log.error("Phone Call List is not empty.")
718                raise _CallException("Clear call list failed.")
719
720        log.info("Step1: Call From PhoneB to PhoneA.")
721        if not call_setup_teardown(
722                log,
723                ads[1],
724                ads[0],
725                ad_hangup=None,
726                verify_caller_func=verify_func_b,
727                verify_callee_func=verify_func_a):
728            raise _CallException("PhoneB call PhoneA failed.")
729
730        calls = ads[0].droid.telecomCallGetCallIds()
731        ads[0].log.info("Calls in PhoneA %s", calls)
732        if num_active_calls(log, ads[0]) != 1:
733            raise _CallException("Call list verify failed.")
734        call_ab_id = calls[0]
735
736        log.info("Step2: Call From PhoneC to PhoneA.")
737        if not call_setup_teardown(
738                log,
739                ads[2],
740                ads[0],
741                ad_hangup=None,
742                verify_caller_func=verify_func_c,
743                verify_callee_func=verify_func_a):
744            raise _CallException("PhoneA call PhoneC failed.")
745        if not verify_incall_state(log, [ads[0], ads[1], ads[2]],
746                                  True):
747            raise _CallException("Not All phones are in-call.")
748
749    except _CallException:
750        return None
751
752    return call_ab_id
753
754def _test_call_mt_mt_add_swap_x(log,
755                                ads,
756                                num_swaps,
757                                phone_setup_a=None,
758                                phone_setup_b=None,
759                                phone_setup_c=None,
760                                verify_phone_a_network_subscription=None,
761                                verify_phone_b_network_subscription=None,
762                                verify_phone_c_network_subscription=None):
763    """Test swap feature in VoLTE call.
764
765    PhoneB call PhoneA, accept on PhoneA.
766    PhoneC call PhoneA, accept on PhoneA.
767    Swap active call on PhoneA. (N times)
768
769    Args:
770        num_swaps: do swap for 'num_swaps' times.
771            This value can be 0 (no swap operation).
772
773    Returns:
774        call_ab_id, call_ac_id if succeed;
775        None, None if failed.
776
777    """
778    if ((phone_setup_a == phone_setup_voice_3g) or (phone_setup_a == phone_setup_voice_2g)):
779        # make sure PhoneA is GSM phone before proceeSSd.
780        if (ads[0].droid.telephonyGetPhoneType() != PHONE_TYPE_GSM):
781            raise signals.TestSkip("not GSM phone, abort swap test.")
782
783    if (((phone_setup_b == phone_setup_voice_3g)
784        and (phone_setup_c == phone_setup_voice_3g)) or
785        ((phone_setup_b == phone_setup_voice_2g)
786        and (phone_setup_c == phone_setup_voice_2g))):
787        # make sure PhoneB and PhoneC are GSM phone before proceed.
788        for ad in [ads[1], ads[2]]:
789            if (ad.droid.telephonyGetPhoneType() != PHONE_TYPE_GSM):
790                raise signals.TestSkip("not GSM phone, abort swap test.")
791
792    call_ab_id = _three_phone_call_mt_add_mt(log,
793        [ads[0], ads[1], ads[2]],
794        [phone_setup_a, phone_setup_b, phone_setup_c], [
795            verify_phone_a_network_subscription, verify_phone_b_network_subscription,
796            verify_phone_c_network_subscription
797        ])
798    if call_ab_id is None:
799        log.error("Failed to get call_ab_id")
800        return None, None
801
802    calls = ads[0].droid.telecomCallGetCallIds()
803    ads[0].log.info("Calls in PhoneA %s", calls)
804    if num_active_calls(log, ads[0]) != 2:
805        return None, None
806    if calls[0] == call_ab_id:
807        call_ac_id = calls[1]
808    else:
809        call_ac_id = calls[0]
810
811    if num_swaps > 0:
812        log.info("Step3: Begin Swap x%s test.", num_swaps)
813        if not swap_calls(log, ads, call_ab_id, call_ac_id,
814                          num_swaps):
815            log.error("Swap test failed.")
816            return None, None
817
818    return call_ab_id, call_ac_id
819
820
821def _three_phone_hangup_call_verify_call_state(
822        log, ad_hangup, ad_verify, call_id, call_state, ads_active):
823    """Private Test utility for swap test.
824
825    Hangup on 'ad_hangup'.
826    Verify 'call_id' on 'ad_verify' is in expected 'call_state'
827    Verify each ad in ads_active are 'in-call'.
828
829    Args:
830        ad_hangup: android object to hangup call.
831        ad_verify: android object to verify call id state.
832        call_id: call id in 'ad_verify'.
833        call_state: expected state for 'call_id'.
834            'call_state' is either CALL_STATE_HOLDING or CALL_STATE_ACTIVE.
835        ads_active: list of android object.
836            Each one of them should be 'in-call' after 'hangup' operation.
837
838    Returns:
839        True if no error happened. Otherwise False.
840
841    """
842    ad_hangup.log.info("Hangup, verify call continues.")
843    if not _hangup_call(log, ad_hangup):
844        ad_hangup.log.error("Phone fails to hang up")
845        return False
846    time.sleep(WAIT_TIME_IN_CALL)
847
848    if ad_verify.droid.telecomCallGetCallState(call_id) != call_state:
849        ad_verify.log.error(
850            "Call_id: %s, state: %s, expected: %s", call_id,
851            ad_verify.droid.telecomCallGetCallState(call_id), call_state)
852        return False
853    ad_verify.log.info("Call in expected %s state", call_state)
854
855    if not verify_incall_state(log, ads_active, True):
856        ads_active.log.error("Phone not in call state")
857        return False
858    if not verify_incall_state(log, [ad_hangup], False):
859        ad_hangup.log.error("Phone not in hangup state")
860        return False
861
862    return True
863
864
865def _get_expected_call_state(ad):
866    if "vzw" in [
867            sub["operator"]
868            for sub in ad.telephony["subscription"].values()
869    ]:
870        return CALL_STATE_ACTIVE
871    return CALL_STATE_HOLDING
872
873def _test_wcdma_conference_merge_drop(log, ads, call_ab_id, call_ac_id):
874    """Test conference merge and drop in WCDMA/CSFB_WCDMA call.
875
876    PhoneA in WCDMA (or CSFB_WCDMA) call with PhoneB.
877    PhoneA in WCDMA (or CSFB_WCDMA) call with PhoneC.
878    Merge calls to conference on PhoneA.
879    Hangup on PhoneC, check call continues between AB.
880    Hangup on PhoneB, check A ends.
881
882    Args:
883        call_ab_id: call id for call_AB on PhoneA.
884        call_ac_id: call id for call_AC on PhoneA.
885
886    Returns:
887        True if succeed;
888        False if failed.
889    """
890    log.info("Step4: Merge to Conf Call and verify Conf Call.")
891    ads[0].droid.telecomCallJoinCallsInConf(call_ab_id, call_ac_id)
892    time.sleep(WAIT_TIME_IN_CALL)
893    calls = ads[0].droid.telecomCallGetCallIds()
894    ads[0].log.info("Calls in PhoneA %s", calls)
895    num_calls = num_active_calls(log, ads[0])
896    if num_calls != 3:
897        ads[0].log.error("Total number of call ids is not 3.")
898        if num_calls == 2:
899            if call_ab_id in calls and call_ac_id in calls:
900                ads[0].log.error("Calls were not merged."
901                    " Failed to merge calls.")
902                raise signals.TestFailure(
903                    "Calls were not merged. Failed to merge calls.",
904                    extras={"fail_reason": "Calls were not merged."
905                        " Failed to merge calls."})
906        return False
907    call_conf_id = None
908    for call_id in calls:
909        if call_id != call_ab_id and call_id != call_ac_id:
910            call_conf_id = call_id
911    if not call_conf_id:
912        log.error("Merge call fail, no new conference call id.")
913        return False
914    if not verify_incall_state(log, [ads[0], ads[1], ads[2]], True):
915        return False
916
917    if ads[0].droid.telecomCallGetCallState(
918            call_conf_id) != CALL_STATE_ACTIVE:
919        ads[0].log.error(
920            "Call_id: %s, state: %s, expected: STATE_ACTIVE", call_conf_id,
921            ads[0].droid.telecomCallGetCallState(call_conf_id))
922        return False
923
924    log.info("Step5: End call on PhoneC and verify call continues.")
925    if not _hangup_call(log, ads[2], "PhoneC"):
926        return False
927    time.sleep(WAIT_TIME_IN_CALL)
928    calls = ads[0].droid.telecomCallGetCallIds()
929    ads[0].log.info("Calls in PhoneA %s", calls)
930    if num_active_calls(log, ads[0]) != 1:
931        return False
932    if not verify_incall_state(log, [ads[0], ads[1]], True):
933        return False
934    if not verify_incall_state(log, [ads[2]], False):
935        return False
936
937    log.info("Step6: End call on PhoneB and verify PhoneA end.")
938    if not _hangup_call(log, ads[1], "PhoneB"):
939        return False
940    time.sleep(WAIT_TIME_IN_CALL)
941    if not verify_incall_state(log, [ads[0], ads[1], ads[2]], False):
942        return False
943    return True