#!/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 re import time from queue import Empty from acts import signals from acts.logger import epoch_to_log_line_timestamp from acts.utils import get_current_epoch_time from acts_contrib.test_utils.tel.loggers.protos.telephony_metric_pb2 import TelephonyVoiceTestResult from acts_contrib.test_utils.tel.tel_defines import CarrierConfigs from acts_contrib.test_utils.tel.tel_defines import CARRIER_NTT_DOCOMO, CARRIER_KDDI, CARRIER_RAKUTEN, CARRIER_SBM from acts_contrib.test_utils.tel.tel_defines import CALL_PROPERTY_HIGH_DEF_AUDIO 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 CAPABILITY_VOLTE from acts_contrib.test_utils.tel.tel_defines import CAPABILITY_WFC from acts_contrib.test_utils.tel.tel_defines import DIRECTION_MOBILE_ORIGINATED from acts_contrib.test_utils.tel.tel_defines import GEN_2G from acts_contrib.test_utils.tel.tel_defines import GEN_3G from acts_contrib.test_utils.tel.tel_defines import INCALL_UI_DISPLAY_BACKGROUND from acts_contrib.test_utils.tel.tel_defines import INCALL_UI_DISPLAY_FOREGROUND from acts_contrib.test_utils.tel.tel_defines import INVALID_SUB_ID from acts_contrib.test_utils.tel.tel_defines import MAX_SAVED_VOICE_MAIL from acts_contrib.test_utils.tel.tel_defines import MAX_WAIT_TIME_ACCEPT_CALL_TO_OFFHOOK_EVENT from acts_contrib.test_utils.tel.tel_defines import MAX_WAIT_TIME_CALL_DROP from acts_contrib.test_utils.tel.tel_defines import MAX_WAIT_TIME_CALL_IDLE_EVENT from acts_contrib.test_utils.tel.tel_defines import MAX_WAIT_TIME_CALL_INITIATION from acts_contrib.test_utils.tel.tel_defines import MAX_WAIT_TIME_CALLEE_RINGING from acts_contrib.test_utils.tel.tel_defines import MAX_WAIT_TIME_TELECOM_RINGING from acts_contrib.test_utils.tel.tel_defines import MAX_WAIT_TIME_VOICE_MAIL_COUNT from acts_contrib.test_utils.tel.tel_defines import NETWORK_SERVICE_DATA from acts_contrib.test_utils.tel.tel_defines import NETWORK_SERVICE_VOICE from acts_contrib.test_utils.tel.tel_defines import RAT_1XRTT from acts_contrib.test_utils.tel.tel_defines import RAT_IWLAN from acts_contrib.test_utils.tel.tel_defines import RAT_LTE from acts_contrib.test_utils.tel.tel_defines import RAT_UMTS from acts_contrib.test_utils.tel.tel_defines import RAT_UNKNOWN from acts_contrib.test_utils.tel.tel_defines import TELEPHONY_STATE_IDLE from acts_contrib.test_utils.tel.tel_defines import TELEPHONY_STATE_OFFHOOK from acts_contrib.test_utils.tel.tel_defines import TELEPHONY_STATE_RINGING from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_ANDROID_STATE_SETTLING 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_BETWEEN_STATE_CHECK from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_IN_CALL from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_LEAVE_VOICE_MAIL from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_REJECT_CALL from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_VOICE_MAIL_SERVER_RESPONSE from acts_contrib.test_utils.tel.tel_defines import WFC_MODE_WIFI_PREFERRED from acts_contrib.test_utils.tel.tel_defines import EventCallStateChanged from acts_contrib.test_utils.tel.tel_defines import EventMessageWaitingIndicatorChanged from acts_contrib.test_utils.tel.tel_defines import CallStateContainer from acts_contrib.test_utils.tel.tel_defines import MessageWaitingIndicatorContainer from acts_contrib.test_utils.tel.tel_ims_utils import is_wfc_enabled from acts_contrib.test_utils.tel.tel_ims_utils import toggle_volte from acts_contrib.test_utils.tel.tel_ims_utils import toggle_wfc from acts_contrib.test_utils.tel.tel_ims_utils import set_wfc_mode from acts_contrib.test_utils.tel.tel_ims_utils import wait_for_volte_enabled from acts_contrib.test_utils.tel.tel_ims_utils import wait_for_wfc_enabled from acts_contrib.test_utils.tel.tel_ims_utils import wait_for_wfc_disabled from acts_contrib.test_utils.tel.tel_lookup_tables import get_voice_mail_delete_digit from acts_contrib.test_utils.tel.tel_phone_setup_utils import ensure_phones_idle from acts_contrib.test_utils.tel.tel_phone_setup_utils import wait_for_network_rat from acts_contrib.test_utils.tel.tel_phone_setup_utils import ensure_phone_subscription from acts_contrib.test_utils.tel.tel_phone_setup_utils import wait_for_not_network_rat from acts_contrib.test_utils.tel.tel_phone_setup_utils import wait_for_voice_attach from acts_contrib.test_utils.tel.tel_subscription_utils import get_incoming_voice_sub_id from acts_contrib.test_utils.tel.tel_subscription_utils import get_outgoing_voice_sub_id from acts_contrib.test_utils.tel.tel_subscription_utils import set_subid_for_outgoing_call from acts_contrib.test_utils.tel.tel_subscription_utils import get_subid_from_slot_index from acts_contrib.test_utils.tel.tel_test_utils import _wait_for_droid_in_state from acts_contrib.test_utils.tel.tel_test_utils import check_call_state_connected_by_adb from acts_contrib.test_utils.tel.tel_test_utils import check_call_state_idle_by_adb from acts_contrib.test_utils.tel.tel_test_utils import check_phone_number_match from acts_contrib.test_utils.tel.tel_test_utils import check_voice_mail_count from acts_contrib.test_utils.tel.tel_test_utils import check_voice_network_type from acts_contrib.test_utils.tel.tel_test_utils import get_call_uri from acts_contrib.test_utils.tel.tel_test_utils import get_device_epoch_time from acts_contrib.test_utils.tel.tel_test_utils import get_network_gen_for_subscription from acts_contrib.test_utils.tel.tel_test_utils import get_network_rat from acts_contrib.test_utils.tel.tel_test_utils import get_network_rat_for_subscription from acts_contrib.test_utils.tel.tel_test_utils import get_number_from_tel_uri from acts_contrib.test_utils.tel.tel_test_utils import get_operator_name from acts_contrib.test_utils.tel.tel_test_utils import get_user_config_profile from acts_contrib.test_utils.tel.tel_test_utils import get_voice_mail_number from acts_contrib.test_utils.tel.tel_test_utils import is_event_match from acts_contrib.test_utils.tel.tel_test_utils import is_event_match_for_list from acts_contrib.test_utils.tel.tel_test_utils import num_active_calls from acts_contrib.test_utils.tel.tel_test_utils import TelResultWrapper from acts_contrib.test_utils.tel.tel_test_utils import toggle_airplane_mode_by_adb from acts_contrib.test_utils.tel.tel_test_utils import verify_incall_state from acts_contrib.test_utils.tel.tel_test_utils import wait_for_state from acts_contrib.test_utils.tel.tel_wifi_utils import ensure_wifi_connected from acts_contrib.test_utils.tel.tel_wifi_utils import wifi_toggle_state CallResult = TelephonyVoiceTestResult.CallResult.Value result_dict ={} voice_call_type = {} def check_call_status(ad, voice_type_init=None, voice_type_in_call=None): """" Args: ad: Android device object voice_type_init: Voice network type before initiate call voice_type_in_call: Voice network type in call state Return: voice_call_type_dict: Voice call status """ dut = str(ad.serial) network_type = voice_type_init + "_" + voice_type_in_call if network_type == "NR_NR": voice_call_type_dict = update_voice_call_type_dict(dut, "VoNR") elif network_type == "NR_LTE": voice_call_type_dict = update_voice_call_type_dict(dut, "EPSFB") elif network_type == "LTE_LTE": voice_call_type_dict = update_voice_call_type_dict(dut, "VoLTE") elif (network_type == "LTE_WCDMA" or network_type == "LTE_EDGE" or network_type == "LTE_GSM"): voice_call_type_dict = update_voice_call_type_dict(dut, "CSFB") elif network_type == "EDGE_EDGE": voice_call_type_dict = update_voice_call_type_dict(dut, "EDGE") elif network_type == "GSM_GSM": voice_call_type_dict = update_voice_call_type_dict(dut, "GSM") else: voice_call_type_dict = update_voice_call_type_dict(dut, "UNKNOWN") return voice_call_type_dict def update_voice_call_type_dict(dut, key): """ Args: dut: Serial Number of android device object key: Network subscription parameter (VoNR or EPSFB or VoLTE or CSFB or UNKNOWN) Return: voice_call_type: Voice call status """ if dut not in voice_call_type.keys(): voice_call_type[dut] = {key:0} if key not in voice_call_type[dut].keys(): voice_call_type[dut].update({key:0}) voice_call_type[dut][key] += 1 return voice_call_type def dial_phone_number(ad, callee_number): for number in str(callee_number): if number == "#": ad.send_keycode("POUND") elif number == "*": ad.send_keycode("STAR") elif number in ["1", "2", "3", "4", "5", "6", "7", "8", "9", "0"]: ad.send_keycode("%s" % number) def disconnect_call_by_id(log, ad, call_id): """Disconnect call by call id. """ ad.droid.telecomCallDisconnect(call_id) return True def dumpsys_last_call_info(ad): """ Get call information by dumpsys telecom. """ num = dumpsys_last_call_number(ad) output = ad.adb.shell("dumpsys telecom") result = re.search(r"Call TC@%s: {(.*?)}" % num, output, re.DOTALL) call_info = {"TC": num} if result: result = result.group(1) for attr in ("startTime", "endTime", "direction", "isInterrupted", "callTechnologies", "callTerminationsReason", "isVideoCall", "callProperties"): match = re.search(r"%s: (.*)" % attr, result) if match: if attr in ("startTime", "endTime"): call_info[attr] = epoch_to_log_line_timestamp( int(match.group(1))) else: call_info[attr] = match.group(1) ad.log.debug("call_info = %s", call_info) return call_info def dumpsys_last_call_number(ad): output = ad.adb.shell("dumpsys telecom") call_nums = re.findall("Call TC@(\d+):", output) if not call_nums: return 0 else: return int(call_nums[-1]) def dumpsys_new_call_info(ad, last_tc_number, retries=3, interval=5): for i in range(retries): if dumpsys_last_call_number(ad) > last_tc_number: call_info = dumpsys_last_call_info(ad) ad.log.info("New call info = %s", sorted(call_info.items())) return call_info else: time.sleep(interval) ad.log.error("New call is not in sysdump telecom") return {} def emergency_dialer_call_by_keyevent(ad, callee_number): for i in range(3): if "EmergencyDialer" in ad.get_my_current_focus_window(): ad.log.info("EmergencyDialer is the current focus window") break elif i <= 2: ad.adb.shell("am start -a com.android.phone.EmergencyDialer.DIAL") time.sleep(1) else: ad.log.error("Unable to bring up EmergencyDialer") return False ad.log.info("Make a phone call to %s", callee_number) dial_phone_number(ad, callee_number) ad.send_keycode("CALL") def get_current_voice_rat(log, ad): """Return current Voice RAT Args: ad: Android device object. """ return get_current_voice_rat_for_subscription( log, ad, get_outgoing_voice_sub_id(ad)) def get_current_voice_rat_for_subscription(log, ad, sub_id): """Return current Voice RAT for subscription id. Args: ad: Android device object. sub_id: subscription id. """ return get_network_rat_for_subscription(log, ad, sub_id, NETWORK_SERVICE_VOICE) def hangup_call_by_adb(ad): """Make emergency call by EmergencyDialer. Args: ad: Caller android device object. callee_number: Callee phone number. """ ad.log.info("End call by adb") ad.send_keycode("ENDCALL") def hangup_call(log, ad, is_emergency=False): """Hang up ongoing active call. Args: log: log object. ad: android device object. Returns: True: if all calls are cleared False: for errors """ # short circuit in case no calls are active if not ad.droid.telecomIsInCall(): ad.log.warning("No active call exists.") return True ad.ed.clear_events(EventCallStateChanged) ad.droid.telephonyStartTrackingCallState() ad.log.info("Hangup call.") if is_emergency: for call in ad.droid.telecomCallGetCallIds(): ad.droid.telecomCallDisconnect(call) else: ad.droid.telecomEndCall() try: ad.ed.wait_for_event( EventCallStateChanged, is_event_match, timeout=MAX_WAIT_TIME_CALL_IDLE_EVENT, field=CallStateContainer.CALL_STATE, value=TELEPHONY_STATE_IDLE) except Empty: ad.log.warning("Call state IDLE event is not received after hang up.") finally: ad.droid.telephonyStopTrackingCallStateChange() if not wait_for_state(ad.droid.telecomIsInCall, False, 15, 1): ad.log.error("Telecom is in call, hangup call failed.") return False return True def initiate_emergency_dialer_call_by_adb( log, ad, callee_number, timeout=MAX_WAIT_TIME_CALL_INITIATION, checking_interval=5): """Make emergency call by EmergencyDialer. Args: ad: Caller android device object. callee_number: Callee phone number. emergency : specify the call is emergency. Optional. Default value is False. Returns: result: if phone call is placed successfully. """ try: # Make a Call ad.wakeup_screen() ad.send_keycode("MENU") ad.log.info("Call %s", callee_number) ad.adb.shell("am start -a com.android.phone.EmergencyDialer.DIAL") ad.adb.shell( "am start -a android.intent.action.CALL_EMERGENCY -d tel:%s" % callee_number) if not timeout: return True ad.log.info("Check call state") # Verify Call State elapsed_time = 0 while elapsed_time < timeout: time.sleep(checking_interval) elapsed_time += checking_interval if check_call_state_connected_by_adb(ad): ad.log.info("Call to %s is connected", callee_number) return True if check_call_state_idle_by_adb(ad): ad.log.info("Call to %s failed", callee_number) return False ad.log.info("Make call to %s failed", callee_number) return False except Exception as e: ad.log.error("initiate emergency call failed with error %s", e) def initiate_call(log, ad, callee_number, emergency=False, incall_ui_display=INCALL_UI_DISPLAY_FOREGROUND, video=False): """Make phone call from caller to callee. Args: ad_caller: Caller android device object. callee_number: Callee phone number. emergency : specify the call is emergency. Optional. Default value is False. incall_ui_display: show the dialer UI foreground or backgroud video: whether to initiate as video call Returns: result: if phone call is placed successfully. """ ad.ed.clear_events(EventCallStateChanged) sub_id = get_outgoing_voice_sub_id(ad) begin_time = get_device_epoch_time(ad) ad.droid.telephonyStartTrackingCallStateForSubscription(sub_id) try: # Make a Call ad.log.info("Make a phone call to %s", callee_number) if emergency: ad.droid.telecomCallEmergencyNumber(callee_number) else: ad.droid.telecomCallNumber(callee_number, video) # Verify OFFHOOK state if not wait_for_call_offhook_for_subscription( log, ad, sub_id, event_tracking_started=True): ad.log.info("sub_id %s not in call offhook state", sub_id) last_call_drop_reason(ad, begin_time=begin_time) return False else: return True finally: if hasattr(ad, "sdm_log") and getattr(ad, "sdm_log"): ad.adb.shell("i2cset -fy 3 64 6 1 b", ignore_status=True) ad.adb.shell("i2cset -fy 3 65 6 1 b", ignore_status=True) ad.droid.telephonyStopTrackingCallStateChangeForSubscription(sub_id) if incall_ui_display == INCALL_UI_DISPLAY_FOREGROUND: ad.droid.telecomShowInCallScreen() elif incall_ui_display == INCALL_UI_DISPLAY_BACKGROUND: ad.droid.showHomeScreen() def last_call_drop_reason(ad, begin_time=None): reasons = ad.search_logcat( "qcril_qmi_voice_map_qmi_to_ril_last_call_failure_cause", begin_time) reason_string = "" if reasons: log_msg = "Logcat call drop reasons:" for reason in reasons: log_msg = "%s\n\t%s" % (log_msg, reason["log_message"]) if "ril reason str" in reason["log_message"]: reason_string = reason["log_message"].split(":")[-1].strip() ad.log.info(log_msg) reasons = ad.search_logcat("ACTION_FORBIDDEN_NO_SERVICE_AUTHORIZATION", begin_time) if reasons: ad.log.warning("ACTION_FORBIDDEN_NO_SERVICE_AUTHORIZATION is seen") ad.log.info("last call dumpsys: %s", sorted(dumpsys_last_call_info(ad).items())) return reason_string def call_reject(log, ad_caller, ad_callee, reject=True): """Caller call Callee, then reject on callee. """ subid_caller = ad_caller.droid.subscriptionGetDefaultVoiceSubId() subid_callee = ad_callee.incoming_voice_sub_id ad_caller.log.info("Sub-ID Caller %s, Sub-ID Callee %s", subid_caller, subid_callee) return call_reject_for_subscription(log, ad_caller, ad_callee, subid_caller, subid_callee, reject) def call_reject_for_subscription(log, ad_caller, ad_callee, subid_caller, subid_callee, reject=True): """ """ caller_number = ad_caller.telephony['subscription'][subid_caller][ 'phone_num'] callee_number = ad_callee.telephony['subscription'][subid_callee][ 'phone_num'] ad_caller.log.info("Call from %s to %s", caller_number, callee_number) if not initiate_call(log, ad_caller, callee_number): ad_caller.log.error("Initiate call failed") return False if not wait_and_reject_call_for_subscription( log, ad_callee, subid_callee, caller_number, WAIT_TIME_REJECT_CALL, reject): ad_callee.log.error("Reject call fail.") return False # Check if incoming call is cleared on callee or not. if ad_callee.droid.telephonyGetCallStateForSubscription( subid_callee) == TELEPHONY_STATE_RINGING: ad_callee.log.error("Incoming call is not cleared") return False # Hangup on caller hangup_call(log, ad_caller) return True def call_reject_leave_message(log, ad_caller, ad_callee, verify_caller_func=None, wait_time_in_call=WAIT_TIME_LEAVE_VOICE_MAIL): """On default voice subscription, Call from caller to callee, reject on callee, caller leave a voice mail. 1. Caller call Callee. 2. Callee reject incoming call. 3. Caller leave a voice mail. 4. Verify callee received the voice mail notification. Args: ad_caller: caller android device object. ad_callee: callee android device object. verify_caller_func: function to verify caller is in correct state while in-call. This is optional, default is None. wait_time_in_call: time to wait when leaving a voice mail. This is optional, default is WAIT_TIME_LEAVE_VOICE_MAIL Returns: True: if voice message is received on callee successfully. False: for errors """ subid_caller = get_outgoing_voice_sub_id(ad_caller) subid_callee = get_incoming_voice_sub_id(ad_callee) return call_reject_leave_message_for_subscription( log, ad_caller, ad_callee, subid_caller, subid_callee, verify_caller_func, wait_time_in_call) def check_reject_needed_for_voice_mail(log, ad_callee): """Check if the carrier requires reject call to receive voice mail or just keep ringing Requested in b//155935290 Four Japan carriers do not need to reject SBM, KDDI, Ntt Docomo, Rakuten Args: log: log object ad_callee: android device object Returns: True if callee's carrier is not one of the four Japan carriers False if callee's carrier is one of the four Japan carriers """ operators_no_reject = [CARRIER_NTT_DOCOMO, CARRIER_KDDI, CARRIER_RAKUTEN, CARRIER_SBM] operator_name = get_operator_name(log, ad_callee) return operator_name not in operators_no_reject def _is_on_message_waiting_event_true(event): """Private function to return if the received EventMessageWaitingIndicatorChanged event MessageWaitingIndicatorContainer.IS_MESSAGE_WAITING field is True. """ return event['data'][MessageWaitingIndicatorContainer.IS_MESSAGE_WAITING] def call_reject_leave_message_for_subscription( log, ad_caller, ad_callee, subid_caller, subid_callee, verify_caller_func=None, wait_time_in_call=WAIT_TIME_LEAVE_VOICE_MAIL): """On specific voice subscription, Call from caller to callee, reject on callee, caller leave a voice mail. 1. Caller call Callee. 2. Callee reject incoming call. 3. Caller leave a voice mail. 4. Verify callee received the voice mail notification. Args: ad_caller: caller android device object. ad_callee: callee android device object. subid_caller: caller's subscription id. subid_callee: callee's subscription id. verify_caller_func: function to verify caller is in correct state while in-call. This is optional, default is None. wait_time_in_call: time to wait when leaving a voice mail. This is optional, default is WAIT_TIME_LEAVE_VOICE_MAIL Returns: True: if voice message is received on callee successfully. False: for errors """ # Currently this test utility only works for TMO and ATT and SPT. # It does not work for VZW (see b/21559800) # "with VVM TelephonyManager APIs won't work for vm" caller_number = ad_caller.telephony['subscription'][subid_caller][ 'phone_num'] callee_number = ad_callee.telephony['subscription'][subid_callee][ 'phone_num'] ad_caller.log.info("Call from %s to %s", caller_number, callee_number) try: voice_mail_count_before = ad_callee.droid.telephonyGetVoiceMailCountForSubscription( subid_callee) ad_callee.log.info("voice mail count is %s", voice_mail_count_before) # -1 means there are unread voice mail, but the count is unknown # 0 means either this API not working (VZW) or no unread voice mail. if voice_mail_count_before != 0: log.warning("--Pending new Voice Mail, please clear on phone.--") if not initiate_call(log, ad_caller, callee_number): ad_caller.log.error("Initiate call failed.") return False if check_reject_needed_for_voice_mail(log, ad_callee): carrier_specific_delay_reject = 30 else: carrier_specific_delay_reject = 2 carrier_reject_call = not check_reject_needed_for_voice_mail(log, ad_callee) if not wait_and_reject_call_for_subscription( log, ad_callee, subid_callee, incoming_number=caller_number, delay_reject=carrier_specific_delay_reject, reject=carrier_reject_call): ad_callee.log.error("Reject call fail.") return False ad_callee.droid.telephonyStartTrackingVoiceMailStateChangeForSubscription( subid_callee) # ensure that all internal states are updated in telecom time.sleep(WAIT_TIME_ANDROID_STATE_SETTLING) ad_callee.ed.clear_events(EventCallStateChanged) if verify_caller_func and not verify_caller_func(log, ad_caller): ad_caller.log.error("Caller not in correct state!") return False # TODO: b/26293512 Need to play some sound to leave message. # Otherwise carrier voice mail server may drop this voice mail. time.sleep(wait_time_in_call) if not verify_caller_func: caller_state_result = ad_caller.droid.telecomIsInCall() else: caller_state_result = verify_caller_func(log, ad_caller) if not caller_state_result: ad_caller.log.error("Caller not in correct state after %s seconds", wait_time_in_call) if not hangup_call(log, ad_caller): ad_caller.log.error("Error in Hanging-Up Call") return False ad_callee.log.info("Wait for voice mail indicator on callee.") try: event = ad_callee.ed.wait_for_event( EventMessageWaitingIndicatorChanged, _is_on_message_waiting_event_true) ad_callee.log.info("Got event %s", event) except Empty: ad_callee.log.warning("No expected event %s", EventMessageWaitingIndicatorChanged) return False voice_mail_count_after = ad_callee.droid.telephonyGetVoiceMailCountForSubscription( subid_callee) ad_callee.log.info( "telephonyGetVoiceMailCount output - before: %s, after: %s", voice_mail_count_before, voice_mail_count_after) # voice_mail_count_after should: # either equals to (voice_mail_count_before + 1) [For ATT and SPT] # or equals to -1 [For TMO] # -1 means there are unread voice mail, but the count is unknown if not check_voice_mail_count(log, ad_callee, voice_mail_count_before, voice_mail_count_after): log.error("before and after voice mail count is not incorrect.") return False finally: ad_callee.droid.telephonyStopTrackingVoiceMailStateChangeForSubscription( subid_callee) return True def call_voicemail_erase_all_pending_voicemail(log, ad): """Script for phone to erase all pending voice mail. This script only works for TMO and ATT and SPT currently. This script only works if phone have already set up voice mail options, and phone should disable password protection for voice mail. 1. If phone don't have pending voice message, return True. 2. Dial voice mail number. For TMO, the number is '123' For ATT, the number is phone's number For SPT, the number is phone's number 3. Wait for voice mail connection setup. 4. Wait for voice mail play pending voice message. 5. Send DTMF to delete one message. The digit is '7'. 6. Repeat steps 4 and 5 until voice mail server drop this call. (No pending message) 6. Check telephonyGetVoiceMailCount result. it should be 0. Args: log: log object ad: android device object Returns: False if error happens. True is succeed. """ log.info("Erase all pending voice mail.") count = ad.droid.telephonyGetVoiceMailCount() if count == 0: ad.log.info("No Pending voice mail.") return True if count == -1: ad.log.info("There is pending voice mail, but the count is unknown") count = MAX_SAVED_VOICE_MAIL else: ad.log.info("There are %s voicemails", count) voice_mail_number = get_voice_mail_number(log, ad) delete_digit = get_voice_mail_delete_digit(get_operator_name(log, ad)) if not initiate_call(log, ad, voice_mail_number): log.error("Initiate call to voice mail failed.") return False time.sleep(WAIT_TIME_VOICE_MAIL_SERVER_RESPONSE) callId = ad.droid.telecomCallGetCallIds()[0] time.sleep(WAIT_TIME_VOICE_MAIL_SERVER_RESPONSE) while (is_phone_in_call(log, ad) and (count > 0)): ad.log.info("Press %s to delete voice mail.", delete_digit) ad.droid.telecomCallPlayDtmfTone(callId, delete_digit) ad.droid.telecomCallStopDtmfTone(callId) time.sleep(WAIT_TIME_VOICE_MAIL_SERVER_RESPONSE) count -= 1 if is_phone_in_call(log, ad): hangup_call(log, ad) # wait for telephonyGetVoiceMailCount to update correct result remaining_time = MAX_WAIT_TIME_VOICE_MAIL_COUNT while ((remaining_time > 0) and (ad.droid.telephonyGetVoiceMailCount() != 0)): time.sleep(1) remaining_time -= 1 current_voice_mail_count = ad.droid.telephonyGetVoiceMailCount() ad.log.info("telephonyGetVoiceMailCount: %s", current_voice_mail_count) return (current_voice_mail_count == 0) def call_setup_teardown(log, ad_caller, ad_callee, ad_hangup=None, verify_caller_func=None, verify_callee_func=None, wait_time_in_call=WAIT_TIME_IN_CALL, incall_ui_display=INCALL_UI_DISPLAY_FOREGROUND, dialing_number_length=None, video_state=None, slot_id_callee=None, voice_type_init=None, call_stats_check=False, result_info=result_dict): """ Call process, including make a phone call from caller, accept from callee, and hang up. The call is on default voice subscription In call process, call from to , accept the call, (optional)then hang up from . Args: ad_caller: Caller Android Device Object. ad_callee: Callee Android Device Object. ad_hangup: Android Device Object end the phone call. Optional. Default value is None, and phone call will continue. verify_call_mode_caller: func_ptr to verify caller in correct mode Optional. Default is None verify_call_mode_caller: func_ptr to verify caller in correct mode Optional. Default is None incall_ui_display: after answer the call, bring in-call UI to foreground or background. Optional, default value is INCALL_UI_DISPLAY_FOREGROUND. if = INCALL_UI_DISPLAY_FOREGROUND, bring in-call UI to foreground. if = INCALL_UI_DISPLAY_BACKGROUND, bring in-call UI to background. else, do nothing. dialing_number_length: the number of digits used for dialing slot_id_callee : the slot if of the callee to call to Returns: True if call process without any error. False if error happened. """ subid_caller = get_outgoing_voice_sub_id(ad_caller) if slot_id_callee is None: subid_callee = get_incoming_voice_sub_id(ad_callee) else: subid_callee = get_subid_from_slot_index(log, ad_callee, slot_id_callee) return call_setup_teardown_for_subscription( log, ad_caller, ad_callee, subid_caller, subid_callee, ad_hangup, verify_caller_func, verify_callee_func, wait_time_in_call, incall_ui_display, dialing_number_length, video_state, voice_type_init, call_stats_check, result_info) def call_setup_teardown_for_subscription( log, ad_caller, ad_callee, subid_caller, subid_callee, ad_hangup=None, verify_caller_func=None, verify_callee_func=None, wait_time_in_call=WAIT_TIME_IN_CALL, incall_ui_display=INCALL_UI_DISPLAY_FOREGROUND, dialing_number_length=None, video_state=None, voice_type_init=None, call_stats_check=False, result_info=result_dict): """ Call process, including make a phone call from caller, accept from callee, and hang up. The call is on specified subscription In call process, call from to , accept the call, (optional)then hang up from . Args: ad_caller: Caller Android Device Object. ad_callee: Callee Android Device Object. subid_caller: Caller subscription ID subid_callee: Callee subscription ID ad_hangup: Android Device Object end the phone call. Optional. Default value is None, and phone call will continue. verify_call_mode_caller: func_ptr to verify caller in correct mode Optional. Default is None verify_call_mode_caller: func_ptr to verify caller in correct mode Optional. Default is None incall_ui_display: after answer the call, bring in-call UI to foreground or background. Optional, default value is INCALL_UI_DISPLAY_FOREGROUND. if = INCALL_UI_DISPLAY_FOREGROUND, bring in-call UI to foreground. if = INCALL_UI_DISPLAY_BACKGROUND, bring in-call UI to background. else, do nothing. Returns: TelResultWrapper which will evaluate as False if error. """ CHECK_INTERVAL = 5 begin_time = get_current_epoch_time() if not verify_caller_func: verify_caller_func = is_phone_in_call if not verify_callee_func: verify_callee_func = is_phone_in_call caller_number = ad_caller.telephony['subscription'][subid_caller][ 'phone_num'] callee_number = ad_callee.telephony['subscription'][subid_callee][ 'phone_num'] callee_number = truncate_phone_number( log, caller_number, callee_number, dialing_number_length) tel_result_wrapper = TelResultWrapper(CallResult('SUCCESS')) msg = "Call from %s to %s" % (caller_number, callee_number) if video_state: msg = "Video %s" % msg video = True else: video = False if ad_hangup: msg = "%s for duration of %s seconds" % (msg, wait_time_in_call) ad_caller.log.info(msg) for ad in (ad_caller, ad_callee): call_ids = ad.droid.telecomCallGetCallIds() setattr(ad, "call_ids", call_ids) if call_ids: ad.log.info("Pre-exist CallId %s before making call", call_ids) if not initiate_call( log, ad_caller, callee_number, incall_ui_display=incall_ui_display, video=video): ad_caller.log.error("Initiate call failed.") tel_result_wrapper.result_value = CallResult('INITIATE_FAILED') return tel_result_wrapper else: ad_caller.log.info("Caller initate call successfully") if not wait_and_answer_call_for_subscription( log, ad_callee, subid_callee, incoming_number=caller_number, caller=ad_caller, incall_ui_display=incall_ui_display, video_state=video_state): ad_callee.log.error("Answer call fail.") tel_result_wrapper.result_value = CallResult( 'NO_RING_EVENT_OR_ANSWER_FAILED') return tel_result_wrapper else: ad_callee.log.info("Callee answered the call successfully") for ad, call_func in zip([ad_caller, ad_callee], [verify_caller_func, verify_callee_func]): call_ids = ad.droid.telecomCallGetCallIds() new_call_ids = set(call_ids) - set(ad.call_ids) if not new_call_ids: ad.log.error( "No new call ids are found after call establishment") ad.log.error("telecomCallGetCallIds returns %s", ad.droid.telecomCallGetCallIds()) tel_result_wrapper.result_value = CallResult('NO_CALL_ID_FOUND') for new_call_id in new_call_ids: if not wait_for_in_call_active(ad, call_id=new_call_id): tel_result_wrapper.result_value = CallResult( 'CALL_STATE_NOT_ACTIVE_DURING_ESTABLISHMENT') else: ad.log.info("callProperties = %s", ad.droid.telecomCallGetProperties(new_call_id)) if not ad.droid.telecomCallGetAudioState(): ad.log.error("Audio is not in call state") tel_result_wrapper.result_value = CallResult( 'AUDIO_STATE_NOT_INCALL_DURING_ESTABLISHMENT') if call_func(log, ad): ad.log.info("Call is in %s state", call_func.__name__) else: ad.log.error("Call is not in %s state, voice in RAT %s", call_func.__name__, ad.droid.telephonyGetCurrentVoiceNetworkType()) tel_result_wrapper.result_value = CallResult( 'CALL_DROP_OR_WRONG_STATE_DURING_ESTABLISHMENT') if not tel_result_wrapper: return tel_result_wrapper if call_stats_check: voice_type_in_call = check_voice_network_type([ad_caller, ad_callee], voice_init=False) phone_a_call_type = check_call_status(ad_caller, voice_type_init[0], voice_type_in_call[0]) result_info["Call Stats"] = phone_a_call_type ad_caller.log.debug("Voice Call Type: %s", phone_a_call_type) phone_b_call_type = check_call_status(ad_callee, voice_type_init[1], voice_type_in_call[1]) result_info["Call Stats"] = phone_b_call_type ad_callee.log.debug("Voice Call Type: %s", phone_b_call_type) return wait_for_call_end( log, ad_caller, ad_callee, ad_hangup, verify_caller_func, verify_callee_func, begin_time, check_interval=CHECK_INTERVAL, tel_result_wrapper=TelResultWrapper(CallResult('SUCCESS')), wait_time_in_call=wait_time_in_call) def two_phone_call_leave_voice_mail( log, caller, caller_idle_func, caller_in_call_check_func, callee, callee_idle_func, wait_time_in_call=WAIT_TIME_LEAVE_VOICE_MAIL): """Call from caller to callee, reject on callee, caller leave a voice mail. 1. Caller call Callee. 2. Callee reject incoming call. 3. Caller leave a voice mail. 4. Verify callee received the voice mail notification. Args: caller: caller android device object. caller_idle_func: function to check caller's idle state. caller_in_call_check_func: function to check caller's in-call state. callee: callee android device object. callee_idle_func: function to check callee's idle state. wait_time_in_call: time to wait when leaving a voice mail. This is optional, default is WAIT_TIME_LEAVE_VOICE_MAIL Returns: True: if voice message is received on callee successfully. False: for errors """ ads = [caller, callee] # Make sure phones are idle. ensure_phones_idle(log, ads) if caller_idle_func and not caller_idle_func(log, caller): caller.log.error("Caller Failed to Reselect") return False if callee_idle_func and not callee_idle_func(log, callee): callee.log.error("Callee Failed to Reselect") return False # TODO: b/26337871 Need to use proper API to check phone registered. time.sleep(WAIT_TIME_BETWEEN_REG_AND_CALL) # Make call and leave a message. if not call_reject_leave_message( log, caller, callee, caller_in_call_check_func, wait_time_in_call): log.error("make a call and leave a message failed.") return False return True def two_phone_call_short_seq(log, phone_a, phone_a_idle_func, phone_a_in_call_check_func, phone_b, phone_b_idle_func, phone_b_in_call_check_func, call_sequence_func=None, wait_time_in_call=WAIT_TIME_IN_CALL, call_params=None): """Call process short sequence. 1. Ensure phone idle and in idle_func check return True. 2. Call from PhoneA to PhoneB, accept on PhoneB. 3. Check phone state, hangup on PhoneA. 4. Ensure phone idle and in idle_func check return True. 5. Call from PhoneA to PhoneB, accept on PhoneB. 6. Check phone state, hangup on PhoneB. Args: phone_a: PhoneA's android device object. phone_a_idle_func: function to check PhoneA's idle state. phone_a_in_call_check_func: function to check PhoneA's in-call state. phone_b: PhoneB's android device object. phone_b_idle_func: function to check PhoneB's idle state. phone_b_in_call_check_func: function to check PhoneB's in-call state. call_sequence_func: default parameter, not implemented. wait_time_in_call: time to wait in call. This is optional, default is WAIT_TIME_IN_CALL call_params: list of call parameters for both MO/MT devices, including: - MO device object - MT device object - Device object hanging up the call - Function to check the in-call state of MO device - Function to check the in-call state of MT device and the format should be: [(Args for the 1st call), (Args for the 2nd call), ......] The format of args for each call should be: (mo_device_obj, mt_device_obj, device_obj_hanging_up, mo_in_call_check_func, mt_in_call_check_func) Example of a call, which will not be hung up: call_params = [ (ads[0], ads[1], None, mo_in_call_check_func, mt_in_call_check_func) ] Returns: TelResultWrapper which will evaluate as False if error. """ ads = [phone_a, phone_b] if not call_params: call_params = [ (ads[0], ads[1], ads[0], phone_a_in_call_check_func, phone_b_in_call_check_func), (ads[0], ads[1], ads[1], phone_a_in_call_check_func, phone_b_in_call_check_func), ] tel_result = TelResultWrapper(CallResult('SUCCESS')) for param in call_params: # Make sure phones are idle. ensure_phones_idle(log, ads) if phone_a_idle_func and not phone_a_idle_func(log, phone_a): phone_a.log.error("Phone A Failed to Reselect") return TelResultWrapper(CallResult('CALL_SETUP_FAILURE')) if phone_b_idle_func and not phone_b_idle_func(log, phone_b): phone_b.log.error("Phone B Failed to Reselect") return TelResultWrapper(CallResult('CALL_SETUP_FAILURE')) # TODO: b/26337871 Need to use proper API to check phone registered. time.sleep(WAIT_TIME_BETWEEN_REG_AND_CALL) # Make call. log.info("---> Call test: %s to %s <---", param[0].serial, param[1].serial) tel_result = call_setup_teardown( log, *param, wait_time_in_call=wait_time_in_call) if not tel_result: log.error("Call Iteration Failed") break return tel_result def two_phone_call_msim_short_seq(log, phone_a, phone_a_idle_func, phone_a_in_call_check_func, phone_b, phone_b_idle_func, phone_b_in_call_check_func, call_sequence_func=None, wait_time_in_call=WAIT_TIME_IN_CALL): """Call process short sequence. 1. Ensure phone idle and in idle_func check return True. 2. Call from PhoneA to PhoneB, accept on PhoneB. 3. Check phone state, hangup on PhoneA. 4. Ensure phone idle and in idle_func check return True. 5. Call from PhoneA to PhoneB, accept on PhoneB. 6. Check phone state, hangup on PhoneB. Args: phone_a: PhoneA's android device object. phone_a_idle_func: function to check PhoneA's idle state. phone_a_in_call_check_func: function to check PhoneA's in-call state. phone_b: PhoneB's android device object. phone_b_idle_func: function to check PhoneB's idle state. phone_b_in_call_check_func: function to check PhoneB's in-call state. call_sequence_func: default parameter, not implemented. wait_time_in_call: time to wait in call. This is optional, default is WAIT_TIME_IN_CALL Returns: True: if call sequence succeed. False: for errors """ ads = [phone_a, phone_b] call_params = [ (ads[0], ads[1], ads[0], phone_a_in_call_check_func, phone_b_in_call_check_func), (ads[0], ads[1], ads[1], phone_a_in_call_check_func, phone_b_in_call_check_func), ] for param in call_params: # Make sure phones are idle. ensure_phones_idle(log, ads) if phone_a_idle_func and not phone_a_idle_func(log, phone_a): phone_a.log.error("Phone A Failed to Reselect") return False if phone_b_idle_func and not phone_b_idle_func(log, phone_b): phone_b.log.error("Phone B Failed to Reselect") return False # TODO: b/26337871 Need to use proper API to check phone registered. time.sleep(WAIT_TIME_BETWEEN_REG_AND_CALL) # Make call. log.info("--> Call test: %s to %s <--", phone_a.serial, phone_b.serial) slots = 2 for slot in range(slots): set_subid_for_outgoing_call( ads[0], get_subid_from_slot_index(log,ads[0],slot)) set_subid_for_outgoing_call( ads[1], get_subid_from_slot_index(log,ads[1],slot)) time.sleep(WAIT_TIME_BETWEEN_REG_AND_CALL) if not call_setup_teardown(log, *param,slot_id_callee = slot, wait_time_in_call=wait_time_in_call): log.error("Call Iteration Failed") return False if not call_setup_teardown(log, *param,slot_id_callee = 1-slot, wait_time_in_call=wait_time_in_call): log.error("Call Iteration Failed") return False return True def two_phone_call_long_seq(log, phone_a, phone_a_idle_func, phone_a_in_call_check_func, phone_b, phone_b_idle_func, phone_b_in_call_check_func, call_sequence_func=None, wait_time_in_call=WAIT_TIME_IN_CALL): """Call process long sequence. 1. Ensure phone idle and in idle_func check return True. 2. Call from PhoneA to PhoneB, accept on PhoneB. 3. Check phone state, hangup on PhoneA. 4. Ensure phone idle and in idle_func check return True. 5. Call from PhoneA to PhoneB, accept on PhoneB. 6. Check phone state, hangup on PhoneB. 7. Ensure phone idle and in idle_func check return True. 8. Call from PhoneB to PhoneA, accept on PhoneA. 9. Check phone state, hangup on PhoneA. 10. Ensure phone idle and in idle_func check return True. 11. Call from PhoneB to PhoneA, accept on PhoneA. 12. Check phone state, hangup on PhoneB. Args: phone_a: PhoneA's android device object. phone_a_idle_func: function to check PhoneA's idle state. phone_a_in_call_check_func: function to check PhoneA's in-call state. phone_b: PhoneB's android device object. phone_b_idle_func: function to check PhoneB's idle state. phone_b_in_call_check_func: function to check PhoneB's in-call state. call_sequence_func: default parameter, not implemented. wait_time_in_call: time to wait in call. This is optional, default is WAIT_TIME_IN_CALL Returns: TelResultWrapper which will evaluate as False if error. """ ads = [phone_a, phone_b] call_params = [ (ads[0], ads[1], ads[0], phone_a_in_call_check_func, phone_b_in_call_check_func), (ads[0], ads[1], ads[1], phone_a_in_call_check_func, phone_b_in_call_check_func), (ads[1], ads[0], ads[0], phone_b_in_call_check_func, phone_a_in_call_check_func), (ads[1], ads[0], ads[1], phone_b_in_call_check_func, phone_a_in_call_check_func), ] tel_result = TelResultWrapper(CallResult('SUCCESS')) for param in call_params: # Make sure phones are idle. ensure_phones_idle(log, ads) if phone_a_idle_func and not phone_a_idle_func(log, phone_a): phone_a.log.error("Phone A Failed to Reselect") return TelResultWrapper(CallResult('CALL_SETUP_FAILURE')) if phone_b_idle_func and not phone_b_idle_func(log, phone_b): phone_b.log.error("Phone B Failed to Reselect") return TelResultWrapper(CallResult('CALL_SETUP_FAILURE')) # TODO: b/26337871 Need to use proper API to check phone registered. time.sleep(WAIT_TIME_BETWEEN_REG_AND_CALL) # Make call. log.info("---> Call test: %s to %s <---", param[0].serial, param[1].serial) tel_result = call_setup_teardown( log, *param, wait_time_in_call=wait_time_in_call) if not tel_result: log.error("Call Iteration Failed") break return tel_result def two_phone_call_msim_for_slot(log, phone_a, phone_a_slot, phone_a_idle_func, phone_a_in_call_check_func, phone_b, phone_b_slot, phone_b_idle_func, phone_b_in_call_check_func, call_sequence_func=None, wait_time_in_call=WAIT_TIME_IN_CALL, retry=2): """Call process between 2 phones with specific slot. 1. Ensure phone idle and in idle_func check return True. 2. Call from PhoneA to PhoneB, accept on PhoneB. 3. Check phone state, hangup on PhoneA. 4. Ensure phone idle and in idle_func check return True. 5. Call from PhoneA to PhoneB, accept on PhoneB. 6. Check phone state, hangup on PhoneB. Args: phone_a: PhoneA's android device object. phone_a_slot: 0 or 1 (pSIM or eSIM) phone_a_idle_func: function to check PhoneA's idle state. phone_a_in_call_check_func: function to check PhoneA's in-call state. phone_b: PhoneB's android device object. phone_b_slot: 0 or 1 (pSIM or eSIM) phone_b_idle_func: function to check PhoneB's idle state. phone_b_in_call_check_func: function to check PhoneB's in-call state. call_sequence_func: default parameter, not implemented. wait_time_in_call: time to wait in call. This is optional, default is WAIT_TIME_IN_CALL retry: times of retry if call_setup_teardown failed. Returns: True: if call sequence succeed. False: for errors """ ads = [phone_a, phone_b] call_params = [ (ads[0], ads[1], ads[0], phone_a_in_call_check_func, phone_b_in_call_check_func), (ads[0], ads[1], ads[1], phone_a_in_call_check_func, phone_b_in_call_check_func), ] tel_result = TelResultWrapper(CallResult('SUCCESS')) for param in call_params: # Make sure phones are idle. ensure_phones_idle(log, ads) if phone_a_idle_func and not phone_a_idle_func(log, phone_a): phone_a.log.error("Phone A Failed to Reselect") return TelResultWrapper(CallResult('CALL_SETUP_FAILURE')) if phone_b_idle_func and not phone_b_idle_func(log, phone_b): phone_b.log.error("Phone B Failed to Reselect") return TelResultWrapper(CallResult('CALL_SETUP_FAILURE')) # TODO: b/26337871 Need to use proper API to check phone registered. time.sleep(WAIT_TIME_BETWEEN_REG_AND_CALL) # Make call. log.info("--> Call test: %s slot %s to %s slot %s <--", phone_a.serial, phone_a_slot, phone_b.serial, phone_b_slot) mo_default_voice_subid = get_subid_from_slot_index(log,ads[0], phone_a_slot) if mo_default_voice_subid == INVALID_SUB_ID: log.warning("Sub ID of MO (%s) slot %s is invalid.", phone_a.serial, phone_a_slot) return TelResultWrapper(CallResult('CALL_SETUP_FAILURE')) set_subid_for_outgoing_call( ads[0], mo_default_voice_subid) mt_default_voice_subid = get_subid_from_slot_index(log,ads[1], phone_b_slot) if mt_default_voice_subid == INVALID_SUB_ID: log.warning("Sub ID of MT (%s) slot %s is invalid.", phone_b.serial, phone_b_slot) return TelResultWrapper(CallResult('CALL_SETUP_FAILURE')) tel_result = call_setup_teardown( log, *param, slot_id_callee=phone_b_slot, wait_time_in_call=wait_time_in_call) while not tel_result: if retry <= 0: log.error("Call Iteration failed.") break else: log.info("RERUN call_setup_teardown.") tel_result = call_setup_teardown( log, *param, slot_id_callee=phone_b_slot, wait_time_in_call=wait_time_in_call) retry = retry - 1 return tel_result def is_phone_in_call(log, ad): """Return True if phone in call. Args: log: log object. ad: android device. """ try: return ad.droid.telecomIsInCall() except: return "mCallState=2" in ad.adb.shell( "dumpsys telephony.registry | grep mCallState") def is_phone_in_call_active(ad, call_id=None): """Return True if phone in active call. Args: log: log object. ad: android device. call_id: the call id """ if ad.droid.telecomIsInCall(): if not call_id: call_id = ad.droid.telecomCallGetCallIds()[0] call_state = ad.droid.telecomCallGetCallState(call_id) ad.log.info("%s state is %s", call_id, call_state) return call_state == "ACTIVE" else: ad.log.info("Not in telecomIsInCall") return False def is_phone_in_call_volte(log, ad): """Return if phone is in VoLTE call. Args: ad: Android device object. """ return is_phone_in_call_volte_for_subscription( log, ad, get_outgoing_voice_sub_id(ad)) def is_phone_in_call_volte_for_subscription(log, ad, sub_id): """Return if phone is in VoLTE call for subscription id. Args: ad: Android device object. sub_id: subscription id. """ if not ad.droid.telecomIsInCall(): ad.log.error("Not in call.") return False nw_type = get_network_rat_for_subscription(log, ad, sub_id, NETWORK_SERVICE_VOICE) if nw_type != RAT_LTE: ad.log.error("Voice rat on: %s. Expected: LTE", nw_type) return False return True def is_phone_in_call_csfb(log, ad): """Return if phone is in CSFB call. Args: ad: Android device object. """ return is_phone_in_call_csfb_for_subscription( log, ad, get_outgoing_voice_sub_id(ad)) def is_phone_in_call_csfb_for_subscription(log, ad, sub_id): """Return if phone is in CSFB call for subscription id. Args: ad: Android device object. sub_id: subscription id. """ if not ad.droid.telecomIsInCall(): ad.log.error("Not in call.") return False nw_type = get_network_rat_for_subscription(log, ad, sub_id, NETWORK_SERVICE_VOICE) if nw_type == RAT_LTE: ad.log.error("Voice rat on: %s. Expected: not LTE", nw_type) return False return True def is_phone_in_call_3g(log, ad): """Return if phone is in 3G call. Args: ad: Android device object. """ return is_phone_in_call_3g_for_subscription(log, ad, get_outgoing_voice_sub_id(ad)) def is_phone_in_call_3g_for_subscription(log, ad, sub_id): """Return if phone is in 3G call for subscription id. Args: ad: Android device object. sub_id: subscription id. """ if not ad.droid.telecomIsInCall(): ad.log.error("Not in call.") return False nw_gen = get_network_gen_for_subscription(log, ad, sub_id, NETWORK_SERVICE_VOICE) if nw_gen != GEN_3G: ad.log.error("Voice rat on: %s. Expected: 3g", nw_gen) return False return True def is_phone_in_call_2g(log, ad): """Return if phone is in 2G call. Args: ad: Android device object. """ return is_phone_in_call_2g_for_subscription(log, ad, get_outgoing_voice_sub_id(ad)) def is_phone_in_call_2g_for_subscription(log, ad, sub_id): """Return if phone is in 2G call for subscription id. Args: ad: Android device object. sub_id: subscription id. """ if not ad.droid.telecomIsInCall(): ad.log.error("Not in call.") return False nw_gen = get_network_gen_for_subscription(log, ad, sub_id, NETWORK_SERVICE_VOICE) if nw_gen != GEN_2G: ad.log.error("Voice rat on: %s. Expected: 2g", nw_gen) return False return True def is_phone_in_call_1x(log, ad): """Return if phone is in 1x call. Args: ad: Android device object. """ return is_phone_in_call_1x_for_subscription(log, ad, get_outgoing_voice_sub_id(ad)) def is_phone_in_call_1x_for_subscription(log, ad, sub_id): """Return if phone is in 1x call for subscription id. Args: ad: Android device object. sub_id: subscription id. """ if not ad.droid.telecomIsInCall(): ad.log.error("Not in call.") return False nw_type = get_network_rat_for_subscription(log, ad, sub_id, NETWORK_SERVICE_VOICE) if nw_type != RAT_1XRTT: ad.log.error("Voice rat on: %s. Expected: 1xrtt", nw_type) return False return True def is_phone_in_call_wcdma(log, ad): """Return if phone is in WCDMA call. Args: ad: Android device object. """ return is_phone_in_call_wcdma_for_subscription( log, ad, get_outgoing_voice_sub_id(ad)) def is_phone_in_call_wcdma_for_subscription(log, ad, sub_id): """Return if phone is in WCDMA call for subscription id. Args: ad: Android device object. sub_id: subscription id. """ # Currently checking 'umts'. # Changes may needed in the future. if not ad.droid.telecomIsInCall(): ad.log.error("Not in call.") return False nw_type = get_network_rat_for_subscription(log, ad, sub_id, NETWORK_SERVICE_VOICE) if nw_type != RAT_UMTS: ad.log.error("%s voice rat on: %s. Expected: umts", nw_type) return False return True def is_phone_in_call_iwlan(log, ad, call_id=None): """Return if phone is in WiFi call. Args: ad: Android device object. """ if not ad.droid.telecomIsInCall(): ad.log.error("Not in call.") return False if not ad.droid.telephonyIsImsRegistered(): ad.log.info("IMS is not registered.") return False if not ad.droid.telephonyIsWifiCallingAvailable(): ad.log.info("IsWifiCallingAvailable is False") return False if not call_id: call_ids = ad.droid.telecomCallGetCallIds() if call_ids: call_id = call_ids[-1] if not call_id: ad.log.error("Failed to get call id") return False else: call_prop = ad.droid.telecomCallGetProperties(call_id) if "WIFI" not in call_prop: ad.log.info("callProperties = %s, expecting WIFI", call_prop) return False nw_type = get_network_rat(log, ad, NETWORK_SERVICE_DATA) if nw_type != RAT_IWLAN: ad.log.warning("Data rat on: %s. Expected: iwlan", nw_type) return True def is_phone_in_call_not_iwlan(log, ad): """Return if phone is in WiFi call for subscription id. Args: ad: Android device object. sub_id: subscription id. """ if not ad.droid.telecomIsInCall(): ad.log.error("Not in call.") return False nw_type = get_network_rat(log, ad, NETWORK_SERVICE_DATA) if nw_type == RAT_IWLAN: ad.log.error("Data rat on: %s. Expected: not iwlan", nw_type) return False if is_wfc_enabled(log, ad): ad.log.error("WiFi Calling feature bit is True.") return False return True def swap_calls(log, ads, call_hold_id, call_active_id, num_swaps=1, check_call_status=True): """PhoneA in call with B and C. Swap active/holding call on PhoneA. Swap call and check status on PhoneA. (This step may have multiple times according to 'num_swaps'.) Check if all 3 phones are 'in-call'. Args: ads: list of ad object, at least three need to pass in. Swap operation will happen on ads[0]. ads[1] and ads[2] are call participants. call_hold_id: id for the holding call in ads[0]. call_hold_id should be 'STATE_HOLDING' when calling this function. call_active_id: id for the active call in ads[0]. call_active_id should be 'STATE_ACTIVE' when calling this function. num_swaps: how many swap/check operations will be done before return. check_call_status: This is optional. Default value is True. If this value is True, then call status (active/hold) will be be checked after each swap operation. Returns: If no error happened, return True, otherwise, return False. """ if check_call_status: # Check status before swap. if ads[0].droid.telecomCallGetCallState( call_active_id) != CALL_STATE_ACTIVE: ads[0].log.error( "Call_id:%s, state:%s, expected: STATE_ACTIVE", call_active_id, ads[0].droid.telecomCallGetCallState(call_active_id)) return False if ads[0].droid.telecomCallGetCallState( call_hold_id) != CALL_STATE_HOLDING: ads[0].log.error( "Call_id:%s, state:%s, expected: STATE_HOLDING", call_hold_id, ads[0].droid.telecomCallGetCallState(call_hold_id)) return False i = 1 while (i <= num_swaps): ads[0].log.info("swap_test %s: swap and check call status.", i) ads[0].droid.telecomCallHold(call_active_id) time.sleep(WAIT_TIME_IN_CALL) # Swap object reference call_active_id, call_hold_id = call_hold_id, call_active_id if check_call_status: # Check status if ads[0].droid.telecomCallGetCallState( call_active_id) != CALL_STATE_ACTIVE: ads[0].log.error( "Call_id:%s, state:%s, expected: STATE_ACTIVE", call_active_id, ads[0].droid.telecomCallGetCallState(call_active_id)) return False if ads[0].droid.telecomCallGetCallState( call_hold_id) != CALL_STATE_HOLDING: ads[0].log.error( "Call_id:%s, state:%s, expected: STATE_HOLDING", call_hold_id, ads[0].droid.telecomCallGetCallState(call_hold_id)) return False # TODO: b/26296375 add voice check. i += 1 #In the end, check all three phones are 'in-call'. if not verify_incall_state(log, [ads[0], ads[1], ads[2]], True): return False return True def get_audio_route(log, ad): """Gets the audio route for the active call Args: log: logger object ad: android_device object Returns: Audio route string ["BLUETOOTH", "EARPIECE", "SPEAKER", "WIRED_HEADSET" "WIRED_OR_EARPIECE"] """ audio_state = ad.droid.telecomCallGetAudioState() return audio_state["AudioRoute"] def set_audio_route(log, ad, route): """Sets the audio route for the active call Args: log: logger object ad: android_device object route: string ["BLUETOOTH", "EARPIECE", "SPEAKER", "WIRED_HEADSET" "WIRED_OR_EARPIECE"] Returns: If no error happened, return True, otherwise, return False. """ ad.droid.telecomCallSetAudioRoute(route) return True def is_property_in_call_properties(log, ad, call_id, expected_property): """Return if the call_id has the expected property Args: log: logger object ad: android_device object call_id: call id. expected_property: expected property. Returns: True if call_id has expected_property. False if not. """ properties = ad.droid.telecomCallGetProperties(call_id) return (expected_property in properties) def is_call_hd(log, ad, call_id): """Return if the call_id is HD call. Args: log: logger object ad: android_device object call_id: call id. Returns: True if call_id is HD call. False if not. """ return is_property_in_call_properties(log, ad, call_id, CALL_PROPERTY_HIGH_DEF_AUDIO) def get_cep_conference_call_id(ad): """Get CEP conference call id if there is an ongoing CEP conference call. Args: ad: android device object. Returns: call id for CEP conference call if there is an ongoing CEP conference call. None otherwise. """ for call in ad.droid.telecomCallGetCallIds(): if len(ad.droid.telecomCallGetCallChildren(call)) != 0: return call return None def is_phone_in_call_on_rat(log, ad, rat='volte', only_return_fn=None): if rat.lower() == 'volte' or rat.lower() == '5g_volte': if only_return_fn: return is_phone_in_call_volte else: return is_phone_in_call_volte(log, ad) elif rat.lower() == 'csfb' or rat.lower() == '5g_csfb': if only_return_fn: return is_phone_in_call_csfb else: return is_phone_in_call_csfb(log, ad) elif rat.lower() == '3g': if only_return_fn: return is_phone_in_call_3g else: return is_phone_in_call_3g(log, ad) elif rat.lower() == '2g': if only_return_fn: return is_phone_in_call_2g else: return is_phone_in_call_2g(log, ad) elif rat.lower() == 'wfc' or rat.lower() == '5g_wfc': if only_return_fn: return is_phone_in_call_iwlan else: return is_phone_in_call_iwlan(log, ad) else: return None def hold_unhold_test(log, ads): """ Test hold/unhold functionality. PhoneA is in call with PhoneB. The call on PhoneA is active. Get call list on PhoneA. Hold call_id on PhoneA. Check call_id state. Unhold call_id on PhoneA. Check call_id state. Args: log: log object ads: List of android objects. This list should contain 2 android objects. ads[0] is the ad to do hold/unhold operation. Returns: List of test result and call states. The first element of the list is always the test result. True if pass; False if fail. The rest of the list contains call states. """ call_list = ads[0].droid.telecomCallGetCallIds() log.info("Calls in PhoneA %s", call_list) if num_active_calls(ads[0].log, ads[0]) != 1: log.error("No voice call or too many voice calls in PhoneA!") call_state_list = [ads[0].droid.telecomCallGetCallState(call_id) for call_id in call_list] return [False] + call_state_list call_id = call_list[0] call_state = ads[0].droid.telecomCallGetCallState(call_id) if call_state != CALL_STATE_ACTIVE: log.error("Call_id:%s, state:%s, expected: STATE_ACTIVE", call_id, ads[0].droid.telecomCallGetCallState(call_id)) return [False, call_state] # TODO: b/26296375 add voice check. log.info("Hold call_id %s on PhoneA", call_id) ads[0].droid.telecomCallHold(call_id) time.sleep(WAIT_TIME_IN_CALL) call_state = ads[0].droid.telecomCallGetCallState(call_id) if call_state != CALL_STATE_HOLDING: ads[0].log.error("Call_id:%s, state:%s, expected: STATE_HOLDING", call_id, ads[0].droid.telecomCallGetCallState(call_id)) return [False, call_state] # TODO: b/26296375 add voice check. log.info("Unhold call_id %s on PhoneA", call_id) ads[0].droid.telecomCallUnhold(call_id) time.sleep(WAIT_TIME_IN_CALL) call_state = ads[0].droid.telecomCallGetCallState(call_id) if call_state != CALL_STATE_ACTIVE: log.error("Call_id:%s, state:%s, expected: STATE_ACTIVE", call_id, call_state) return [False, call_state] # TODO: b/26296375 add voice check. return [True, call_state] def phone_setup_call_hold_unhold_test(log, ads, call_direction=DIRECTION_MOBILE_ORIGINATED, caller_func=None, callee_func=None): """Test hold and unhold in voice call. 1. Clear call list. 2. Set up MO/MT call. 3. Test hold and unhold in call. 4. hangup call. Args: log: log object ads: list of android objects, this list should have two ad. call_direction: MO(DIRECTION_MOBILE_ORIGINATED) or MT(DIRECTION_MOBILE_TERMINATED) call. caller_func: function to verify caller is in correct state while in-call. callee_func: function to verify callee is in correct state while in-call. Returns: True if pass; False if fail. """ ads[0].droid.telecomCallClearCallList() if num_active_calls(log, ads[0]) != 0: ads[0].log.error("call list is not empty") return False log.info("begin hold/unhold test") ad_caller = ads[0] ad_callee = ads[1] if call_direction != DIRECTION_MOBILE_ORIGINATED: ad_caller = ads[1] ad_callee = ads[0] if not call_setup_teardown( log, ad_caller, ad_callee, ad_hangup=None, verify_caller_func=caller_func, verify_callee_func=callee_func): return False if not hold_unhold_test(ads[0].log, ads)[0]: log.error("hold/unhold test fail.") # hangup call in case voice call is still active. hangup_call(log, ads[0]) return False if not hangup_call(log, ads[0]): log.error("call hangup failed") return False return True def _test_call_long_duration(log, ads, dut_incall_check_func, total_duration): log.info("Long Duration Call Test. Total duration = %s", total_duration) return call_setup_teardown( log, ads[0], ads[1], ads[0], verify_caller_func=dut_incall_check_func, wait_time_in_call=total_duration) def _wait_for_ringing_event(log, ad, wait_time): """Wait for ringing event. Args: log: log object. ad: android device object. wait_time: max time to wait for ringing event. Returns: event_ringing if received ringing event. otherwise return None. """ event_ringing = None try: event_ringing = ad.ed.wait_for_event( EventCallStateChanged, is_event_match, timeout=wait_time, field=CallStateContainer.CALL_STATE, value=TELEPHONY_STATE_RINGING) ad.log.info("Receive ringing event") except Empty: ad.log.info("No Ringing Event") finally: return event_ringing def wait_for_telecom_ringing(log, ad, max_time=MAX_WAIT_TIME_TELECOM_RINGING): """Wait for android to be in telecom ringing state. Args: log: log object. ad: android device. max_time: maximal wait time. This is optional. Default Value is MAX_WAIT_TIME_TELECOM_RINGING. Returns: If phone become in telecom ringing state within max_time, return True. Return False if timeout. """ return _wait_for_droid_in_state( log, ad, max_time, lambda log, ad: ad.droid.telecomIsRinging()) def wait_for_ringing_call(log, ad, incoming_number=None): """Wait for an incoming call on default voice subscription and accepts the call. Args: log: log object. ad: android device object. incoming_number: Expected incoming number. Optional. Default is None Returns: True: if incoming call is received and answered successfully. False: for errors """ return wait_for_ringing_call_for_subscription( log, ad, get_incoming_voice_sub_id(ad), incoming_number) def wait_for_ringing_call_for_subscription( log, ad, sub_id, incoming_number=None, caller=None, event_tracking_started=False, timeout=MAX_WAIT_TIME_CALLEE_RINGING, interval=WAIT_TIME_BETWEEN_STATE_CHECK): """Wait for an incoming call on specified subscription. Args: log: log object. ad: android device object. sub_id: subscription ID incoming_number: Expected incoming number. Default is None event_tracking_started: True if event tracking already state outside timeout: time to wait for ring interval: checking interval Returns: True: if incoming call is received and answered successfully. False: for errors """ if not event_tracking_started: ad.ed.clear_events(EventCallStateChanged) ad.droid.telephonyStartTrackingCallStateForSubscription(sub_id) ring_event_received = False end_time = time.time() + timeout try: while time.time() < end_time: if not ring_event_received: event_ringing = _wait_for_ringing_event(log, ad, interval) if event_ringing: if incoming_number and not check_phone_number_match( event_ringing['data'] [CallStateContainer.INCOMING_NUMBER], incoming_number): ad.log.error( "Incoming Number not match. Expected number:%s, actual number:%s", incoming_number, event_ringing['data'][ CallStateContainer.INCOMING_NUMBER]) return False ring_event_received = True telephony_state = ad.droid.telephonyGetCallStateForSubscription( sub_id) telecom_state = ad.droid.telecomGetCallState() if telephony_state == TELEPHONY_STATE_RINGING and ( telecom_state == TELEPHONY_STATE_RINGING): ad.log.info("callee is in telephony and telecom RINGING state") if caller: if caller.droid.telecomIsInCall(): caller.log.info("Caller telecom is in call state") return True else: caller.log.info("Caller telecom is NOT in call state") else: return True else: ad.log.info( "telephony in %s, telecom in %s, expecting RINGING state", telephony_state, telecom_state) time.sleep(interval) finally: if not event_tracking_started: ad.droid.telephonyStopTrackingCallStateChangeForSubscription( sub_id) def wait_for_call_offhook_for_subscription( log, ad, sub_id, event_tracking_started=False, timeout=MAX_WAIT_TIME_ACCEPT_CALL_TO_OFFHOOK_EVENT, interval=WAIT_TIME_BETWEEN_STATE_CHECK): """Wait for an incoming call on specified subscription. Args: log: log object. ad: android device object. sub_id: subscription ID timeout: time to wait for ring interval: checking interval Returns: True: if incoming call is received and answered successfully. False: for errors """ if not event_tracking_started: ad.ed.clear_events(EventCallStateChanged) ad.droid.telephonyStartTrackingCallStateForSubscription(sub_id) offhook_event_received = False end_time = time.time() + timeout try: while time.time() < end_time: if not offhook_event_received: if wait_for_call_offhook_event(log, ad, sub_id, True, interval): offhook_event_received = True telephony_state = ad.droid.telephonyGetCallStateForSubscription( sub_id) telecom_state = ad.droid.telecomGetCallState() if telephony_state == TELEPHONY_STATE_OFFHOOK and ( telecom_state == TELEPHONY_STATE_OFFHOOK): ad.log.info("telephony and telecom are in OFFHOOK state") return True else: ad.log.info( "telephony in %s, telecom in %s, expecting OFFHOOK state", telephony_state, telecom_state) if offhook_event_received: time.sleep(interval) finally: if not event_tracking_started: ad.droid.telephonyStopTrackingCallStateChangeForSubscription( sub_id) def wait_for_call_offhook_event( log, ad, sub_id, event_tracking_started=False, timeout=MAX_WAIT_TIME_ACCEPT_CALL_TO_OFFHOOK_EVENT): """Wait for an incoming call on specified subscription. Args: log: log object. ad: android device object. event_tracking_started: True if event tracking already state outside timeout: time to wait for event Returns: True: if call offhook event is received. False: if call offhook event is not received. """ if not event_tracking_started: ad.ed.clear_events(EventCallStateChanged) ad.droid.telephonyStartTrackingCallStateForSubscription(sub_id) try: ad.ed.wait_for_event( EventCallStateChanged, is_event_match, timeout=timeout, field=CallStateContainer.CALL_STATE, value=TELEPHONY_STATE_OFFHOOK) ad.log.info("Got event %s", TELEPHONY_STATE_OFFHOOK) except Empty: ad.log.info("No event for call state change to OFFHOOK") return False finally: if not event_tracking_started: ad.droid.telephonyStopTrackingCallStateChangeForSubscription( sub_id) return True def wait_and_answer_call_for_subscription( log, ad, sub_id, incoming_number=None, incall_ui_display=INCALL_UI_DISPLAY_FOREGROUND, timeout=MAX_WAIT_TIME_CALLEE_RINGING, caller=None, video_state=None): """Wait for an incoming call on specified subscription and accepts the call. Args: log: log object. ad: android device object. sub_id: subscription ID incoming_number: Expected incoming number. Optional. Default is None incall_ui_display: after answer the call, bring in-call UI to foreground or background. Optional, default value is INCALL_UI_DISPLAY_FOREGROUND. if = INCALL_UI_DISPLAY_FOREGROUND, bring in-call UI to foreground. if = INCALL_UI_DISPLAY_BACKGROUND, bring in-call UI to background. else, do nothing. Returns: True: if incoming call is received and answered successfully. False: for errors """ ad.ed.clear_events(EventCallStateChanged) ad.droid.telephonyStartTrackingCallStateForSubscription(sub_id) try: if not wait_for_ringing_call_for_subscription( log, ad, sub_id, incoming_number=incoming_number, caller=caller, event_tracking_started=True, timeout=timeout): ad.log.info("Incoming call ringing check failed.") return False ad.log.info("Accept the ring call") ad.droid.telecomAcceptRingingCall(video_state) if wait_for_call_offhook_for_subscription( log, ad, sub_id, event_tracking_started=True): return True else: ad.log.error("Could not answer the call.") return False except Exception as e: log.error(e) return False finally: ad.droid.telephonyStopTrackingCallStateChangeForSubscription(sub_id) if incall_ui_display == INCALL_UI_DISPLAY_FOREGROUND: ad.droid.telecomShowInCallScreen() elif incall_ui_display == INCALL_UI_DISPLAY_BACKGROUND: ad.droid.showHomeScreen() def wait_and_reject_call(log, ad, incoming_number=None, delay_reject=WAIT_TIME_REJECT_CALL, reject=True): """Wait for an incoming call on default voice subscription and reject the call. Args: log: log object. ad: android device object. incoming_number: Expected incoming number. Optional. Default is None delay_reject: time to wait before rejecting the call Optional. Default is WAIT_TIME_REJECT_CALL Returns: True: if incoming call is received and reject successfully. False: for errors """ return wait_and_reject_call_for_subscription(log, ad, get_incoming_voice_sub_id(ad), incoming_number, delay_reject, reject) def wait_and_reject_call_for_subscription(log, ad, sub_id, incoming_number=None, delay_reject=WAIT_TIME_REJECT_CALL, reject=True): """Wait for an incoming call on specific subscription and reject the call. Args: log: log object. ad: android device object. sub_id: subscription ID incoming_number: Expected incoming number. Optional. Default is None delay_reject: time to wait before rejecting the call Optional. Default is WAIT_TIME_REJECT_CALL Returns: True: if incoming call is received and reject successfully. False: for errors """ if not wait_for_ringing_call_for_subscription(log, ad, sub_id, incoming_number): ad.log.error( "Could not reject a call: incoming call in ringing check failed.") return False ad.ed.clear_events(EventCallStateChanged) ad.droid.telephonyStartTrackingCallStateForSubscription(sub_id) if reject is True: # Delay between ringing and reject. time.sleep(delay_reject) is_find = False # Loop the call list and find the matched one to disconnect. for call in ad.droid.telecomCallGetCallIds(): if check_phone_number_match( get_number_from_tel_uri(get_call_uri(ad, call)), incoming_number): ad.droid.telecomCallDisconnect(call) ad.log.info("Callee reject the call") is_find = True if is_find is False: ad.log.error("Callee did not find matching call to reject.") return False else: # don't reject on callee. Just ignore the incoming call. ad.log.info("Callee received incoming call. Ignore it.") try: ad.ed.wait_for_event( EventCallStateChanged, is_event_match_for_list, timeout=MAX_WAIT_TIME_CALL_IDLE_EVENT, field=CallStateContainer.CALL_STATE, value_list=[TELEPHONY_STATE_IDLE, TELEPHONY_STATE_OFFHOOK]) except Empty: ad.log.error("No onCallStateChangedIdle event received.") return False finally: ad.droid.telephonyStopTrackingCallStateChangeForSubscription(sub_id) return True def wait_and_answer_call(log, ad, incoming_number=None, incall_ui_display=INCALL_UI_DISPLAY_FOREGROUND, caller=None, video_state=None): """Wait for an incoming call on default voice subscription and accepts the call. Args: ad: android device object. incoming_number: Expected incoming number. Optional. Default is None incall_ui_display: after answer the call, bring in-call UI to foreground or background. Optional, default value is INCALL_UI_DISPLAY_FOREGROUND. if = INCALL_UI_DISPLAY_FOREGROUND, bring in-call UI to foreground. if = INCALL_UI_DISPLAY_BACKGROUND, bring in-call UI to background. else, do nothing. Returns: True: if incoming call is received and answered successfully. False: for errors """ return wait_and_answer_call_for_subscription( log, ad, get_incoming_voice_sub_id(ad), incoming_number, incall_ui_display=incall_ui_display, caller=caller, video_state=video_state) def wait_for_in_call_active(ad, timeout=MAX_WAIT_TIME_ACCEPT_CALL_TO_OFFHOOK_EVENT, interval=WAIT_TIME_BETWEEN_STATE_CHECK, call_id=None): """Wait for call reach active state. Args: log: log object. ad: android device. call_id: the call id """ if not call_id: call_id = ad.droid.telecomCallGetCallIds()[0] args = [ad, call_id] if not wait_for_state(is_phone_in_call_active, True, timeout, interval, *args): ad.log.error("Call did not reach ACTIVE state") return False else: return True def wait_for_droid_in_call(log, ad, max_time): """Wait for android to be in call state. Args: log: log object. ad: android device. max_time: maximal wait time. Returns: If phone become in call state within max_time, return True. Return False if timeout. """ return _wait_for_droid_in_state(log, ad, max_time, is_phone_in_call) def wait_for_call_id_clearing(ad, previous_ids, timeout=MAX_WAIT_TIME_CALL_DROP): while timeout > 0: new_call_ids = ad.droid.telecomCallGetCallIds() if len(new_call_ids) <= len(previous_ids): return True time.sleep(5) timeout = timeout - 5 ad.log.error("Call id clearing failed. Before: %s; After: %s", previous_ids, new_call_ids) return False def wait_for_call_end( log, ad_caller, ad_callee, ad_hangup, verify_caller_func, verify_callee_func, call_begin_time, check_interval=5, tel_result_wrapper=TelResultWrapper(CallResult('SUCCESS')), wait_time_in_call=WAIT_TIME_IN_CALL): elapsed_time = 0 while (elapsed_time < wait_time_in_call): check_interval = min(check_interval, wait_time_in_call - elapsed_time) time.sleep(check_interval) elapsed_time += check_interval time_message = "at <%s>/<%s> second." % (elapsed_time, wait_time_in_call) for ad, call_func in [(ad_caller, verify_caller_func), (ad_callee, verify_callee_func)]: if not call_func(log, ad): ad.log.error( "NOT in correct %s state at %s, voice in RAT %s", call_func.__name__, time_message, ad.droid.telephonyGetCurrentVoiceNetworkType()) tel_result_wrapper.result_value = CallResult( 'CALL_DROP_OR_WRONG_STATE_AFTER_CONNECTED') else: ad.log.info("In correct %s state at %s", call_func.__name__, time_message) if not ad.droid.telecomCallGetAudioState(): ad.log.error("Audio is not in call state at %s", time_message) tel_result_wrapper.result_value = CallResult( 'AUDIO_STATE_NOT_INCALL_AFTER_CONNECTED') if not tel_result_wrapper: break if not tel_result_wrapper: for ad in (ad_caller, ad_callee): last_call_drop_reason(ad, call_begin_time) try: if ad.droid.telecomIsInCall(): ad.log.info("In call. End now.") ad.droid.telecomEndCall() except Exception as e: log.error(str(e)) else: if ad_hangup: if not hangup_call(log, ad_hangup): ad_hangup.log.info("Failed to hang up the call") tel_result_wrapper.result_value = CallResult('CALL_HANGUP_FAIL') if ad_hangup or not tel_result_wrapper: for ad in (ad_caller, ad_callee): if not wait_for_call_id_clearing(ad, getattr(ad, "caller_ids", [])): tel_result_wrapper.result_value = CallResult( 'CALL_ID_CLEANUP_FAIL') return tel_result_wrapper def check_call(log, dut, dut_client): result = True if not call_setup_teardown(log, dut_client, dut, dut): if not call_setup_teardown(log, dut_client, dut, dut): dut.log.error("MT call failed") result = False if not call_setup_teardown(log, dut, dut_client, dut): dut.log.error("MO call failed") result = False return result def check_call_in_wfc(log, dut, dut_client): result = True if not call_setup_teardown(log, dut_client, dut, dut, None, is_phone_in_call_iwlan): if not call_setup_teardown(log, dut_client, dut, dut, None, is_phone_in_call_iwlan): dut.log.error("MT WFC call failed") result = False if not call_setup_teardown(log, dut, dut_client, dut, is_phone_in_call_iwlan): dut.log.error("MO WFC call failed") result = False return result def check_call_in_volte(log, dut, dut_client): result = True if not call_setup_teardown(log, dut_client, dut, dut, None, is_phone_in_call_volte): if not call_setup_teardown(log, dut_client, dut, dut, None, is_phone_in_call_volte): dut.log.error("MT VoLTE call failed") result = False if not call_setup_teardown(log, dut, dut_client, dut, is_phone_in_call_volte): dut.log.error("MO VoLTE call failed") result = False return result def change_ims_setting(log, ad, dut_client, wifi_network_ssid, wifi_network_pass, subid, dut_capabilities, airplane_mode, wifi_enabled, volte_enabled, wfc_enabled, nw_gen=RAT_LTE, wfc_mode=None): result = True ad.log.info( "Setting APM %s, WIFI %s, VoLTE %s, WFC %s, WFC mode %s", airplane_mode, wifi_enabled, volte_enabled, wfc_enabled, wfc_mode) toggle_airplane_mode_by_adb(log, ad, airplane_mode) if wifi_enabled: if not ensure_wifi_connected(log, ad, wifi_network_ssid, wifi_network_pass, apm=airplane_mode): ad.log.error("Fail to connected to WiFi") result = False else: if not wifi_toggle_state(log, ad, False): ad.log.error("Failed to turn off WiFi.") result = False toggle_volte(log, ad, volte_enabled) toggle_wfc(log, ad, wfc_enabled) if wfc_mode: set_wfc_mode(log, ad, wfc_mode) wfc_mode = ad.droid.imsGetWfcMode() if wifi_enabled or not airplane_mode: if not ensure_phone_subscription(log, ad): ad.log.error("Failed to find valid subscription") result = False if airplane_mode: if (CAPABILITY_WFC in dut_capabilities) and (wifi_enabled and wfc_enabled): if not wait_for_wfc_enabled(log, ad): result = False elif not check_call_in_wfc(log, ad, dut_client): result = False else: if not wait_for_state( ad.droid.telephonyGetCurrentVoiceNetworkType, RAT_UNKNOWN): ad.log.error( "Voice RAT is %s not UNKNOWN", ad.droid.telephonyGetCurrentVoiceNetworkType()) result = False else: ad.log.info("Voice RAT is in UNKKNOWN") else: if (wifi_enabled and wfc_enabled) and ( wfc_mode == WFC_MODE_WIFI_PREFERRED) and ( CAPABILITY_WFC in dut_capabilities): if not wait_for_wfc_enabled(log, ad): result = False if not wait_for_state( ad.droid.telephonyGetCurrentVoiceNetworkType, RAT_UNKNOWN): ad.log.error( "Voice RAT is %s, not UNKNOWN", ad.droid.telephonyGetCurrentVoiceNetworkType()) if not check_call_in_wfc(log, ad, dut_client): result = False else: if not wait_for_wfc_disabled(log, ad): ad.log.error("WFC is not disabled") result = False if volte_enabled and CAPABILITY_VOLTE in dut_capabilities: if not wait_for_volte_enabled(log, ad): result = False if not check_call_in_volte(log, ad, dut_client): result = False else: if not wait_for_not_network_rat( log, ad, nw_gen, voice_or_data=NETWORK_SERVICE_VOICE): ad.log.error( "Voice RAT is %s", ad.droid.telephonyGetCurrentVoiceNetworkType( )) result = False if not wait_for_voice_attach(log, ad): result = False if not check_call(log, ad, dut_client): result = False user_config_profile = get_user_config_profile(ad) ad.log.info("user_config_profile: %s ", sorted(user_config_profile.items())) return result def verify_default_ims_setting(log, ad, dut_client, carrier_configs, default_wfc_enabled, default_volte, wfc_mode=None): result = True airplane_mode = ad.droid.connectivityCheckAirplaneMode() default_wfc_mode = carrier_configs.get( CarrierConfigs.DEFAULT_WFC_IMS_MODE_INT, wfc_mode) if default_wfc_enabled: wait_for_wfc_enabled(log, ad) else: wait_for_wfc_disabled(log, ad) if airplane_mode: wait_for_network_rat( log, ad, RAT_UNKNOWN, voice_or_data=NETWORK_SERVICE_VOICE) else: if default_volte: wait_for_volte_enabled(log, ad) else: wait_for_not_network_rat( log, ad, RAT_UNKNOWN, voice_or_data=NETWORK_SERVICE_VOICE) if not ensure_phone_subscription(log, ad): ad.log.error("Failed to find valid subscription") result = False user_config_profile = get_user_config_profile(ad) ad.log.info("user_config_profile = %s ", sorted(user_config_profile.items())) if user_config_profile["VoLTE Enabled"] != default_volte: ad.log.error("VoLTE mode is not %s", default_volte) result = False else: ad.log.info("VoLTE mode is %s as expected", default_volte) if user_config_profile["WFC Enabled"] != default_wfc_enabled: ad.log.error("WFC enabled is not %s", default_wfc_enabled) if user_config_profile["WFC Enabled"]: if user_config_profile["WFC Mode"] != default_wfc_mode: ad.log.error( "WFC mode is not %s after IMS factory reset", default_wfc_mode) result = False else: ad.log.info("WFC mode is %s as expected", default_wfc_mode) if default_wfc_enabled and \ default_wfc_mode == WFC_MODE_WIFI_PREFERRED: if not check_call_in_wfc(log, ad, dut_client): result = False elif not airplane_mode: if default_volte: if not check_call_in_volte(log, ad, dut_client): result = False else: if not check_call(log, ad, dut_client): result = False if result == False: user_config_profile = get_user_config_profile(ad) ad.log.info("user_config_profile = %s ", sorted(user_config_profile.items())) return result def truncate_phone_number( log, caller_number, callee_number, dialing_number_length, skip_inter_area_call=False): """This function truncates the phone number of the caller/callee to test 7/10/11/12 digit dialing for North American numbering plan, and distinguish if this is an inter-area call by comparing the area code. Args: log: logger object caller_number: phone number of the caller callee_number: phone number of the callee dialing_number_length: the length of phone number (usually 7/10/11/12) skip_inter_area_call: True to raise a TestSkip exception to skip dialing the inter-area call. Otherwise False. Returns: The truncated phone number of the callee """ if not dialing_number_length: return callee_number trunc_position = 0 - int(dialing_number_length) try: caller_area_code = caller_number[:trunc_position] callee_area_code = callee_number[:trunc_position] callee_dial_number = callee_number[trunc_position:] if caller_area_code != callee_area_code: skip_inter_area_call = True except: skip_inter_area_call = True if skip_inter_area_call: msg = "Cannot make call from %s to %s by %s digits since inter-area \ call is not allowed" % ( caller_number, callee_number, dialing_number_length) log.info(msg) raise signals.TestSkip(msg) else: callee_number = callee_dial_number return callee_number