| #!/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.utils import rand_ascii_str |
| from acts.libs.utils.multithread import multithread_func |
| from acts.libs.utils.multithread import run_multithread_func |
| from acts_contrib.test_utils.tel.tel_defines import EventCallStateChanged |
| from acts_contrib.test_utils.tel.tel_defines import EventMmsSentFailure |
| from acts_contrib.test_utils.tel.tel_defines import EventMmsSentSuccess |
| from acts_contrib.test_utils.tel.tel_defines import EventMmsDownloaded |
| from acts_contrib.test_utils.tel.tel_defines import EventSmsDeliverFailure |
| from acts_contrib.test_utils.tel.tel_defines import EventSmsDeliverSuccess |
| from acts_contrib.test_utils.tel.tel_defines import EventSmsReceived |
| from acts_contrib.test_utils.tel.tel_defines import EventSmsSentFailure |
| from acts_contrib.test_utils.tel.tel_defines import EventSmsSentSuccess |
| 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 MAX_WAIT_TIME_SMS_RECEIVE |
| from acts_contrib.test_utils.tel.tel_defines import MAX_WAIT_TIME_SMS_RECEIVE_IN_COLLISION |
| from acts_contrib.test_utils.tel.tel_defines import MAX_WAIT_TIME_SMS_SENT_SUCCESS_IN_COLLISION |
| from acts_contrib.test_utils.tel.tel_defines import SMS_OVER_WIFI_PROVIDERS |
| 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_IN_CALL |
| from acts_contrib.test_utils.tel.tel_defines import VT_STATE_BIDIRECTIONAL |
| from acts_contrib.test_utils.tel.tel_defines import WFC_MODE_DISABLED |
| from acts_contrib.test_utils.tel.tel_phone_setup_utils import phone_setup_on_rat |
| from acts_contrib.test_utils.tel.tel_subscription_utils import get_incoming_message_sub_id |
| from acts_contrib.test_utils.tel.tel_subscription_utils import get_outgoing_message_sub_id |
| from acts_contrib.test_utils.tel.tel_subscription_utils import get_subid_from_slot_index |
| from acts_contrib.test_utils.tel.tel_subscription_utils import get_subid_on_same_network_of_host_ad |
| from acts_contrib.test_utils.tel.tel_subscription_utils import set_subid_for_message |
| from acts_contrib.test_utils.tel.tel_test_utils import CallResult |
| from acts_contrib.test_utils.tel.tel_test_utils import TelResultWrapper |
| from acts_contrib.test_utils.tel.tel_test_utils import check_phone_number_match |
| from acts_contrib.test_utils.tel.tel_test_utils import get_device_epoch_time |
| from acts_contrib.test_utils.tel.tel_test_utils import toggle_airplane_mode |
| from acts_contrib.test_utils.tel.tel_voice_utils import call_setup_teardown |
| from acts_contrib.test_utils.tel.tel_voice_utils import hangup_call |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call |
| from acts_contrib.test_utils.tel.tel_voice_utils import last_call_drop_reason |
| from acts_contrib.test_utils.tel.tel_voice_utils import is_phone_in_call_on_rat |
| from acts_contrib.test_utils.tel.tel_voice_utils import wait_and_answer_call_for_subscription |
| from acts_contrib.test_utils.tel.tel_voice_utils import wait_for_in_call_active |
| from acts_contrib.test_utils.tel.tel_voice_utils import wait_for_call_end |
| from acts_contrib.test_utils.tel.tel_voice_utils import wait_for_call_offhook_for_subscription |
| from acts_contrib.test_utils.tel.tel_video_utils import is_phone_in_call_video_bidirectional |
| from acts_contrib.test_utils.tel.tel_video_utils import video_call_setup_teardown |
| from acts_contrib.test_utils.tel.tel_video_utils import phone_idle_video |
| from acts_contrib.test_utils.tel.tel_wifi_utils import ensure_wifi_connected |
| |
| |
| def send_message_with_random_message_body( |
| log, ad_mo, ad_mt, msg_type='sms', long_msg=False, mms_expected_result=True): |
| """Test SMS/MMS between two phones. |
| Returns: |
| True if success. |
| False if failed. |
| """ |
| message_lengths = (50, 160, 180) |
| |
| if long_msg: |
| message_lengths = (800, 1600) |
| message_lengths_of_jp_carriers = (800, 1530) |
| sender_message_sub_id = get_outgoing_message_sub_id(ad_mo) |
| sender_mcc = ad_mo.telephony["subscription"][sender_message_sub_id]["mcc"] |
| if str(sender_mcc) in ["440", "441"]: |
| message_lengths = message_lengths_of_jp_carriers |
| |
| if msg_type == 'sms': |
| for length in message_lengths: |
| message_array = [rand_ascii_str(length)] |
| if not sms_send_receive_verify(log, ad_mo, ad_mt, message_array): |
| ad_mo.log.error("SMS of length %s test failed", length) |
| return False |
| else: |
| ad_mo.log.info("SMS of length %s test succeeded", length) |
| log.info("SMS test of length %s characters succeeded.", |
| message_lengths) |
| elif msg_type == 'mms': |
| is_roaming = False |
| for ad in [ad_mo, ad_mt]: |
| ad.sms_over_wifi = False |
| # verizon supports sms over wifi. will add more carriers later |
| for sub in ad.telephony["subscription"].values(): |
| if sub["operator"] in SMS_OVER_WIFI_PROVIDERS: |
| ad.sms_over_wifi = True |
| |
| if getattr(ad, 'roaming', False): |
| is_roaming = True |
| |
| if is_roaming: |
| # roaming device does not allow message of length 180 |
| message_lengths = (50, 160) |
| |
| for length in message_lengths: |
| message_array = [("Test Message", rand_ascii_str(length), None)] |
| result = True |
| if not mms_send_receive_verify( |
| log, |
| ad_mo, |
| ad_mt, |
| message_array, |
| expected_result=mms_expected_result): |
| |
| if mms_expected_result is True: |
| if ad_mo.droid.telecomIsInCall() or ad_mt.droid.telecomIsInCall(): |
| if not mms_receive_verify_after_call_hangup( |
| log, ad_mo, ad_mt, message_array): |
| result = False |
| else: |
| result = False |
| |
| if not result: |
| log.error("MMS of body length %s test failed", length) |
| return False |
| else: |
| log.info("MMS of body length %s test succeeded", length) |
| log.info("MMS test of body lengths %s succeeded", message_lengths) |
| return True |
| |
| def message_test( |
| log, |
| ad_mo, |
| ad_mt, |
| mo_rat='general', |
| mt_rat='general', |
| msg_type='sms', |
| long_msg=False, |
| mms_expected_result=True, |
| msg_in_call=False, |
| video_or_voice='voice', |
| is_airplane_mode=False, |
| wfc_mode=None, |
| wifi_ssid=None, |
| wifi_pwd=None): |
| |
| mo_phone_setup_argv = ( |
| log, ad_mo, 'general', None, False, None, None, None, None, 'sms') |
| mt_phone_setup_argv = ( |
| log, ad_mt, 'general', None, False, None, None, None, None, 'sms') |
| verify_caller_func = None |
| verify_callee_func = None |
| |
| if mo_rat: |
| mo_phone_setup_argv = ( |
| log, |
| ad_mo, |
| mo_rat, |
| None, |
| is_airplane_mode, |
| wfc_mode, |
| wifi_ssid, |
| wifi_pwd, |
| None, |
| 'sms') |
| verify_caller_func = is_phone_in_call_on_rat( |
| log, ad_mo, rat=mo_rat, only_return_fn=True) |
| |
| if mt_rat: |
| mt_phone_setup_argv = ( |
| log, |
| ad_mt, |
| mt_rat, |
| None, |
| is_airplane_mode, |
| wfc_mode, |
| wifi_ssid, |
| wifi_pwd, |
| None, |
| 'sms') |
| verify_callee_func = is_phone_in_call_on_rat( |
| log, ad_mo, rat=mt_rat, only_return_fn=True) |
| |
| tasks = [(phone_setup_on_rat, mo_phone_setup_argv), |
| (phone_setup_on_rat, mt_phone_setup_argv)] |
| if not multithread_func(log, tasks): |
| log.error("Phone Failed to Set Up Properly.") |
| return False |
| time.sleep(WAIT_TIME_ANDROID_STATE_SETTLING) |
| |
| if wifi_ssid: |
| if not wfc_mode or wfc_mode == WFC_MODE_DISABLED: |
| tasks = [(ensure_wifi_connected, (log, ad_mo, wifi_ssid, wifi_pwd)), |
| (ensure_wifi_connected, (log, ad_mt, wifi_ssid, wifi_pwd))] |
| if not multithread_func(log, tasks): |
| log.error("Failed to connected to Wi-Fi.") |
| return False |
| |
| if msg_in_call: |
| if video_or_voice == 'voice': |
| if not call_setup_teardown( |
| log, |
| ad_mo, |
| ad_mt, |
| ad_hangup=None, |
| verify_caller_func=verify_caller_func, |
| verify_callee_func=verify_callee_func): |
| log.error("Failed to setup a voice call") |
| return False |
| elif video_or_voice == 'video': |
| tasks = [ |
| (phone_idle_video, (log, ad_mo)), |
| (phone_idle_video, (log, ad_mt))] |
| if not multithread_func(log, tasks): |
| log.error("Phone Failed to Set Up Properly.") |
| return False |
| time.sleep(WAIT_TIME_ANDROID_STATE_SETTLING) |
| if not video_call_setup_teardown( |
| log, |
| ad_mo, |
| ad_mt, |
| None, |
| video_state=VT_STATE_BIDIRECTIONAL, |
| verify_caller_func=is_phone_in_call_video_bidirectional, |
| verify_callee_func=is_phone_in_call_video_bidirectional): |
| log.error("Failed to setup a video call") |
| return False |
| |
| result = True |
| if not send_message_with_random_message_body( |
| log, ad_mo, ad_mt, msg_type, long_msg, mms_expected_result): |
| log.error("Test failed.") |
| result = False |
| |
| if msg_in_call: |
| if not hangup_call(log, ad_mo): |
| ad_mo.log.info("Failed to hang up call!") |
| result = False |
| |
| return result |
| |
| def sms_send_receive_verify(log, |
| ad_tx, |
| ad_rx, |
| array_message, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE, |
| expected_result=True, |
| slot_id_rx=None): |
| """Send SMS, receive SMS, and verify content and sender's number. |
| |
| Send (several) SMS from droid_tx to droid_rx. |
| Verify SMS is sent, delivered and received. |
| Verify received content and sender's number are correct. |
| |
| Args: |
| log: Log object. |
| ad_tx: Sender's Android Device Object |
| ad_rx: Receiver's Android Device Object |
| array_message: the array of message to send/receive |
| slot_id_rx: the slot on the Receiver's android device (0/1) |
| """ |
| subid_tx = get_outgoing_message_sub_id(ad_tx) |
| if slot_id_rx is None: |
| subid_rx = get_incoming_message_sub_id(ad_rx) |
| else: |
| subid_rx = get_subid_from_slot_index(log, ad_rx, slot_id_rx) |
| |
| result = sms_send_receive_verify_for_subscription( |
| log, ad_tx, ad_rx, subid_tx, subid_rx, array_message, max_wait_time) |
| if result != expected_result: |
| log_messaging_screen_shot(ad_tx, test_name="sms_tx") |
| log_messaging_screen_shot(ad_rx, test_name="sms_rx") |
| return result == expected_result |
| |
| def sms_send_receive_verify_for_subscription( |
| log, |
| ad_tx, |
| ad_rx, |
| subid_tx, |
| subid_rx, |
| array_message, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE): |
| """Send SMS, receive SMS, and verify content and sender's number. |
| |
| Send (several) SMS from droid_tx to droid_rx. |
| Verify SMS is sent, delivered and received. |
| Verify received content and sender's number are correct. |
| |
| Args: |
| log: Log object. |
| ad_tx: Sender's Android Device Object.. |
| ad_rx: Receiver's Android Device Object. |
| subid_tx: Sender's subscription ID to be used for SMS |
| subid_rx: Receiver's subscription ID to be used for SMS |
| array_message: the array of message to send/receive |
| """ |
| phonenumber_tx = ad_tx.telephony['subscription'][subid_tx]['phone_num'] |
| phonenumber_rx = ad_rx.telephony['subscription'][subid_rx]['phone_num'] |
| |
| for ad in (ad_tx, ad_rx): |
| if not getattr(ad, "messaging_droid", None): |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| else: |
| try: |
| if not ad.messaging_droid.is_live: |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| else: |
| ad.messaging_ed.clear_all_events() |
| ad.messaging_droid.logI( |
| "Start sms_send_receive_verify_for_subscription test") |
| except Exception: |
| ad.log.info("Create new sl4a session for messaging") |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| |
| for text in array_message: |
| length = len(text) |
| ad_tx.log.info("Sending SMS from %s to %s, len: %s, content: %s.", |
| phonenumber_tx, phonenumber_rx, length, text) |
| try: |
| ad_rx.messaging_ed.clear_events(EventSmsReceived) |
| ad_tx.messaging_ed.clear_events(EventSmsSentSuccess) |
| ad_tx.messaging_ed.clear_events(EventSmsSentFailure) |
| ad_rx.messaging_droid.smsStartTrackingIncomingSmsMessage() |
| time.sleep(1) #sleep 100ms after starting event tracking |
| ad_tx.messaging_droid.logI("Sending SMS of length %s" % length) |
| ad_rx.messaging_droid.logI("Expecting SMS of length %s" % length) |
| ad_tx.messaging_droid.smsSendTextMessage(phonenumber_rx, text, |
| True) |
| try: |
| events = ad_tx.messaging_ed.pop_events( |
| "(%s|%s|%s|%s)" % |
| (EventSmsSentSuccess, EventSmsSentFailure, |
| EventSmsDeliverSuccess, |
| EventSmsDeliverFailure), max_wait_time) |
| for event in events: |
| ad_tx.log.info("Got event %s", event["name"]) |
| if event["name"] == EventSmsSentFailure or event["name"] == EventSmsDeliverFailure: |
| if event.get("data") and event["data"].get("Reason"): |
| ad_tx.log.error("%s with reason: %s", |
| event["name"], |
| event["data"]["Reason"]) |
| return False |
| elif event["name"] == EventSmsSentSuccess or event["name"] == EventSmsDeliverSuccess: |
| break |
| except Empty: |
| ad_tx.log.error("No %s or %s event for SMS of length %s.", |
| EventSmsSentSuccess, EventSmsSentFailure, |
| length) |
| return False |
| |
| if not wait_for_matching_sms( |
| log, |
| ad_rx, |
| phonenumber_tx, |
| text, |
| max_wait_time, |
| allow_multi_part_long_sms=True): |
| ad_rx.log.error("No matching received SMS of length %s.", |
| length) |
| return False |
| except Exception as e: |
| log.error("Exception error %s", e) |
| raise |
| finally: |
| ad_rx.messaging_droid.smsStopTrackingIncomingSmsMessage() |
| return True |
| |
| def sms_in_collision_send_receive_verify( |
| log, |
| ad_rx, |
| ad_rx2, |
| ad_tx, |
| ad_tx2, |
| array_message, |
| array_message2, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE_IN_COLLISION): |
| """Send 2 SMS', receive both SMS', and verify content and sender's number of |
| each SMS. |
| |
| Send 2 SMS'. One from ad_tx to ad_rx and the other from ad_tx2 to ad_rx2. |
| When ad_rx is identical to ad_rx2, the scenario of SMS' in collision can |
| be tested. |
| Verify both SMS' are sent, delivered and received. |
| Verify received content and sender's number of each SMS is correct. |
| |
| Args: |
| log: Log object. |
| ad_tx: Sender's Android Device Object.. |
| ad_rx: Receiver's Android Device Object. |
| ad_tx2: 2nd sender's Android Device Object.. |
| ad_rx2: 2nd receiver's Android Device Object. |
| array_message: the array of message to send/receive from ad_tx to ad_rx |
| array_message2: the array of message to send/receive from ad_tx2 to |
| ad_rx2 |
| max_wait_time: Max time to wait for reception of SMS |
| """ |
| rx_sub_id = get_outgoing_message_sub_id(ad_rx) |
| rx2_sub_id = get_outgoing_message_sub_id(ad_rx2) |
| |
| _, tx_sub_id, _ = get_subid_on_same_network_of_host_ad( |
| [ad_rx, ad_tx, ad_tx2], |
| host_sub_id=rx_sub_id) |
| set_subid_for_message(ad_tx, tx_sub_id) |
| |
| _, _, tx2_sub_id = get_subid_on_same_network_of_host_ad( |
| [ad_rx2, ad_tx, ad_tx2], |
| host_sub_id=rx2_sub_id) |
| set_subid_for_message(ad_tx2, tx2_sub_id) |
| |
| if not sms_in_collision_send_receive_verify_for_subscription( |
| log, |
| ad_tx, |
| ad_tx2, |
| ad_rx, |
| ad_rx2, |
| tx_sub_id, |
| tx2_sub_id, |
| rx_sub_id, |
| rx_sub_id, |
| array_message, |
| array_message2, |
| max_wait_time): |
| log_messaging_screen_shot( |
| ad_rx, test_name="sms rx subid: %s" % rx_sub_id) |
| log_messaging_screen_shot( |
| ad_rx2, test_name="sms rx2 subid: %s" % rx2_sub_id) |
| log_messaging_screen_shot( |
| ad_tx, test_name="sms tx subid: %s" % tx_sub_id) |
| log_messaging_screen_shot( |
| ad_tx2, test_name="sms tx subid: %s" % tx2_sub_id) |
| return False |
| return True |
| |
| def sms_in_collision_send_receive_verify_for_subscription( |
| log, |
| ad_tx, |
| ad_tx2, |
| ad_rx, |
| ad_rx2, |
| subid_tx, |
| subid_tx2, |
| subid_rx, |
| subid_rx2, |
| array_message, |
| array_message2, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE_IN_COLLISION): |
| """Send 2 SMS', receive both SMS', and verify content and sender's number of |
| each SMS. |
| |
| Send 2 SMS'. One from ad_tx to ad_rx and the other from ad_tx2 to ad_rx2. |
| When ad_rx is identical to ad_rx2, the scenario of SMS' in collision can |
| be tested. |
| Verify both SMS' are sent, delivered and received. |
| Verify received content and sender's number of each SMS is correct. |
| |
| Args: |
| log: Log object. |
| ad_tx: Sender's Android Device Object.. |
| ad_rx: Receiver's Android Device Object. |
| ad_tx2: 2nd sender's Android Device Object.. |
| ad_rx2: 2nd receiver's Android Device Object. |
| subid_tx: Sub ID of ad_tx as default Sub ID for outgoing SMS |
| subid_tx2: Sub ID of ad_tx2 as default Sub ID for outgoing SMS |
| subid_rx: Sub ID of ad_rx as default Sub ID for incoming SMS |
| subid_rx2: Sub ID of ad_rx2 as default Sub ID for incoming SMS |
| array_message: the array of message to send/receive from ad_tx to ad_rx |
| array_message2: the array of message to send/receive from ad_tx2 to |
| ad_rx2 |
| max_wait_time: Max time to wait for reception of SMS |
| """ |
| |
| phonenumber_tx = ad_tx.telephony['subscription'][subid_tx]['phone_num'] |
| phonenumber_tx2 = ad_tx2.telephony['subscription'][subid_tx2]['phone_num'] |
| phonenumber_rx = ad_rx.telephony['subscription'][subid_rx]['phone_num'] |
| phonenumber_rx2 = ad_rx2.telephony['subscription'][subid_rx2]['phone_num'] |
| |
| for ad in (ad_tx, ad_tx2, ad_rx, ad_rx2): |
| ad.send_keycode("BACK") |
| if not getattr(ad, "messaging_droid", None): |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| else: |
| try: |
| if not ad.messaging_droid.is_live: |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| else: |
| ad.messaging_ed.clear_all_events() |
| ad.messaging_droid.logI( |
| "Start sms_send_receive_verify_for_subscription test") |
| except Exception: |
| ad.log.info("Create new sl4a session for messaging") |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| |
| for text, text2 in zip(array_message, array_message2): |
| length = len(text) |
| length2 = len(text2) |
| ad_tx.log.info("Sending SMS from %s to %s, len: %s, content: %s.", |
| phonenumber_tx, phonenumber_rx, length, text) |
| ad_tx2.log.info("Sending SMS from %s to %s, len: %s, content: %s.", |
| phonenumber_tx2, phonenumber_rx2, length2, text2) |
| |
| try: |
| ad_rx.messaging_ed.clear_events(EventSmsReceived) |
| ad_rx2.messaging_ed.clear_events(EventSmsReceived) |
| ad_tx.messaging_ed.clear_events(EventSmsSentSuccess) |
| ad_tx.messaging_ed.clear_events(EventSmsSentFailure) |
| ad_tx2.messaging_ed.clear_events(EventSmsSentSuccess) |
| ad_tx2.messaging_ed.clear_events(EventSmsSentFailure) |
| ad_rx.messaging_droid.smsStartTrackingIncomingSmsMessage() |
| if ad_rx2 != ad_rx: |
| ad_rx2.messaging_droid.smsStartTrackingIncomingSmsMessage() |
| time.sleep(1) |
| ad_tx.messaging_droid.logI("Sending SMS of length %s" % length) |
| ad_tx2.messaging_droid.logI("Sending SMS of length %s" % length2) |
| ad_rx.messaging_droid.logI( |
| "Expecting SMS of length %s from %s" % (length, ad_tx.serial)) |
| ad_rx2.messaging_droid.logI( |
| "Expecting SMS of length %s from %s" % (length2, ad_tx2.serial)) |
| |
| tasks = [ |
| (ad_tx.messaging_droid.smsSendTextMessage, |
| (phonenumber_rx, text, True)), |
| (ad_tx2.messaging_droid.smsSendTextMessage, |
| (phonenumber_rx2, text2, True))] |
| multithread_func(log, tasks) |
| try: |
| tasks = [ |
| (ad_tx.messaging_ed.pop_events, ("(%s|%s|%s|%s)" % ( |
| EventSmsSentSuccess, |
| EventSmsSentFailure, |
| EventSmsDeliverSuccess, |
| EventSmsDeliverFailure), max_wait_time)), |
| (ad_tx2.messaging_ed.pop_events, ("(%s|%s|%s|%s)" % ( |
| EventSmsSentSuccess, |
| EventSmsSentFailure, |
| EventSmsDeliverSuccess, |
| EventSmsDeliverFailure), max_wait_time)) |
| ] |
| results = run_multithread_func(log, tasks) |
| res = True |
| _ad = ad_tx |
| for ad, events in [(ad_tx, results[0]),(ad_tx2, results[1])]: |
| _ad = ad |
| for event in events: |
| ad.log.info("Got event %s", event["name"]) |
| if event["name"] == EventSmsSentFailure or \ |
| event["name"] == EventSmsDeliverFailure: |
| if event.get("data") and event["data"].get("Reason"): |
| ad.log.error("%s with reason: %s", |
| event["name"], |
| event["data"]["Reason"]) |
| res = False |
| elif event["name"] == EventSmsSentSuccess or \ |
| event["name"] == EventSmsDeliverSuccess: |
| break |
| if not res: |
| return False |
| except Empty: |
| _ad.log.error("No %s or %s event for SMS of length %s.", |
| EventSmsSentSuccess, EventSmsSentFailure, |
| length) |
| return False |
| if ad_rx == ad_rx2: |
| if not wait_for_matching_mt_sms_in_collision( |
| log, |
| ad_rx, |
| phonenumber_tx, |
| phonenumber_tx2, |
| text, |
| text2, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE_IN_COLLISION): |
| |
| ad_rx.log.error( |
| "No matching received SMS of length %s from %s.", |
| length, |
| ad_rx.serial) |
| return False |
| else: |
| if not wait_for_matching_mt_sms_in_collision_with_mo_sms( |
| log, |
| ad_rx, |
| ad_rx2, |
| phonenumber_tx, |
| phonenumber_tx2, |
| text, |
| text2, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE_IN_COLLISION): |
| return False |
| except Exception as e: |
| log.error("Exception error %s", e) |
| raise |
| finally: |
| ad_rx.messaging_droid.smsStopTrackingIncomingSmsMessage() |
| ad_rx2.messaging_droid.smsStopTrackingIncomingSmsMessage() |
| return True |
| |
| def sms_rx_power_off_multiple_send_receive_verify( |
| log, |
| ad_rx, |
| ad_tx, |
| ad_tx2, |
| array_message_length, |
| array_message2_length, |
| num_array_message, |
| num_array_message2, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE_IN_COLLISION): |
| |
| rx_sub_id = get_outgoing_message_sub_id(ad_rx) |
| |
| _, tx_sub_id, _ = get_subid_on_same_network_of_host_ad( |
| [ad_rx, ad_tx, ad_tx2], |
| host_sub_id=rx_sub_id) |
| set_subid_for_message(ad_tx, tx_sub_id) |
| |
| _, _, tx2_sub_id = get_subid_on_same_network_of_host_ad( |
| [ad_rx, ad_tx, ad_tx2], |
| host_sub_id=rx_sub_id) |
| set_subid_for_message(ad_tx2, tx2_sub_id) |
| |
| if not sms_rx_power_off_multiple_send_receive_verify_for_subscription( |
| log, |
| ad_tx, |
| ad_tx2, |
| ad_rx, |
| tx_sub_id, |
| tx2_sub_id, |
| rx_sub_id, |
| rx_sub_id, |
| array_message_length, |
| array_message2_length, |
| num_array_message, |
| num_array_message2): |
| log_messaging_screen_shot( |
| ad_rx, test_name="sms rx subid: %s" % rx_sub_id) |
| log_messaging_screen_shot( |
| ad_tx, test_name="sms tx subid: %s" % tx_sub_id) |
| log_messaging_screen_shot( |
| ad_tx2, test_name="sms tx subid: %s" % tx2_sub_id) |
| return False |
| return True |
| |
| def sms_rx_power_off_multiple_send_receive_verify_for_subscription( |
| log, |
| ad_tx, |
| ad_tx2, |
| ad_rx, |
| subid_tx, |
| subid_tx2, |
| subid_rx, |
| subid_rx2, |
| array_message_length, |
| array_message2_length, |
| num_array_message, |
| num_array_message2, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE_IN_COLLISION): |
| |
| phonenumber_tx = ad_tx.telephony['subscription'][subid_tx]['phone_num'] |
| phonenumber_tx2 = ad_tx2.telephony['subscription'][subid_tx2]['phone_num'] |
| phonenumber_rx = ad_rx.telephony['subscription'][subid_rx]['phone_num'] |
| phonenumber_rx2 = ad_rx.telephony['subscription'][subid_rx2]['phone_num'] |
| |
| if not toggle_airplane_mode(log, ad_rx, True): |
| ad_rx.log.error("Failed to enable Airplane Mode") |
| return False |
| ad_rx.stop_services() |
| ad_rx.log.info("Rebooting......") |
| ad_rx.adb.reboot() |
| |
| message_dict = {phonenumber_tx: [], phonenumber_tx2: []} |
| for index in range(max(num_array_message, num_array_message2)): |
| array_message = [rand_ascii_str(array_message_length)] |
| array_message2 = [rand_ascii_str(array_message2_length)] |
| for text, text2 in zip(array_message, array_message2): |
| message_dict[phonenumber_tx].append(text) |
| message_dict[phonenumber_tx2].append(text2) |
| length = len(text) |
| length2 = len(text2) |
| |
| ad_tx.log.info("Sending SMS from %s to %s, len: %s, content: %s.", |
| phonenumber_tx, phonenumber_rx, length, text) |
| ad_tx2.log.info("Sending SMS from %s to %s, len: %s, content: %s.", |
| phonenumber_tx2, phonenumber_rx2, length2, text2) |
| |
| try: |
| for ad in (ad_tx, ad_tx2): |
| ad.send_keycode("BACK") |
| if not getattr(ad, "messaging_droid", None): |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| else: |
| try: |
| if not ad.messaging_droid.is_live: |
| ad.messaging_droid, ad.messaging_ed = \ |
| ad.get_droid() |
| ad.messaging_ed.start() |
| else: |
| ad.messaging_ed.clear_all_events() |
| ad.messaging_droid.logI( |
| "Start sms_send_receive_verify_for_subscription" |
| " test") |
| except Exception: |
| ad.log.info("Create new sl4a session for messaging") |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| |
| ad_tx.messaging_ed.clear_events(EventSmsSentSuccess) |
| ad_tx.messaging_ed.clear_events(EventSmsSentFailure) |
| ad_tx2.messaging_ed.clear_events(EventSmsSentSuccess) |
| ad_tx2.messaging_ed.clear_events(EventSmsSentFailure) |
| |
| if index < num_array_message and index < num_array_message2: |
| ad_tx.messaging_droid.logI( |
| "Sending SMS of length %s" % length) |
| ad_tx2.messaging_droid.logI( |
| "Sending SMS of length %s" % length2) |
| tasks = [ |
| (ad_tx.messaging_droid.smsSendTextMessage, |
| (phonenumber_rx, text, True)), |
| (ad_tx2.messaging_droid.smsSendTextMessage, |
| (phonenumber_rx2, text2, True))] |
| multithread_func(log, tasks) |
| else: |
| if index < num_array_message: |
| ad_tx.messaging_droid.logI( |
| "Sending SMS of length %s" % length) |
| ad_tx.messaging_droid.smsSendTextMessage( |
| phonenumber_rx, text, True) |
| if index < num_array_message2: |
| ad_tx2.messaging_droid.logI( |
| "Sending SMS of length %s" % length2) |
| ad_tx2.messaging_droid.smsSendTextMessage( |
| phonenumber_rx2, text2, True) |
| |
| try: |
| if index < num_array_message and index < num_array_message2: |
| tasks = [ |
| (ad_tx.messaging_ed.pop_events, ("(%s|%s|%s|%s)" % ( |
| EventSmsSentSuccess, |
| EventSmsSentFailure, |
| EventSmsDeliverSuccess, |
| EventSmsDeliverFailure), |
| max_wait_time)), |
| (ad_tx2.messaging_ed.pop_events, ("(%s|%s|%s|%s)" % ( |
| EventSmsSentSuccess, |
| EventSmsSentFailure, |
| EventSmsDeliverSuccess, |
| EventSmsDeliverFailure), |
| max_wait_time)) |
| ] |
| results = run_multithread_func(log, tasks) |
| res = True |
| _ad = ad_tx |
| for ad, events in [ |
| (ad_tx, results[0]), (ad_tx2, results[1])]: |
| _ad = ad |
| for event in events: |
| ad.log.info("Got event %s", event["name"]) |
| if event["name"] == EventSmsSentFailure or \ |
| event["name"] == EventSmsDeliverFailure: |
| if event.get("data") and \ |
| event["data"].get("Reason"): |
| ad.log.error("%s with reason: %s", |
| event["name"], |
| event["data"]["Reason"]) |
| res = False |
| elif event["name"] == EventSmsSentSuccess or \ |
| event["name"] == EventSmsDeliverSuccess: |
| break |
| if not res: |
| return False |
| else: |
| if index < num_array_message: |
| result = ad_tx.messaging_ed.pop_events( |
| "(%s|%s|%s|%s)" % ( |
| EventSmsSentSuccess, |
| EventSmsSentFailure, |
| EventSmsDeliverSuccess, |
| EventSmsDeliverFailure), |
| max_wait_time) |
| res = True |
| _ad = ad_tx |
| for ad, events in [(ad_tx, result)]: |
| _ad = ad |
| for event in events: |
| ad.log.info("Got event %s", event["name"]) |
| if event["name"] == EventSmsSentFailure or \ |
| event["name"] == EventSmsDeliverFailure: |
| if event.get("data") and \ |
| event["data"].get("Reason"): |
| ad.log.error( |
| "%s with reason: %s", |
| event["name"], |
| event["data"]["Reason"]) |
| res = False |
| elif event["name"] == EventSmsSentSuccess \ |
| or event["name"] == EventSmsDeliverSuccess: |
| break |
| if not res: |
| return False |
| if index < num_array_message2: |
| result = ad_tx2.messaging_ed.pop_events( |
| "(%s|%s|%s|%s)" % ( |
| EventSmsSentSuccess, |
| EventSmsSentFailure, |
| EventSmsDeliverSuccess, |
| EventSmsDeliverFailure), |
| max_wait_time) |
| res = True |
| _ad = ad_tx2 |
| for ad, events in [(ad_tx2, result)]: |
| _ad = ad |
| for event in events: |
| ad.log.info("Got event %s", event["name"]) |
| if event["name"] == EventSmsSentFailure or \ |
| event["name"] == EventSmsDeliverFailure: |
| if event.get("data") and \ |
| event["data"].get("Reason"): |
| ad.log.error( |
| "%s with reason: %s", |
| event["name"], |
| event["data"]["Reason"]) |
| res = False |
| elif event["name"] == EventSmsSentSuccess \ |
| or event["name"] == EventSmsDeliverSuccess: |
| break |
| if not res: |
| return False |
| |
| |
| except Empty: |
| _ad.log.error("No %s or %s event for SMS of length %s.", |
| EventSmsSentSuccess, EventSmsSentFailure, |
| length) |
| return False |
| |
| except Exception as e: |
| log.error("Exception error %s", e) |
| raise |
| |
| ad_rx.wait_for_boot_completion() |
| ad_rx.root_adb() |
| ad_rx.start_services(skip_setup_wizard=False) |
| |
| output = ad_rx.adb.logcat("-t 1") |
| match = re.search(r"\d+-\d+\s\d+:\d+:\d+.\d+", output) |
| if match: |
| ad_rx.test_log_begin_time = match.group(0) |
| |
| ad_rx.messaging_droid, ad_rx.messaging_ed = ad_rx.get_droid() |
| ad_rx.messaging_ed.start() |
| ad_rx.messaging_droid.smsStartTrackingIncomingSmsMessage() |
| time.sleep(1) #sleep 100ms after starting event tracking |
| |
| if not toggle_airplane_mode(log, ad_rx, False): |
| ad_rx.log.error("Failed to disable Airplane Mode") |
| return False |
| |
| res = True |
| try: |
| if not wait_for_matching_multiple_sms(log, |
| ad_rx, |
| phonenumber_tx, |
| phonenumber_tx2, |
| messages=message_dict, |
| max_wait_time=max_wait_time): |
| res = False |
| except Exception as e: |
| log.error("Exception error %s", e) |
| raise |
| finally: |
| ad_rx.messaging_droid.smsStopTrackingIncomingSmsMessage() |
| |
| return res |
| |
| def is_sms_match(event, phonenumber_tx, text): |
| """Return True if 'text' equals to event['data']['Text'] |
| and phone number match. |
| |
| Args: |
| event: Event object to verify. |
| phonenumber_tx: phone number for sender. |
| text: text string to verify. |
| |
| Returns: |
| Return True if 'text' equals to event['data']['Text'] |
| and phone number match. |
| """ |
| return (check_phone_number_match(event['data']['Sender'], phonenumber_tx) |
| and event['data']['Text'].strip() == text) |
| |
| def is_sms_partial_match(event, phonenumber_tx, text): |
| """Return True if 'text' starts with event['data']['Text'] |
| and phone number match. |
| |
| Args: |
| event: Event object to verify. |
| phonenumber_tx: phone number for sender. |
| text: text string to verify. |
| |
| Returns: |
| Return True if 'text' starts with event['data']['Text'] |
| and phone number match. |
| """ |
| event_text = event['data']['Text'].strip() |
| if event_text.startswith("("): |
| event_text = event_text.split(")")[-1] |
| return (check_phone_number_match(event['data']['Sender'], phonenumber_tx) |
| and text.startswith(event_text)) |
| |
| def is_sms_in_collision_match( |
| event, phonenumber_tx, phonenumber_tx2, text, text2): |
| event_text = event['data']['Text'].strip() |
| if event_text.startswith("("): |
| event_text = event_text.split(")")[-1] |
| |
| for phonenumber, txt in [[phonenumber_tx, text], [phonenumber_tx2, text2]]: |
| if check_phone_number_match( |
| event['data']['Sender'], phonenumber) and txt.startswith(event_text): |
| return True |
| return False |
| |
| def is_sms_in_collision_partial_match( |
| event, phonenumber_tx, phonenumber_tx2, text, text2): |
| for phonenumber, txt in [[phonenumber_tx, text], [phonenumber_tx2, text2]]: |
| if check_phone_number_match( |
| event['data']['Sender'], phonenumber) and \ |
| event['data']['Text'].strip() == txt: |
| return True |
| return False |
| |
| def is_sms_match_among_multiple_sms( |
| event, phonenumber_tx, phonenumber_tx2, texts=[], texts2=[]): |
| for txt in texts: |
| if check_phone_number_match( |
| event['data']['Sender'], phonenumber_tx) and \ |
| event['data']['Text'].strip() == txt: |
| return True |
| |
| for txt in texts2: |
| if check_phone_number_match( |
| event['data']['Sender'], phonenumber_tx2) and \ |
| event['data']['Text'].strip() == txt: |
| return True |
| |
| return False |
| |
| def is_sms_partial_match_among_multiple_sms( |
| event, phonenumber_tx, phonenumber_tx2, texts=[], texts2=[]): |
| event_text = event['data']['Text'].strip() |
| if event_text.startswith("("): |
| event_text = event_text.split(")")[-1] |
| |
| for txt in texts: |
| if check_phone_number_match( |
| event['data']['Sender'], phonenumber_tx) and \ |
| txt.startswith(event_text): |
| return True |
| |
| for txt in texts2: |
| if check_phone_number_match( |
| event['data']['Sender'], phonenumber_tx2) and \ |
| txt.startswith(event_text): |
| return True |
| |
| return False |
| |
| def wait_for_matching_sms(log, |
| ad_rx, |
| phonenumber_tx, |
| text, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE, |
| allow_multi_part_long_sms=True): |
| """Wait for matching incoming SMS. |
| |
| Args: |
| log: Log object. |
| ad_rx: Receiver's Android Device Object |
| phonenumber_tx: Sender's phone number. |
| text: SMS content string. |
| allow_multi_part_long_sms: is long SMS allowed to be received as |
| multiple short SMS. This is optional, default value is True. |
| |
| Returns: |
| True if matching incoming SMS is received. |
| """ |
| if not allow_multi_part_long_sms: |
| try: |
| ad_rx.messaging_ed.wait_for_event(EventSmsReceived, is_sms_match, |
| max_wait_time, phonenumber_tx, |
| text) |
| ad_rx.log.info("Got event %s", EventSmsReceived) |
| return True |
| except Empty: |
| ad_rx.log.error("No matched SMS received event.") |
| return False |
| else: |
| try: |
| received_sms = '' |
| remaining_text = text |
| while (remaining_text != ''): |
| event = ad_rx.messaging_ed.wait_for_event( |
| EventSmsReceived, is_sms_partial_match, max_wait_time, |
| phonenumber_tx, remaining_text) |
| event_text = event['data']['Text'].split(")")[-1].strip() |
| event_text_length = len(event_text) |
| ad_rx.log.info("Got event %s of text length %s from %s", |
| EventSmsReceived, event_text_length, |
| phonenumber_tx) |
| remaining_text = remaining_text[event_text_length:] |
| received_sms += event_text |
| ad_rx.log.info("Received SMS of length %s", len(received_sms)) |
| return True |
| except Empty: |
| ad_rx.log.error( |
| "Missing SMS received event of text length %s from %s", |
| len(remaining_text), phonenumber_tx) |
| if received_sms != '': |
| ad_rx.log.error( |
| "Only received partial matched SMS of length %s", |
| len(received_sms)) |
| return False |
| |
| def wait_for_matching_mt_sms_in_collision(log, |
| ad_rx, |
| phonenumber_tx, |
| phonenumber_tx2, |
| text, |
| text2, |
| allow_multi_part_long_sms=True, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE_IN_COLLISION): |
| |
| if not allow_multi_part_long_sms: |
| try: |
| ad_rx.messaging_ed.wait_for_event( |
| EventSmsReceived, |
| is_sms_in_collision_match, |
| max_wait_time, |
| phonenumber_tx, |
| phonenumber_tx2, |
| text, |
| text2) |
| ad_rx.log.info("Got event %s", EventSmsReceived) |
| return True |
| except Empty: |
| ad_rx.log.error("No matched SMS received event.") |
| return False |
| else: |
| try: |
| received_sms = '' |
| received_sms2 = '' |
| remaining_text = text |
| remaining_text2 = text2 |
| while (remaining_text != '' or remaining_text2 != ''): |
| event = ad_rx.messaging_ed.wait_for_event( |
| EventSmsReceived, |
| is_sms_in_collision_partial_match, |
| max_wait_time, |
| phonenumber_tx, |
| phonenumber_tx2, |
| remaining_text, |
| remaining_text2) |
| event_text = event['data']['Text'].split(")")[-1].strip() |
| event_text_length = len(event_text) |
| |
| if event_text in remaining_text: |
| ad_rx.log.info("Got event %s of text length %s from %s", |
| EventSmsReceived, event_text_length, |
| phonenumber_tx) |
| remaining_text = remaining_text[event_text_length:] |
| received_sms += event_text |
| elif event_text in remaining_text2: |
| ad_rx.log.info("Got event %s of text length %s from %s", |
| EventSmsReceived, event_text_length, |
| phonenumber_tx2) |
| remaining_text2 = remaining_text2[event_text_length:] |
| received_sms2 += event_text |
| |
| ad_rx.log.info("Received SMS of length %s", len(received_sms)) |
| ad_rx.log.info("Received SMS of length %s", len(received_sms2)) |
| return True |
| except Empty: |
| ad_rx.log.error( |
| "Missing SMS received event.") |
| if received_sms != '': |
| ad_rx.log.error( |
| "Only received partial matched SMS of length %s from %s", |
| len(received_sms), phonenumber_tx) |
| if received_sms2 != '': |
| ad_rx.log.error( |
| "Only received partial matched SMS of length %s from %s", |
| len(received_sms2), phonenumber_tx2) |
| return False |
| |
| def wait_for_matching_mt_sms_in_collision_with_mo_sms(log, |
| ad_rx, |
| ad_rx2, |
| phonenumber_tx, |
| phonenumber_tx2, |
| text, |
| text2, |
| allow_multi_part_long_sms=True, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE_IN_COLLISION): |
| |
| if not allow_multi_part_long_sms: |
| result = True |
| try: |
| ad_rx.messaging_ed.wait_for_call_offhook_event( |
| EventSmsReceived, |
| is_sms_match, |
| max_wait_time, |
| phonenumber_tx, |
| text) |
| ad_rx.log.info("Got event %s", EventSmsReceived) |
| except Empty: |
| ad_rx.log.error("No matched SMS received event.") |
| result = False |
| |
| try: |
| ad_rx2.messaging_ed.wait_for_call_offhook_event( |
| EventSmsReceived, |
| is_sms_match, |
| max_wait_time, |
| phonenumber_tx2, |
| text2) |
| ad_rx2.log.info("Got event %s", EventSmsReceived) |
| except Empty: |
| ad_rx2.log.error("No matched SMS received event.") |
| result = False |
| |
| return result |
| else: |
| result = True |
| try: |
| received_sms = '' |
| remaining_text = text |
| while remaining_text != '': |
| event = ad_rx.messaging_ed.wait_for_event( |
| EventSmsReceived, is_sms_partial_match, max_wait_time, |
| phonenumber_tx, remaining_text) |
| event_text = event['data']['Text'].split(")")[-1].strip() |
| event_text_length = len(event_text) |
| |
| if event_text in remaining_text: |
| ad_rx.log.info("Got event %s of text length %s from %s", |
| EventSmsReceived, event_text_length, |
| phonenumber_tx) |
| remaining_text = remaining_text[event_text_length:] |
| received_sms += event_text |
| |
| ad_rx.log.info("Received SMS of length %s", len(received_sms)) |
| except Empty: |
| ad_rx.log.error( |
| "Missing SMS received event.") |
| if received_sms != '': |
| ad_rx.log.error( |
| "Only received partial matched SMS of length %s from %s", |
| len(received_sms), phonenumber_tx) |
| result = False |
| |
| try: |
| received_sms2 = '' |
| remaining_text2 = text2 |
| while remaining_text2 != '': |
| event2 = ad_rx2.messaging_ed.wait_for_event( |
| EventSmsReceived, is_sms_partial_match, max_wait_time, |
| phonenumber_tx2, remaining_text2) |
| event_text2 = event2['data']['Text'].split(")")[-1].strip() |
| event_text_length2 = len(event_text2) |
| |
| if event_text2 in remaining_text2: |
| ad_rx2.log.info("Got event %s of text length %s from %s", |
| EventSmsReceived, event_text_length2, |
| phonenumber_tx2) |
| remaining_text2 = remaining_text2[event_text_length2:] |
| received_sms2 += event_text2 |
| |
| ad_rx2.log.info("Received SMS of length %s", len(received_sms2)) |
| except Empty: |
| ad_rx2.log.error( |
| "Missing SMS received event.") |
| if received_sms2 != '': |
| ad_rx2.log.error( |
| "Only received partial matched SMS of length %s from %s", |
| len(received_sms2), phonenumber_tx2) |
| result = False |
| |
| return result |
| |
| def wait_for_matching_multiple_sms(log, |
| ad_rx, |
| phonenumber_tx, |
| phonenumber_tx2, |
| messages={}, |
| allow_multi_part_long_sms=True, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE): |
| |
| if not allow_multi_part_long_sms: |
| try: |
| ad_rx.messaging_ed.wait_for_event( |
| EventSmsReceived, |
| is_sms_match_among_multiple_sms, |
| max_wait_time, |
| phonenumber_tx, |
| phonenumber_tx2, |
| messages[phonenumber_tx], |
| messages[phonenumber_tx2]) |
| ad_rx.log.info("Got event %s", EventSmsReceived) |
| return True |
| except Empty: |
| ad_rx.log.error("No matched SMS received event.") |
| return False |
| else: |
| all_msgs = [] |
| for tx, msgs in messages.items(): |
| for msg in msgs: |
| all_msgs.append([tx, msg, msg, '']) |
| |
| all_msgs_copy = all_msgs.copy() |
| |
| try: |
| while (all_msgs != []): |
| event = ad_rx.messaging_ed.wait_for_event( |
| EventSmsReceived, |
| is_sms_partial_match_among_multiple_sms, |
| max_wait_time, |
| phonenumber_tx, |
| phonenumber_tx2, |
| messages[phonenumber_tx], |
| messages[phonenumber_tx2]) |
| event_text = event['data']['Text'].split(")")[-1].strip() |
| event_text_length = len(event_text) |
| |
| for msg in all_msgs_copy: |
| if event_text in msg[2]: |
| ad_rx.log.info("Got event %s of text length %s from %s", |
| EventSmsReceived, event_text_length, |
| msg[0]) |
| msg[2] = msg[2][event_text_length:] |
| msg[3] += event_text |
| |
| if msg[2] == "": |
| all_msgs.remove(msg) |
| |
| ad_rx.log.info("Received all SMS' sent when power-off.") |
| except Empty: |
| ad_rx.log.error( |
| "Missing SMS received event.") |
| |
| for msg in all_msgs_copy: |
| if msg[3] != '': |
| ad_rx.log.error( |
| "Only received partial matched SMS of length %s from %s", |
| len(msg[3]), msg[0]) |
| return False |
| |
| return True |
| |
| def wait_for_sending_sms(ad_tx, max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE): |
| try: |
| events = ad_tx.messaging_ed.pop_events( |
| "(%s|%s|%s|%s)" % |
| (EventSmsSentSuccess, EventSmsSentFailure, |
| EventSmsDeliverSuccess, |
| EventSmsDeliverFailure), max_wait_time) |
| for event in events: |
| ad_tx.log.info("Got event %s", event["name"]) |
| if event["name"] == EventSmsSentFailure or \ |
| event["name"] == EventSmsDeliverFailure: |
| if event.get("data") and event["data"].get("Reason"): |
| ad_tx.log.error("%s with reason: %s", |
| event["name"], |
| event["data"]["Reason"]) |
| return False |
| elif event["name"] == EventSmsSentSuccess or \ |
| event["name"] == EventSmsDeliverSuccess: |
| return True |
| except Empty: |
| ad_tx.log.error("No %s or %s event for SMS.", |
| EventSmsSentSuccess, EventSmsSentFailure) |
| return False |
| |
| def voice_call_in_collision_with_mt_sms_msim( |
| log, |
| ad_primary, |
| ad_sms, |
| ad_voice, |
| sms_subid_ad_primary, |
| sms_subid_ad_sms, |
| voice_subid_ad_primary, |
| voice_subid_ad_voice, |
| array_message, |
| ad_hangup=None, |
| verify_caller_func=None, |
| verify_callee_func=None, |
| call_direction="mo", |
| wait_time_in_call=WAIT_TIME_IN_CALL, |
| incall_ui_display=INCALL_UI_DISPLAY_FOREGROUND, |
| dialing_number_length=None, |
| video_state=None): |
| |
| ad_tx = ad_sms |
| ad_rx = ad_primary |
| subid_tx = sms_subid_ad_sms |
| subid_rx = sms_subid_ad_primary |
| |
| if call_direction == "mo": |
| ad_caller = ad_primary |
| ad_callee = ad_voice |
| subid_caller = voice_subid_ad_primary |
| subid_callee = voice_subid_ad_voice |
| elif call_direction == "mt": |
| ad_callee = ad_primary |
| ad_caller = ad_voice |
| subid_callee = voice_subid_ad_primary |
| subid_caller = voice_subid_ad_voice |
| |
| |
| phonenumber_tx = ad_tx.telephony['subscription'][subid_tx]['phone_num'] |
| phonenumber_rx = ad_rx.telephony['subscription'][subid_rx]['phone_num'] |
| |
| tel_result_wrapper = TelResultWrapper(CallResult('SUCCESS')) |
| |
| for ad in (ad_tx, ad_rx): |
| ad.send_keycode("BACK") |
| if not getattr(ad, "messaging_droid", None): |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| else: |
| try: |
| if not ad.messaging_droid.is_live: |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| else: |
| ad.messaging_ed.clear_all_events() |
| except Exception: |
| ad.log.info("Create new sl4a session for messaging") |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| |
| 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'] |
| if dialing_number_length: |
| skip_test = False |
| 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:] |
| except: |
| skip_test = True |
| if caller_area_code != callee_area_code: |
| skip_test = True |
| if skip_test: |
| msg = "Cannot make call from %s to %s by %s digits" % ( |
| caller_number, callee_number, dialing_number_length) |
| ad_caller.log.info(msg) |
| raise signals.TestSkip(msg) |
| else: |
| callee_number = callee_dial_number |
| |
| 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) |
| |
| ad_caller.ed.clear_events(EventCallStateChanged) |
| call_begin_time = get_device_epoch_time(ad) |
| ad_caller.droid.telephonyStartTrackingCallStateForSubscription(subid_caller) |
| |
| for text in array_message: |
| length = len(text) |
| ad_tx.log.info("Sending SMS from %s to %s, len: %s, content: %s.", |
| phonenumber_tx, phonenumber_rx, length, text) |
| try: |
| ad_rx.messaging_ed.clear_events(EventSmsReceived) |
| ad_tx.messaging_ed.clear_events(EventSmsSentSuccess) |
| ad_tx.messaging_ed.clear_events(EventSmsSentFailure) |
| ad_rx.messaging_droid.smsStartTrackingIncomingSmsMessage() |
| time.sleep(1) #sleep 100ms after starting event tracking |
| ad_tx.messaging_droid.logI("Sending SMS of length %s" % length) |
| ad_rx.messaging_droid.logI("Expecting SMS of length %s" % length) |
| ad_caller.log.info("Make a phone call to %s", callee_number) |
| |
| tasks = [ |
| (ad_tx.messaging_droid.smsSendTextMessage, |
| (phonenumber_rx, text, True)), |
| (ad_caller.droid.telecomCallNumber, |
| (callee_number, video))] |
| |
| run_multithread_func(log, tasks) |
| |
| try: |
| # Verify OFFHOOK state |
| if not wait_for_call_offhook_for_subscription( |
| log, |
| ad_caller, |
| subid_caller, |
| event_tracking_started=True): |
| ad_caller.log.info( |
| "sub_id %s not in call offhook state", subid_caller) |
| last_call_drop_reason(ad_caller, begin_time=call_begin_time) |
| |
| 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") |
| finally: |
| ad_caller.droid.telephonyStopTrackingCallStateChangeForSubscription( |
| subid_caller) |
| if incall_ui_display == INCALL_UI_DISPLAY_FOREGROUND: |
| ad_caller.droid.telecomShowInCallScreen() |
| elif incall_ui_display == INCALL_UI_DISPLAY_BACKGROUND: |
| ad_caller.droid.showHomeScreen() |
| |
| 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 not wait_for_sending_sms( |
| ad_tx, |
| max_wait_time=MAX_WAIT_TIME_SMS_SENT_SUCCESS_IN_COLLISION): |
| return False |
| |
| tasks = [ |
| (wait_for_matching_sms, |
| (log, ad_rx, phonenumber_tx, text, |
| MAX_WAIT_TIME_SMS_RECEIVE_IN_COLLISION, True)), |
| (wait_for_call_end, |
| (log, ad_caller, ad_callee, ad_hangup, verify_caller_func, |
| verify_callee_func, call_begin_time, 5, tel_result_wrapper, |
| WAIT_TIME_IN_CALL))] |
| |
| results = run_multithread_func(log, tasks) |
| |
| if not results[0]: |
| ad_rx.log.error("No matching received SMS of length %s.", |
| length) |
| return False |
| |
| tel_result_wrapper = results[1] |
| |
| except Exception as e: |
| log.error("Exception error %s", e) |
| raise |
| finally: |
| ad_rx.messaging_droid.smsStopTrackingIncomingSmsMessage() |
| |
| return tel_result_wrapper |
| |
| |
| def is_mms_match(event, phonenumber_tx, text): |
| """Return True if 'text' equals to event['data']['Text'] |
| and phone number match. |
| |
| Args: |
| event: Event object to verify. |
| phonenumber_tx: phone number for sender. |
| text: text string to verify. |
| |
| Returns: |
| Return True if 'text' equals to event['data']['Text'] |
| and phone number match. |
| """ |
| #TODO: add mms matching after mms message parser is added in sl4a. b/34276948 |
| return True |
| |
| |
| def wait_for_matching_mms(log, |
| ad_rx, |
| phonenumber_tx, |
| text, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE): |
| """Wait for matching incoming SMS. |
| |
| Args: |
| log: Log object. |
| ad_rx: Receiver's Android Device Object |
| phonenumber_tx: Sender's phone number. |
| text: SMS content string. |
| allow_multi_part_long_sms: is long SMS allowed to be received as |
| multiple short SMS. This is optional, default value is True. |
| |
| Returns: |
| True if matching incoming SMS is received. |
| """ |
| try: |
| #TODO: add mms matching after mms message parser is added in sl4a. b/34276948 |
| ad_rx.messaging_ed.wait_for_event(EventMmsDownloaded, is_mms_match, |
| max_wait_time, phonenumber_tx, text) |
| ad_rx.log.info("Got event %s", EventMmsDownloaded) |
| return True |
| except Empty: |
| ad_rx.log.warning("No matched MMS downloaded event.") |
| return False |
| |
| |
| def mms_send_receive_verify(log, |
| ad_tx, |
| ad_rx, |
| array_message, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE, |
| expected_result=True, |
| slot_id_rx=None): |
| """Send MMS, receive MMS, and verify content and sender's number. |
| |
| Send (several) MMS from droid_tx to droid_rx. |
| Verify MMS is sent, delivered and received. |
| Verify received content and sender's number are correct. |
| |
| Args: |
| log: Log object. |
| ad_tx: Sender's Android Device Object |
| ad_rx: Receiver's Android Device Object |
| array_message: the array of message to send/receive |
| """ |
| subid_tx = get_outgoing_message_sub_id(ad_tx) |
| if slot_id_rx is None: |
| subid_rx = get_incoming_message_sub_id(ad_rx) |
| else: |
| subid_rx = get_subid_from_slot_index(log, ad_rx, slot_id_rx) |
| |
| result = mms_send_receive_verify_for_subscription( |
| log, ad_tx, ad_rx, subid_tx, subid_rx, array_message, max_wait_time) |
| if result != expected_result: |
| log_messaging_screen_shot(ad_tx, test_name="mms_tx") |
| log_messaging_screen_shot(ad_rx, test_name="mms_rx") |
| return result == expected_result |
| |
| |
| def sms_mms_send_logcat_check(ad, type, begin_time): |
| type = type.upper() |
| log_results = ad.search_logcat( |
| "%s Message sent successfully" % type, begin_time=begin_time) |
| if log_results: |
| ad.log.info("Found %s sent successful log message: %s", type, |
| log_results[-1]["log_message"]) |
| return True |
| else: |
| log_results = ad.search_logcat( |
| "ProcessSentMessageAction: Done sending %s message" % type, |
| begin_time=begin_time) |
| if log_results: |
| for log_result in log_results: |
| if "status is SUCCEEDED" in log_result["log_message"]: |
| ad.log.info( |
| "Found BugleDataModel %s send succeed log message: %s", |
| type, log_result["log_message"]) |
| return True |
| return False |
| |
| |
| def sms_mms_receive_logcat_check(ad, type, begin_time): |
| type = type.upper() |
| smshandle_logs = ad.search_logcat( |
| "InboundSmsHandler: No broadcast sent on processing EVENT_BROADCAST_SMS", |
| begin_time=begin_time) |
| if smshandle_logs: |
| ad.log.warning("Found %s", smshandle_logs[-1]["log_message"]) |
| log_results = ad.search_logcat( |
| "New %s Received" % type, begin_time=begin_time) or \ |
| ad.search_logcat("New %s Downloaded" % type, begin_time=begin_time) |
| if log_results: |
| ad.log.info("Found SL4A %s received log message: %s", type, |
| log_results[-1]["log_message"]) |
| return True |
| else: |
| log_results = ad.search_logcat( |
| "Received %s message" % type, begin_time=begin_time) |
| if log_results: |
| ad.log.info("Found %s received log message: %s", type, |
| log_results[-1]["log_message"]) |
| log_results = ad.search_logcat( |
| "ProcessDownloadedMmsAction", begin_time=begin_time) |
| for log_result in log_results: |
| ad.log.info("Found %s", log_result["log_message"]) |
| if "status is SUCCEEDED" in log_result["log_message"]: |
| ad.log.info("Download succeed with ProcessDownloadedMmsAction") |
| return True |
| return False |
| |
| |
| #TODO: add mms matching after mms message parser is added in sl4a. b/34276948 |
| def mms_send_receive_verify_for_subscription( |
| log, |
| ad_tx, |
| ad_rx, |
| subid_tx, |
| subid_rx, |
| array_payload, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE): |
| """Send MMS, receive MMS, and verify content and sender's number. |
| |
| Send (several) MMS from droid_tx to droid_rx. |
| Verify MMS is sent, delivered and received. |
| Verify received content and sender's number are correct. |
| |
| Args: |
| log: Log object. |
| ad_tx: Sender's Android Device Object.. |
| ad_rx: Receiver's Android Device Object. |
| subid_tx: Sender's subscription ID to be used for SMS |
| subid_rx: Receiver's subscription ID to be used for SMS |
| array_message: the array of message to send/receive |
| """ |
| |
| phonenumber_tx = ad_tx.telephony['subscription'][subid_tx]['phone_num'] |
| phonenumber_rx = ad_rx.telephony['subscription'][subid_rx]['phone_num'] |
| toggle_enforce = False |
| |
| for ad in (ad_tx, ad_rx): |
| if "Permissive" not in ad.adb.shell("su root getenforce"): |
| ad.adb.shell("su root setenforce 0") |
| toggle_enforce = True |
| if not getattr(ad, "messaging_droid", None): |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| else: |
| try: |
| if not ad.messaging_droid.is_live: |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| else: |
| ad.messaging_ed.clear_all_events() |
| ad.messaging_droid.logI( |
| "Start mms_send_receive_verify_for_subscription test") |
| except Exception: |
| ad.log.info("Create new sl4a session for messaging") |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| |
| for subject, message, filename in array_payload: |
| ad_tx.messaging_ed.clear_events(EventMmsSentSuccess) |
| ad_tx.messaging_ed.clear_events(EventMmsSentFailure) |
| ad_rx.messaging_ed.clear_events(EventMmsDownloaded) |
| ad_rx.messaging_droid.smsStartTrackingIncomingMmsMessage() |
| ad_tx.log.info( |
| "Sending MMS from %s to %s, subject: %s, message: %s, file: %s.", |
| phonenumber_tx, phonenumber_rx, subject, message, filename) |
| try: |
| ad_tx.messaging_droid.smsSendMultimediaMessage( |
| phonenumber_rx, subject, message, phonenumber_tx, filename) |
| try: |
| events = ad_tx.messaging_ed.pop_events( |
| "(%s|%s)" % (EventMmsSentSuccess, |
| EventMmsSentFailure), max_wait_time) |
| for event in events: |
| ad_tx.log.info("Got event %s", event["name"]) |
| if event["name"] == EventMmsSentFailure: |
| if event.get("data") and event["data"].get("Reason"): |
| ad_tx.log.error("%s with reason: %s", |
| event["name"], |
| event["data"]["Reason"]) |
| return False |
| elif event["name"] == EventMmsSentSuccess: |
| break |
| except Empty: |
| ad_tx.log.warning("No %s or %s event.", EventMmsSentSuccess, |
| EventMmsSentFailure) |
| return False |
| |
| if not wait_for_matching_mms(log, ad_rx, phonenumber_tx, |
| message, max_wait_time): |
| return False |
| except Exception as e: |
| log.error("Exception error %s", e) |
| raise |
| finally: |
| ad_rx.messaging_droid.smsStopTrackingIncomingMmsMessage() |
| for ad in (ad_tx, ad_rx): |
| if toggle_enforce: |
| ad.send_keycode("BACK") |
| ad.adb.shell("su root setenforce 1") |
| return True |
| |
| |
| def mms_receive_verify_after_call_hangup( |
| log, ad_tx, ad_rx, array_message, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE): |
| """Verify the suspanded MMS during call will send out after call release. |
| |
| Hangup call from droid_tx to droid_rx. |
| Verify MMS is sent, delivered and received. |
| Verify received content and sender's number are correct. |
| |
| Args: |
| log: Log object. |
| ad_tx: Sender's Android Device Object |
| ad_rx: Receiver's Android Device Object |
| array_message: the array of message to send/receive |
| """ |
| return mms_receive_verify_after_call_hangup_for_subscription( |
| log, ad_tx, ad_rx, get_outgoing_message_sub_id(ad_tx), |
| get_incoming_message_sub_id(ad_rx), array_message, max_wait_time) |
| |
| |
| #TODO: add mms matching after mms message parser is added in sl4a. b/34276948 |
| def mms_receive_verify_after_call_hangup_for_subscription( |
| log, |
| ad_tx, |
| ad_rx, |
| subid_tx, |
| subid_rx, |
| array_payload, |
| max_wait_time=MAX_WAIT_TIME_SMS_RECEIVE): |
| """Verify the suspanded MMS during call will send out after call release. |
| |
| Hangup call from droid_tx to droid_rx. |
| Verify MMS is sent, delivered and received. |
| Verify received content and sender's number are correct. |
| |
| Args: |
| log: Log object. |
| ad_tx: Sender's Android Device Object.. |
| ad_rx: Receiver's Android Device Object. |
| subid_tx: Sender's subscription ID to be used for SMS |
| subid_rx: Receiver's subscription ID to be used for SMS |
| array_message: the array of message to send/receive |
| """ |
| |
| phonenumber_tx = ad_tx.telephony['subscription'][subid_tx]['phone_num'] |
| phonenumber_rx = ad_rx.telephony['subscription'][subid_rx]['phone_num'] |
| for ad in (ad_tx, ad_rx): |
| if not getattr(ad, "messaging_droid", None): |
| ad.messaging_droid, ad.messaging_ed = ad.get_droid() |
| ad.messaging_ed.start() |
| for subject, message, filename in array_payload: |
| ad_rx.log.info( |
| "Waiting MMS from %s to %s, subject: %s, message: %s, file: %s.", |
| phonenumber_tx, phonenumber_rx, subject, message, filename) |
| ad_rx.messaging_droid.smsStartTrackingIncomingMmsMessage() |
| time.sleep(5) |
| try: |
| hangup_call(log, ad_tx) |
| hangup_call(log, ad_rx) |
| try: |
| ad_tx.messaging_ed.pop_event(EventMmsSentSuccess, |
| max_wait_time) |
| ad_tx.log.info("Got event %s", EventMmsSentSuccess) |
| except Empty: |
| log.warning("No sent_success event.") |
| if not wait_for_matching_mms(log, ad_rx, phonenumber_tx, message): |
| return False |
| finally: |
| ad_rx.messaging_droid.smsStopTrackingIncomingMmsMessage() |
| return True |
| |
| |
| def log_messaging_screen_shot(ad, test_name=""): |
| ad.ensure_screen_on() |
| ad.send_keycode("HOME") |
| ad.adb.shell("am start -n com.google.android.apps.messaging/.ui." |
| "ConversationListActivity") |
| time.sleep(3) |
| ad.screenshot(test_name) |
| ad.adb.shell("am start -n com.google.android.apps.messaging/com.google." |
| "android.apps.messaging.ui.conversation." |
| "LaunchConversationShimActivity -e conversation_id 1") |
| time.sleep(3) |
| ad.screenshot(test_name) |
| ad.send_keycode("HOME") |
