| #!/usr/bin/env python3 |
| # |
| # Copyright 2016 - 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. |
| |
| # Defines utilities that can be used for making calls indenpendent of |
| # subscription IDs. This can be useful when making calls over mediums not SIM |
| # based. |
| |
| # Make a phone call to the specified URI. It is assumed that we are making the |
| # call to the user selected default account. |
| # |
| # We usually want to make sure that the call has ended up in a good state. |
| # |
| # NOTE: This util is applicable to only non-conference type calls. It is best |
| # suited to test cases where only one call is in action at any point of time. |
| |
| import queue |
| import time |
| |
| from acts import logger |
| from acts_contrib.test_utils.tel import tel_defines |
| |
| def dial_number(log, ad, uri): |
| """Dial a number |
| |
| Args: |
| log: log object |
| ad: android device object |
| uri: Tel number to dial |
| |
| Returns: |
| True if success, False if fail. |
| """ |
| log.info("Dialing up droid {} call uri {}".format( |
| ad.serial, uri)) |
| |
| # First check that we are not in call. |
| if ad.droid.telecomIsInCall(): |
| log.info("We're still in call {}".format(ad.serial)) |
| return False |
| |
| # Start tracking updates. |
| ad.droid.telecomStartListeningForCallAdded() |
| #If a phone number is passed in |
| if "tel:" not in uri: |
| uri = "tel:" + uri |
| ad.droid.telecomCallTelUri(uri) |
| |
| event = None |
| try: |
| event = ad.ed.pop_event( |
| tel_defines.EventTelecomCallAdded, |
| tel_defines.MAX_WAIT_TIME_CONNECTION_STATE_UPDATE) |
| except queue.Empty: |
| log.info( |
| "Did not get {} event!".format(tel_defines.EventTelecomCallAdded)) |
| # Return failure. |
| return False |
| finally: |
| ad.droid.telecomStopListeningForCallAdded() |
| |
| call_id = event['data']['CallId'] |
| log.info("Call ID: {} dev {}".format(call_id, ad.serial)) |
| |
| if not call_id: |
| log.info("CallId is empty!") |
| return False |
| if not wait_for_dialing(log, ad): |
| return False |
| |
| return call_id |
| |
| def wait_for_call_state(log, ad, call_id, state): |
| """Wait for the given call id to transition to the given call state. |
| |
| Args: |
| log: log object |
| ad: android device object |
| call_id: ID of the call that we're waiting for the call state to |
| transition into. |
| state: desired final state. |
| |
| Returns: |
| True if success, False if fail. |
| """ |
| # Lets track the call now. |
| # NOTE: Disable this everywhere we return. |
| ad.droid.telecomCallStartListeningForEvent( |
| call_id, tel_defines.EVENT_CALL_STATE_CHANGED) |
| |
| # We may have missed the update so do a quick check. |
| if ad.droid.telecomCallGetCallState(call_id) == state: |
| log.info("Call ID {} already in {} dev {}!".format( |
| call_id, state, ad.serial)) |
| ad.droid.telecomCallStopListeningForEvent(call_id, |
| tel_defines.EventTelecomCallStateChanged) |
| return True |
| |
| # If not then we need to poll for the event. |
| # We return if we have found a match or we timeout for further updates of |
| # the call |
| end_time = time.time() + 10 |
| while True and time.time() < end_time: |
| try: |
| event = ad.ed.pop_event( |
| tel_defines.EventTelecomCallStateChanged, |
| tel_defines.MAX_WAIT_TIME_CONNECTION_STATE_UPDATE) |
| call_state = event['data']['Event'] |
| if call_state == state: |
| ad.droid.telecomCallStopListeningForEvent(call_id, |
| tel_defines.EventTelecomCallStateChanged) |
| return True |
| else: |
| log.info("Droid {} in call {} state {}".format( |
| ad.serial, call_id, call_state)) |
| continue |
| except queue.Empty: |
| log.info("Did not get into state {} dev {}".format( |
| state, ad.serial)) |
| ad.droid.telecomCallStopListeningForEvent(call_id, |
| tel_defines.EventTelecomCallStateChanged) |
| return False |
| return False |
| |
| def hangup_call(log, ad, call_id): |
| """Hangup a number |
| |
| Args: |
| log: log object |
| ad: android device object |
| call_id: Call to hangup. |
| |
| Returns: |
| True if success, False if fail. |
| """ |
| log.info("Hanging up droid {} call {}".format( |
| ad.serial, call_id)) |
| # First check that we are in call, otherwise fail. |
| if not ad.droid.telecomIsInCall(): |
| log.info("We are not in-call {}".format(ad.serial)) |
| return False |
| |
| # Make sure we are registered with the events. |
| ad.droid.telecomStartListeningForCallRemoved() |
| |
| # Disconnect call. |
| ad.droid.telecomCallDisconnect(call_id) |
| |
| # Wait for removed event. |
| event = None |
| try: |
| event = ad.ed.pop_event( |
| tel_defines.EventTelecomCallRemoved, |
| tel_defines.MAX_WAIT_TIME_CONNECTION_STATE_UPDATE) |
| except queue.Empty: |
| log.info("Did not get TelecomCallRemoved event") |
| return False |
| finally: |
| ad.droid.telecomStopListeningForCallRemoved() |
| |
| log.info("Removed call {}".format(event)) |
| if event['data']['CallId'] != call_id: |
| return False |
| |
| return True |
| |
| def hangup_conf(log, ad, conf_id, timeout=10): |
| """Hangup a conference call |
| |
| Args: |
| log: log object |
| ad: android device object |
| conf_id: Conf call to hangup. |
| |
| Returns: |
| True if success, False if fail. |
| """ |
| log.info("hangup_conf: Hanging up droid {} call {}".format( |
| ad.serial, conf_id)) |
| |
| # First check that we are in call, otherwise fail. |
| if not ad.droid.telecomIsInCall(): |
| log.info("We are not in-call {}".format(ad.serial)) |
| return False |
| |
| # Disconnect call. |
| ad.droid.telecomCallDisconnect(conf_id) |
| |
| start_time = time.time() |
| while time.time() < start_time + timeout: |
| call_ids = get_calls_in_states(log, ad, [tel_defines.CALL_STATE_ACTIVE]) |
| log.debug("Active calls {}".format(call_ids)) |
| if not call_ids: |
| return True |
| time.sleep(1) |
| log.error("Failed to hang up all conference participants") |
| return False |
| |
| def accept_call(log, ad, call_id): |
| """Accept a number |
| |
| Args: |
| log: log object |
| ad: android device object |
| call_id: Call to accept. |
| |
| Returns: |
| True if success, False if fail. |
| """ |
| log.info("Accepting call at droid {} call {}".format( |
| ad.serial, call_id)) |
| # First check we are in call, otherwise fail. |
| if not ad.droid.telecomIsInCall(): |
| log.info("We are not in-call {}".format(ad.serial)) |
| return False |
| |
| # Accept the call and wait for the call to be accepted on AG. |
| ad.droid.telecomCallAnswer(call_id, tel_defines.VT_STATE_AUDIO_ONLY) |
| if not wait_for_call_state(log, ad, call_id, tel_defines.CALL_STATE_ACTIVE): |
| log.error("Call {} on droid {} not active".format( |
| call_id, ad.serial)) |
| return False |
| |
| return True |
| |
| def wait_for_not_in_call(log, ad): |
| """Wait for the droid to be OUT OF CALLING state. |
| |
| Args: |
| log: log object |
| ad: android device object |
| |
| Returns: |
| True if success, False if fail. |
| """ |
| ad.droid.telecomStartListeningForCallRemoved() |
| |
| calls = ad.droid.telecomCallGetCallIds() |
| if len(calls) > 1: |
| log.info("More than one call {} {}".format(calls, ad.serial)) |
| return False |
| |
| if len(calls) > 0: |
| log.info("Got calls {} for {}".format( |
| calls, ad.serial)) |
| try: |
| event = ad.ed.pop_event( |
| tel_defines.EventTelecomCallRemoved, |
| tel_defines.MAX_WAIT_TIME_CONNECTION_STATE_UPDATE) |
| except queue.Empty: |
| log.info("wait_for_not_in_call Did not get {} droid {}".format( |
| tel_defines.EventTelecomCallRemoved, |
| ad.serial)) |
| return False |
| finally: |
| ad.droid.telecomStopListeningForCallRemoved() |
| |
| # Either we removed the only call or we never had a call previously, either |
| # ways simply check if we are in in call now. |
| return (not ad.droid.telecomIsInCall()) |
| |
| def wait_for_dialing(log, ad): |
| """Wait for the droid to be in dialing state. |
| |
| Args: |
| log: log object |
| ad: android device object |
| |
| Returns: |
| True if success, False if fail. |
| """ |
| # Start listening for events before anything else happens. |
| ad.droid.telecomStartListeningForCallAdded() |
| |
| # First check if we re in call, then simply return. |
| if ad.droid.telecomIsInCall(): |
| ad.droid.telecomStopListeningForCallAdded() |
| return True |
| |
| call_id = None |
| # Now check if we already have calls matching the state. |
| calls_in_state = get_calls_in_states(log, ad, |
| [tel_defines.CALL_STATE_CONNECTING, |
| tel_defines.CALL_STATE_DIALING]) |
| |
| # If not then we need to poll for the calls themselves. |
| if len(calls_in_state) == 0: |
| event = None |
| try: |
| event = ad.ed.pop_event( |
| tel_defines.EventTelecomCallAdded, |
| tel_defines.MAX_WAIT_TIME_CONNECTION_STATE_UPDATE) |
| except queue.Empty: |
| log.info("Did not get {}".format( |
| tel_defines.EventTelecomCallAdded)) |
| return False |
| finally: |
| ad.droid.telecomStopListeningForCallAdded() |
| call_id = event['data']['CallId'] |
| else: |
| call_id = calls_in_state[0] |
| |
| # We may still not be in-call if the call setup is going on. |
| # We wait for the call state to move to dialing. |
| log.info("call id {} droid {}".format(call_id, ad.serial)) |
| if not wait_for_call_state( |
| log, ad, call_id, tel_defines.CALL_STATE_DIALING): |
| return False |
| |
| # Finally check the call state. |
| return ad.droid.telecomIsInCall() |
| |
| def wait_for_ringing(log, ad): |
| """Wait for the droid to be in ringing state. |
| |
| Args: |
| log: log object |
| ad: android device object |
| |
| Returns: |
| True if success, False if fail. |
| """ |
| log.info("waiting for ringing {}".format(ad.serial)) |
| # Start listening for events before anything else happens. |
| ad.droid.telecomStartListeningForCallAdded() |
| |
| # First check if we re in call, then simply return. |
| if ad.droid.telecomIsInCall(): |
| log.info("Device already in call {}".format(ad.serial)) |
| ad.droid.telecomStopListeningForCallAdded() |
| return True |
| |
| call_id = None |
| # Now check if we already have calls matching the state. |
| calls_in_state = ad.droid.telecomCallGetCallIds() |
| |
| for c_id in calls_in_state: |
| if ad.droid.telecomCallGetCallState(c_id) == tel_defines.CALL_STATE_RINGING: |
| return True |
| |
| event = None |
| call_id = None |
| try: |
| event = ad.ed.pop_event( |
| tel_defines.EventTelecomCallAdded, |
| tel_defines.MAX_WAIT_TIME_CALLEE_RINGING) |
| except queue.Empty: |
| log.info("Did not get {} droid {}".format( |
| tel_defines.EventTelecomCallAdded, |
| ad.serial)) |
| return False |
| finally: |
| ad.droid.telecomStopListeningForCallAdded() |
| call_id = event['data']['CallId'] |
| log.info("wait_for_ringing call found {} dev {}".format( |
| call_id, ad.serial)) |
| |
| # If the call already existed then we would have returned above otherwise |
| # we will verify that the newly added call is indeed ringing. |
| if not wait_for_call_state( |
| log, ad, call_id, tel_defines.CALL_STATE_RINGING): |
| log.info("No ringing call id {} droid {}".format( |
| call_id, ad.serial)) |
| return False |
| return True |
| |
| def wait_for_active(log, ad): |
| """Wait for the droid to be in active call state. |
| |
| Args: |
| log: log object |
| ad: android device object |
| |
| Returns: |
| True if success, False if fail. |
| """ |
| log.info("waiting for active {}".format(ad.serial)) |
| # Start listening for events before anything else happens. |
| ad.droid.telecomStartListeningForCallAdded() |
| |
| call_id = None |
| # Now check if we already have calls matching the state. |
| calls_in_state = ad.droid.telecomCallGetCallIds() |
| |
| if len(calls_in_state) == 0: |
| event = None |
| try: |
| event = ad.ed.pop_event( |
| tel_defines.EventTelecomCallAdded, |
| tel_defines.MAX_WAIT_TIME_CALLEE_RINGING) |
| except queue.Empty: |
| log.info("Did not get {} droid {}".format( |
| tel_defines.EventTelecomCallAdded, |
| ad.serial)) |
| return False |
| finally: |
| ad.droid.telecomStopListeningForCallAdded() |
| call_id = event['data']['CallId'] |
| log.info("wait_for_ringing call found {} dev {}".format( |
| call_id, ad.serial)) |
| else: |
| call_id = calls_in_state[0] |
| |
| # We have found a new call to be added now wait it to transition into |
| # active state. |
| if not wait_for_call_state( |
| log, ad, call_id, tel_defines.CALL_STATE_ACTIVE): |
| log.info("No active call id {} droid {}".format( |
| call_id, ad.serial)) |
| return False |
| return True |
| |
| def wait_for_conference(log, ad, participants=2, timeout=10): |
| """Wait for the droid to be in a conference with calls specified |
| in conf_calls. |
| |
| Args: |
| log: log object |
| ad: android device object |
| participants: conference participant count |
| |
| Returns: |
| call_id if success, None if fail. |
| """ |
| |
| def get_conference_id(callers): |
| for call_id in callers: |
| call_details = ad.droid.telecomCallGetCallById(call_id).get("Details") |
| if set([tel_defines.CALL_PROPERTY_CONFERENCE]).issubset(call_details.get("Properties")): |
| return call_id |
| return None |
| |
| log.info("waiting for conference {}".format(ad.serial)) |
| start_time = time.time() |
| while time.time() < start_time + timeout: |
| participant_callers = get_calls_in_states(log, ad, [tel_defines.CALL_STATE_ACTIVE]) |
| if (len(participant_callers) == participants + 1): |
| return get_conference_id(participant_callers) |
| time.sleep(1) |
| return None |
| |
| def get_call_id_children(log, ad, call_id): |
| """Return the list of calls that are children to call_id |
| |
| Args: |
| log: log object |
| ad: android device object |
| call_id: Conference call id |
| |
| Returns: |
| List containing call_ids. |
| """ |
| call = ad.droid.telecomCallGetCallById(call_id) |
| call_chld = set(call['Children']) |
| log.info("get_call_id_children droid {} call {} children {}".format( |
| ad.serial, call, call_chld)) |
| return call_chld |
| |
| def get_calls_in_states(log, ad, call_states): |
| """Return the list of calls that are any of the states passed in call_states |
| |
| Args: |
| log: log object |
| ad: android device object |
| call_states: List of desired call states |
| |
| Returns: |
| List containing call_ids. |
| """ |
| # Get the list of calls. |
| call_ids = ad.droid.telecomCallGetCallIds() |
| call_in_state = [] |
| for call_id in call_ids: |
| call = ad.droid.telecomCallGetCallById(call_id) |
| log.info("Call id: {} desc: {}".format(call_id, call)) |
| if call['State'] in call_states: |
| log.info("Adding call id {} to result set.".format(call_id)) |
| call_in_state.append(call_id) |
| return call_in_state |