| #!/usr/bin/env python3 |
| # |
| # Copyright 2022 - 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 copy |
| import re |
| import statistics |
| |
| from acts import signals |
| from acts_contrib.test_utils.tel.tel_defines import INVALID_SUB_ID |
| from acts_contrib.test_utils.tel.tel_subscription_utils import get_slot_index_from_data_sub_id |
| from acts_contrib.test_utils.tel.tel_subscription_utils import get_slot_index_from_voice_sub_id |
| from acts_contrib.test_utils.tel.tel_subscription_utils import get_subid_from_slot_index |
| |
| SETUP_DATA_CALL = 'SETUP_DATA_CALL' |
| SETUP_DATA_CALL_REQUEST = '> SETUP_DATA_CALL' |
| SETUP_DATA_CALL_RESPONSE = '< SETUP_DATA_CALL' |
| IS_CAPTIVEPORTAL = r'isCaptivePortal: isSuccessful()=true' |
| |
| DEACTIVATE_DATA_CALL = 'DEACTIVATE_DATA_CALL' |
| DEACTIVATE_DATA_CALL_REQUEST = '> DEACTIVATE_DATA_CALL' |
| DEACTIVATE_DATA_CALL_RESPONSE = '< DEACTIVATE_DATA_CALL' |
| UNSOL_DATA_CALL_LIST_CHANGED = 'UNSOL_DATA_CALL_LIST_CHANGED' |
| |
| IWLAN_DATA_SERVICE = 'IWlanDataService' |
| IWLAN_SETUP_DATA_CALL_REQUEST = '> REQUEST_SETUP_DATA_CALL' |
| IWLAN_SETUP_DATA_CALL_RESPONSE = 'setupDataCallResponse' |
| IWLAN_SEND_ACK = '> send ACK for serial' |
| |
| IWLAN_DEACTIVATE_DATA_CALL_REQUEST = '> REQUEST_DEACTIVATE_DATA_CALL' |
| IWLAN_DEACTIVATE_DATA_CALL_RESPONSE = 'deactivateDataCallResponse' |
| |
| SET_PREFERRED_DATA_MODEM = 'SET_PREFERRED_DATA_MODEM' |
| |
| WHI_IWLAN_DATA_SERVICE = 'IwlanDataService' |
| WHI_IWLAN_SETUP_DATA_CALL_REQUEST = r'IwlanDataService\[\d\]: Setup data call' |
| WHI_IWLAN_SETUP_DATA_CALL_RESPONSE = r'IwlanDataService\[\d\]: Tunnel opened!' |
| WHI_IWLAN_DEACTIVATE_DATA_CALL_REQUEST = r'IwlanDataService\[\d\]: Deactivate data call' |
| WHI_IWLAN_DEACTIVATE_DATA_CALL_RESPONSE = r'IwlanDataService\[\d\]: Tunnel closed!' |
| |
| ON_ENABLE_APN_IMS_SLOT0 = 'DCT-C-0 : onEnableApn: apnType=ims, request type=NORMAL' |
| ON_ENABLE_APN_IMS_SLOT1 = 'DCT-C-1 : onEnableApn: apnType=ims, request type=NORMAL' |
| ON_ENABLE_APN_IMS_HANDOVER_SLOT0 = 'DCT-C-0 : onEnableApn: apnType=ims, request type=HANDOVER' |
| ON_ENABLE_APN_IMS_HANDOVER_SLOT1 = 'DCT-C-1 : onEnableApn: apnType=ims, request type=HANDOVER' |
| RADIO_ON_4G_SLOT0 = r'GsmCdmaPhone: \[0\] Event EVENT_RADIO_ON Received' |
| RADIO_ON_4G_SLOT1 = r'GsmCdmaPhone: \[1\] Event EVENT_RADIO_ON Received' |
| RADIO_ON_IWLAN = 'Switching to new default network.*WIFI CONNECTED' |
| WIFI_OFF = 'setWifiEnabled.*enable=false' |
| ON_IMS_MM_TEL_CONNECTED_4G_SLOT0 = r'ImsPhone: \[0\].*onImsMmTelConnected imsRadioTech=WWAN' |
| ON_IMS_MM_TEL_CONNECTED_4G_SLOT1 = r'ImsPhone: \[1\].*onImsMmTelConnected imsRadioTech=WWAN' |
| ON_IMS_MM_TEL_CONNECTED_IWLAN_SLOT0 = r'ImsPhone: \[0\].*onImsMmTelConnected imsRadioTech=WLAN' |
| ON_IMS_MM_TEL_CONNECTED_IWLAN_SLOT1 = r'ImsPhone: \[1\].*onImsMmTelConnected imsRadioTech=WLAN' |
| |
| DEFAULT_MO_SMS_BODY = 'MO SMS body not yet found' |
| DEFAULT_MT_SMS_BODY = 'MT SMS body not yet found' |
| |
| MMS_SERVICE = 'MmsService:' |
| MMS_SEND_REQUEST_ID_PATTERN = r'SendRequest@(\S+)' |
| MMS_DOWNLOAD_REQUEST_ID_PATTERN = r'DownloadRequest@(\S+)' |
| MMS_START_NEW_NW_REQUEST = 'start new network request' |
| MMS_200_OK = '200 OK' |
| |
| SMS_SEND_TEXT_MESSAGE = 'smsSendTextMessage' |
| MO_SMS_LOGCAT_PATTERN = r'smsSendTextMessage.*"(\S+)", true|false' |
| SEND_SMS = 'SEND_SMS' |
| SEND_SMS_REQUEST = '> SEND_SMS' |
| SEND_SMS_RESPONSE = '< SEND_SMS' |
| SEND_SMS_EXPECT_MORE = 'SEND_SMS_EXPECT_MORE' |
| UNSOL_RESPONSE_NEW_SMS = '< UNSOL_RESPONSE_NEW_SMS' |
| SMS_RECEIVED = 'SmsReceived' |
| MT_SMS_CONTENT_PATTERN = 'sl4a.*?SmsReceived.*?"Text":"(.*?)"' |
| |
| SEND_SMS_OVER_IMS = r'ImsSmsDispatcher \[(\d)\]' |
| SEND_SMS_REQUEST_OVER_IMS = 'sendSms: mRetryCount' |
| SEND_SMS_RESPONSE_OVER_IMS = 'onSendSmsResult token' |
| SMS_RECEIVED_OVER_IMS = 'SMS received' |
| SMS_RECEIVED_OVER_IMS_SLOT0 = r'ImsSmsDispatcher \[0\]: SMS received' |
| SMS_RECEIVED_OVER_IMS_SLOT1 = r'ImsSmsDispatcher \[1\]: SMS received' |
| |
| IMS_REGISTERED_CST_SLOT0 = 'IMS_REGISTERED.*CrossStackEpdg.*SLID:0' |
| IMS_REGISTERED_CST_SLOT1 = 'IMS_REGISTERED.*CrossStackEpdg.*SLID:1' |
| ON_IMS_MM_TEL_CONNECTED_IWLAN_SLOT0 = r'ImsPhone: \[0\].*onImsMmTelConnected imsRadioTech=WLAN' |
| ON_IMS_MM_TEL_CONNECTED_IWLAN_SLOT1 = r'ImsPhone: \[1\].*onImsMmTelConnected imsRadioTech=WLAN' |
| |
| |
| def print_nested_dict(ad, d): |
| divider = "------" |
| for k, v in d.items(): |
| if isinstance(v, dict): |
| ad.log.info('%s %s %s', divider, k, divider) |
| print_nested_dict(ad, v) |
| else: |
| ad.log.info('%s: %s', k, v) |
| |
| |
| def get_slot_from_logcat(msg): |
| """Get slot index from specific pattern in logcat |
| |
| Args: |
| msg: logcat message string |
| |
| Returns: |
| 0 for pSIM or 1 for eSIM |
| """ |
| res = re.findall(r'\[(PHONE[\d])\]', msg) |
| try: |
| phone = res[0] |
| except: |
| phone = None |
| return phone |
| |
| |
| def get_apn_from_logcat(msg): |
| """Get APN from logcat |
| |
| Args: |
| msg: logcat message string |
| |
| Returns: |
| APN |
| """ |
| res = re.findall(r'DataProfile=[^/]+/[^/]+/[^/]+/([^/]+)/', msg) |
| try: |
| apn = res[0] |
| except: |
| apn = None |
| return apn |
| |
| |
| def parse_setup_data_call(ad, apn='internet', dds_switch=False): |
| """Search in logcat for lines containing data call setup procedure. |
| Calculate the data call setup time with given APN and validation |
| time on LTE. |
| |
| Args: |
| ad: Android object |
| apn: access point name |
| dds_switch: True for switching DDS. Otherwise False. |
| |
| Returns: |
| setup_data_call: Dictionary containing data call setup request and |
| response messages for each data call. The format is shown as |
| below: |
| { |
| message_id: |
| { |
| 'request': |
| { |
| 'message': logcat message body of data call setup |
| request message |
| 'time_stamp': time stamp in text format |
| 'datetime_obj': datetime object of time stamp |
| 'apn': access point name of this request |
| 'phone': 0 for pSIM or 1 for eSIM |
| } |
| 'response': |
| { |
| 'message': logcat message body of data call setup |
| response message |
| 'time_stamp': time stamp in text format |
| 'datetime_obj': datetime object of time stamp |
| 'cause': failure cause if data call setup failed |
| 'cid': PDP context ID |
| 'ifname': the name of the interface of the network |
| 'phone': 0 for pSIM or 1 for eSIM |
| 'unsol_data_call_list_changed': message of |
| unsol_data_call_list_changed |
| 'unsol_data_call_list_changed_time': time stamp of |
| the message unsol_data_call_list_changed |
| 'is_captive_portal': message of LTE validation pass |
| 'data_call_setup_time': time between data call setup |
| request and unsol_data_call_list_changed |
| 'validation_time_on_lte': time between data call |
| setup response and LTE validation pass |
| } |
| } |
| } |
| |
| data_call_setup_time_list: List. This is a summary of necessary |
| messages of data call setup procedure The format is shown as |
| below: |
| [ |
| { |
| 'request': logcat message body of data call setup |
| request message |
| 'response': logcat message body of data call setup |
| response message |
| 'unsol_data_call_list_changed': message of |
| unsol_data_call_list_changed |
| 'start': time stamp of data call setup request |
| 'end': time stamp of the message |
| unsol_data_call_list_changed |
| 'duration': time between data call setup request and |
| unsol_data_call_list_changed |
| 'validation_time_on_lte': time between data call |
| setup response and LTE validation pass |
| } |
| ] |
| |
| avg_data_call_setup_time: average of data call setup time |
| |
| avg_validation_time_on_lte: average of time for validation time on |
| LTE |
| """ |
| ad.log.info('====== Start to search logcat ====== ') |
| logcat = ad.search_logcat( |
| r'%s\|%s\|%s\|%s' % ( |
| SET_PREFERRED_DATA_MODEM, |
| SETUP_DATA_CALL, |
| UNSOL_DATA_CALL_LIST_CHANGED, IS_CAPTIVEPORTAL)) |
| |
| if not logcat: |
| return False |
| |
| for msg in logcat: |
| ad.log.info(msg["log_message"]) |
| |
| dds_slot = get_slot_index_from_data_sub_id(ad) |
| |
| set_preferred_data_modem = {} |
| setup_data_call = {} |
| data_call_setup_time_list = [] |
| last_message_id = None |
| |
| for line in logcat: |
| if line['message_id']: |
| if SET_PREFERRED_DATA_MODEM in line['log_message']: |
| set_preferred_data_modem['message'] = line['log_message'] |
| set_preferred_data_modem['time_stamp'] = line['time_stamp'] |
| set_preferred_data_modem[ |
| 'datetime_obj'] = line['datetime_obj'] |
| |
| if SETUP_DATA_CALL_REQUEST in line['log_message']: |
| found_apn = get_apn_from_logcat(line['log_message']) |
| if found_apn != apn: |
| continue |
| |
| phone = get_slot_from_logcat(line['log_message']) |
| if not phone: |
| continue |
| |
| if not dds_switch: |
| if str(dds_slot) not in phone: |
| continue |
| |
| msg_id = line['message_id'] |
| last_message_id = line['message_id'] |
| if msg_id not in setup_data_call: |
| setup_data_call[msg_id] = {} |
| |
| setup_data_call[msg_id]['request'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj'], |
| 'apn': found_apn, |
| 'phone': phone} |
| |
| if set_preferred_data_modem: |
| setup_data_call[msg_id]['request'][ |
| 'set_preferred_data_modem_message'] = set_preferred_data_modem['message'] |
| setup_data_call[msg_id]['request'][ |
| 'set_preferred_data_modem_time_stamp'] = set_preferred_data_modem['time_stamp'] |
| setup_data_call[msg_id]['request'][ |
| 'set_preferred_data_modem_datetime_obj'] = set_preferred_data_modem['datetime_obj'] |
| set_preferred_data_modem = {} |
| |
| if SETUP_DATA_CALL_RESPONSE in line['log_message']: |
| phone = get_slot_from_logcat(line['log_message']) |
| if not phone: |
| continue |
| |
| if not dds_switch: |
| if str(dds_slot) not in phone: |
| continue |
| |
| msg_id = line['message_id'] |
| if msg_id not in setup_data_call: |
| continue |
| |
| if 'request' not in setup_data_call[msg_id]: |
| continue |
| |
| last_message_id = line['message_id'] |
| |
| setup_data_call[msg_id]['response'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj'], |
| 'cause': '0', |
| 'cid': None, |
| 'ifname': None, |
| 'phone': phone, |
| 'unsol_data_call_list_changed': None, |
| 'unsol_data_call_list_changed_time': None, |
| 'is_captive_portal': None, |
| 'data_call_setup_time': None, |
| 'validation_time_on_lte': None} |
| |
| res = re.findall(r'cause=(\d+)', line['log_message']) |
| try: |
| cause = res[0] |
| setup_data_call[msg_id]['response']['cause'] = cause |
| except: |
| pass |
| |
| res = re.findall(r'cid=(\d+)', line['log_message']) |
| try: |
| cid = res[0] |
| setup_data_call[msg_id]['response']['cid'] = cid |
| except: |
| pass |
| |
| res = re.findall(r'ifname=(\S+)', line['log_message']) |
| try: |
| ifname = res[0] |
| setup_data_call[msg_id]['response']['ifname'] = ifname |
| except: |
| pass |
| |
| if UNSOL_DATA_CALL_LIST_CHANGED in line['log_message']: |
| if not last_message_id: |
| continue |
| |
| phone = get_slot_from_logcat(line['log_message']) |
| if not phone: |
| continue |
| |
| if not dds_switch: |
| if str(dds_slot) not in phone: |
| continue |
| |
| if 'request' not in setup_data_call[last_message_id]: |
| continue |
| |
| if 'response' not in setup_data_call[last_message_id]: |
| continue |
| |
| cid = setup_data_call[last_message_id]['response']['cid'] |
| if 'cid = %s' % cid not in line['log_message']: |
| continue |
| |
| if setup_data_call[last_message_id]['response']['cause'] != '0': |
| continue |
| |
| if dds_switch: |
| if 'set_preferred_data_modem_message' not in setup_data_call[ |
| last_message_id]['request']: |
| continue |
| data_call_start_time = setup_data_call[last_message_id][ |
| 'request']['set_preferred_data_modem_datetime_obj'] |
| |
| else: |
| data_call_start_time = setup_data_call[last_message_id][ |
| 'request']['datetime_obj'] |
| |
| data_call_end_time = line['datetime_obj'] |
| setup_data_call[last_message_id]['response'][ |
| 'unsol_data_call_list_changed_time'] = data_call_end_time |
| setup_data_call[last_message_id]['response'][ |
| 'unsol_data_call_list_changed'] = line['log_message'] |
| data_call_setup_time = data_call_end_time - data_call_start_time |
| setup_data_call[last_message_id]['response'][ |
| 'data_call_setup_time'] = data_call_setup_time.total_seconds() |
| |
| if apn == 'ims': |
| data_call_setup_time_list.append( |
| {'request': setup_data_call[ |
| last_message_id]['request']['message'], |
| 'response': setup_data_call[ |
| last_message_id]['response']['message'], |
| 'unsol_data_call_list_changed': setup_data_call[ |
| last_message_id]['response'][ |
| 'unsol_data_call_list_changed'], |
| 'start': data_call_start_time, |
| 'end': data_call_end_time, |
| 'duration': setup_data_call[last_message_id]['response'][ |
| 'data_call_setup_time']}) |
| |
| last_message_id = None |
| |
| if IS_CAPTIVEPORTAL in line['log_message']: |
| if not last_message_id: |
| continue |
| |
| if 'request' not in setup_data_call[last_message_id]: |
| continue |
| |
| if 'response' not in setup_data_call[last_message_id]: |
| continue |
| |
| if dds_switch: |
| data_call_start_time = setup_data_call[last_message_id][ |
| 'request']['set_preferred_data_modem_datetime_obj'] |
| |
| else: |
| data_call_start_time = setup_data_call[last_message_id][ |
| 'request']['datetime_obj'] |
| |
| setup_data_call[last_message_id]['response'][ |
| 'is_captive_portal'] = line['log_message'] |
| validation_start_time_on_lte = setup_data_call[ |
| last_message_id]['response']['datetime_obj'] |
| validation_end_time_on_lte = line['datetime_obj'] |
| validation_time_on_lte = ( |
| validation_end_time_on_lte - validation_start_time_on_lte).total_seconds() |
| setup_data_call[last_message_id]['response'][ |
| 'validation_time_on_lte'] = validation_time_on_lte |
| |
| data_call_setup_time_list.append( |
| {'request': setup_data_call[last_message_id]['request'][ |
| 'message'], |
| 'response': setup_data_call[last_message_id]['response'][ |
| 'message'], |
| 'unsol_data_call_list_changed': setup_data_call[ |
| last_message_id]['response']['unsol_data_call_list_changed'], |
| 'start': data_call_start_time, |
| 'end': setup_data_call[last_message_id]['response'][ |
| 'unsol_data_call_list_changed_time'], |
| 'duration': setup_data_call[last_message_id]['response'][ |
| 'data_call_setup_time'], |
| 'validation_time_on_lte': validation_time_on_lte}) |
| |
| last_message_id = None |
| |
| duration_list = [] |
| for item in data_call_setup_time_list: |
| if 'duration' in item: |
| duration_list.append(item['duration']) |
| |
| try: |
| avg_data_call_setup_time = statistics.mean(duration_list) |
| except: |
| avg_data_call_setup_time = None |
| |
| validation_time_on_lte_list = [] |
| for item in data_call_setup_time_list: |
| if 'validation_time_on_lte' in item: |
| validation_time_on_lte_list.append( |
| item['validation_time_on_lte']) |
| |
| try: |
| avg_validation_time_on_lte = statistics.mean( |
| validation_time_on_lte_list) |
| except: |
| avg_validation_time_on_lte = None |
| |
| return ( |
| setup_data_call, |
| data_call_setup_time_list, |
| avg_data_call_setup_time, |
| avg_validation_time_on_lte) |
| |
| |
| def parse_setup_data_call_on_iwlan(ad): |
| """Search in logcat for lines containing data call setup procedure. |
| Calculate the data call setup time with given APN on iwlan. |
| |
| Args: |
| ad: Android object |
| apn: access point name |
| |
| Returns: |
| setup_data_call: Dictionary containing data call setup request and |
| response messages for each data call. The format is shown as |
| below: |
| { |
| message_id: |
| { |
| 'request': |
| { |
| 'message': logcat message body of data call setup |
| request message |
| 'time_stamp': time stamp in text format |
| 'datetime_obj': datetime object of time stamp |
| } |
| 'response': |
| { |
| 'message': logcat message body of data call setup |
| response message |
| 'time_stamp': time stamp in text format |
| 'datetime_obj': datetime object of time stamp |
| 'cause': failure cause if data call setup failed |
| 'data_call_setup_time': time between data call setup |
| request and response |
| } |
| } |
| } |
| |
| data_call_setup_time_list: |
| List. This is a summary of mecessary messages of data call setup |
| procedure The format is shown as below: |
| [ |
| { |
| 'request': logcat message body of data call setup |
| request message |
| 'response': logcat message body of data call setup |
| response message |
| 'start': time stamp of data call setup request |
| 'end': time stamp of data call setup response |
| 'duration': time between data call setup request and |
| response |
| } |
| ] |
| |
| avg_data_call_setup_time: average of data call setup time |
| """ |
| ad.log.info('====== Start to search logcat ====== ') |
| logcat = ad.search_logcat(r'%s\|%s' % ( |
| IWLAN_DATA_SERVICE, WHI_IWLAN_DATA_SERVICE)) |
| |
| found_iwlan_data_service = 1 |
| if not logcat: |
| found_iwlan_data_service = 0 |
| |
| if not found_iwlan_data_service: |
| ( |
| setup_data_call, |
| data_call_setup_time_list, |
| avg_data_call_setup_time, |
| _) = parse_setup_data_call(ad, apn='ims') |
| |
| return ( |
| setup_data_call, |
| data_call_setup_time_list, |
| avg_data_call_setup_time) |
| |
| for msg in logcat: |
| ad.log.info(msg["log_message"]) |
| |
| setup_data_call = {} |
| data_call_setup_time_list = [] |
| last_message_id = None |
| |
| whi_msg_index = None |
| for line in logcat: |
| serial = None |
| cause = None |
| if IWLAN_SETUP_DATA_CALL_REQUEST in line['log_message']: |
| match_res = re.findall( |
| r'%s:\s(\d+)' % IWLAN_DATA_SERVICE, line['log_message']) |
| if match_res: |
| try: |
| serial = match_res[0] |
| except: |
| pass |
| |
| if not serial: |
| continue |
| |
| msg_id = serial |
| last_message_id = serial |
| if msg_id not in setup_data_call: |
| setup_data_call[msg_id] = {} |
| |
| setup_data_call[msg_id]['request'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj']} |
| |
| else: |
| if re.search( |
| WHI_IWLAN_SETUP_DATA_CALL_REQUEST, line['log_message']): |
| if whi_msg_index is None: |
| whi_msg_index = 0 |
| else: |
| whi_msg_index = whi_msg_index + 1 |
| |
| if str(whi_msg_index) not in setup_data_call: |
| setup_data_call[str(whi_msg_index)] = {} |
| |
| setup_data_call[str(whi_msg_index)]['request'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj']} |
| |
| if IWLAN_SETUP_DATA_CALL_RESPONSE in line['log_message']: |
| match_res = re.findall(r'Serial = (\d+)', line['log_message']) |
| if match_res: |
| try: |
| serial = match_res[0] |
| except: |
| pass |
| |
| if serial: |
| msg_id = serial |
| else: |
| msg_id = last_message_id |
| |
| if msg_id not in setup_data_call: |
| continue |
| |
| if 'request' not in setup_data_call[msg_id]: |
| continue |
| |
| setup_data_call[msg_id]['response'] = { |
| 'message': None, |
| 'time_stamp': None, |
| 'datetime_obj': None, |
| 'cause': None, |
| 'data_call_setup_time': None} |
| |
| match_res = re.findall( |
| r'Fail Cause = (\d+)', line['log_message']) |
| if match_res: |
| try: |
| cause = match_res[0] |
| except: |
| cause = None |
| |
| if cause != '0': |
| continue |
| |
| setup_data_call[msg_id]['response']['message'] = line[ |
| 'log_message'] |
| setup_data_call[msg_id]['response']['time_stamp'] = line[ |
| 'time_stamp'] |
| setup_data_call[msg_id]['response']['datetime_obj'] = line[ |
| 'datetime_obj'] |
| setup_data_call[msg_id]['response']['cause'] = 0 |
| |
| data_call_start_time = setup_data_call[last_message_id][ |
| 'request']['datetime_obj'] |
| data_call_end_time = line['datetime_obj'] |
| data_call_setup_time = data_call_end_time - data_call_start_time |
| setup_data_call[last_message_id]['response'][ |
| 'data_call_setup_time'] = data_call_setup_time.total_seconds() |
| |
| data_call_setup_time_list.append( |
| {'request': setup_data_call[last_message_id]['request'][ |
| 'message'], |
| 'response': setup_data_call[last_message_id]['response'][ |
| 'message'], |
| 'start': setup_data_call[last_message_id]['request'][ |
| 'datetime_obj'], |
| 'end': setup_data_call[last_message_id]['response'][ |
| 'datetime_obj'], |
| 'duration': setup_data_call[last_message_id]['response'][ |
| 'data_call_setup_time']}) |
| |
| last_message_id = None |
| |
| else: |
| if re.search( |
| WHI_IWLAN_SETUP_DATA_CALL_RESPONSE, line['log_message']): |
| if whi_msg_index is None: |
| continue |
| |
| if 'response' in setup_data_call[str(whi_msg_index)]: |
| ad.log.error('Duplicated setup data call response is ' |
| 'found or the request message is lost.') |
| continue |
| |
| setup_data_call[str(whi_msg_index)]['response'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj'], |
| 'data_call_setup_time': None} |
| |
| data_call_start_time = setup_data_call[str(whi_msg_index)][ |
| 'request']['datetime_obj'] |
| data_call_end_time = line['datetime_obj'] |
| data_call_setup_time = data_call_end_time - data_call_start_time |
| setup_data_call[str(whi_msg_index)]['response'][ |
| 'data_call_setup_time'] = data_call_setup_time.total_seconds() |
| |
| data_call_setup_time_list.append( |
| {'request': setup_data_call[str(whi_msg_index)][ |
| 'request']['message'], |
| 'response': setup_data_call[str(whi_msg_index)][ |
| 'response']['message'], |
| 'start': setup_data_call[str(whi_msg_index)]['request'][ |
| 'datetime_obj'], |
| 'end': setup_data_call[str(whi_msg_index)]['response'][ |
| 'datetime_obj'], |
| 'duration': setup_data_call[str(whi_msg_index)][ |
| 'response']['data_call_setup_time']}) |
| |
| duration_list = [] |
| for item in data_call_setup_time_list: |
| if 'duration' in item: |
| duration_list.append(item['duration']) |
| |
| try: |
| avg_data_call_setup_time = statistics.mean(duration_list) |
| except: |
| avg_data_call_setup_time = None |
| |
| ad.log.warning('setup_data_call: %s', setup_data_call) |
| ad.log.warning('duration list: %s', duration_list) |
| ad.log.warning('avg_data_call_setup_time: %s', avg_data_call_setup_time) |
| |
| return ( |
| setup_data_call, |
| data_call_setup_time_list, |
| avg_data_call_setup_time) |
| |
| |
| def parse_deactivate_data_call(ad): |
| """Search in logcat for lines containing data call deactivation procedure. |
| Calculate the data call deactivation time on LTE. |
| |
| Args: |
| ad: Android object |
| |
| Returns: |
| deactivate_data_call: Dictionary containing data call deactivation |
| request and response messages for each data call. The format is |
| shown as below: |
| { |
| message_id: |
| { |
| 'request': |
| { |
| 'message': logcat message body of data call |
| deactivation request message |
| 'time_stamp': time stamp in text format |
| 'datetime_obj': datetime object of time stamp |
| 'cid': PDP context ID |
| 'phone': 0 for pSIM or 1 for eSIM |
| } |
| 'response': |
| { |
| 'message': logcat message body of data call |
| deactivation response message |
| 'time_stamp': time stamp in text format |
| 'datetime_obj': datetime object of time stamp |
| 'phone': 0 for pSIM or 1 for eSIM |
| 'unsol_data_call_list_changed': message of |
| unsol_data_call_list_changed |
| 'deactivate_data_call_time': time between data call |
| deactivation request and unsol_data_call_list_changed |
| } |
| } |
| } |
| |
| deactivate_data_call_time_list: List. This is a summary of necessary |
| messages of data call deactivation procedure The format is shown |
| as below: |
| [ |
| { |
| 'request': logcat message body of data call |
| deactivation request message |
| 'response': logcat message body of data call |
| deactivation response message |
| 'unsol_data_call_list_changed': message of |
| unsol_data_call_list_changed |
| 'start': time stamp of data call deactivation request |
| 'end': time stamp of the message |
| unsol_data_call_list_changed |
| 'duration': time between data call deactivation |
| request and unsol_data_call_list_changed |
| } |
| ] |
| |
| avg_deactivate_data_call_time: average of data call deactivation time |
| """ |
| ad.log.info('====== Start to search logcat ====== ') |
| logcat = ad.search_logcat( |
| r'%s\|%s' % (DEACTIVATE_DATA_CALL, UNSOL_DATA_CALL_LIST_CHANGED)) |
| if not logcat: |
| return False |
| |
| for msg in logcat: |
| ad.log.info(msg["log_message"]) |
| |
| dds_slot = get_slot_index_from_data_sub_id(ad) |
| |
| deactivate_data_call = {} |
| deactivate_data_call_time_list = [] |
| last_message_id = None |
| |
| for line in logcat: |
| if line['message_id']: |
| if DEACTIVATE_DATA_CALL_REQUEST in line['log_message']: |
| phone = get_slot_from_logcat(line['log_message']) |
| if not phone: |
| continue |
| |
| if str(dds_slot) not in phone: |
| continue |
| |
| msg_id = line['message_id'] |
| last_message_id = line['message_id'] |
| if msg_id not in deactivate_data_call: |
| deactivate_data_call[msg_id] = {} |
| |
| deactivate_data_call[msg_id]['request'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj'], |
| 'cid': None, |
| 'phone': dds_slot} |
| |
| res = re.findall(r'cid = (\d+)', line['log_message']) |
| try: |
| cid = res[0] |
| deactivate_data_call[msg_id]['request']['cid'] = cid |
| except: |
| pass |
| |
| if DEACTIVATE_DATA_CALL_RESPONSE in line['log_message']: |
| phone = get_slot_from_logcat(line['log_message']) |
| if not phone: |
| continue |
| |
| if str(dds_slot) not in phone: |
| continue |
| |
| msg_id = line['message_id'] |
| if msg_id not in deactivate_data_call: |
| continue |
| |
| if 'request' not in deactivate_data_call[msg_id]: |
| continue |
| |
| last_message_id = line['message_id'] |
| |
| deactivate_data_call[msg_id]['response'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj'], |
| 'phone': dds_slot, |
| 'unsol_data_call_list_changed': None, |
| 'deactivate_data_call_time': None} |
| |
| if UNSOL_DATA_CALL_LIST_CHANGED in line['log_message']: |
| if not last_message_id: |
| continue |
| |
| phone = get_slot_from_logcat(line['log_message']) |
| if not phone: |
| continue |
| |
| if str(dds_slot) not in phone: |
| continue |
| |
| if 'request' not in deactivate_data_call[last_message_id]: |
| continue |
| |
| if 'response' not in deactivate_data_call[last_message_id]: |
| continue |
| |
| cid = deactivate_data_call[last_message_id]['request']['cid'] |
| if 'cid = %s' % cid not in line['log_message']: |
| continue |
| |
| deactivate_data_call_start_time = deactivate_data_call[ |
| last_message_id]['request']['datetime_obj'] |
| deactivate_data_call_end_time = line['datetime_obj'] |
| deactivate_data_call[last_message_id]['response'][ |
| 'unsol_data_call_list_changed'] = line['log_message'] |
| deactivate_data_call_time = ( |
| deactivate_data_call_end_time - deactivate_data_call_start_time) |
| deactivate_data_call[last_message_id]['response'][ |
| 'deactivate_data_call_time'] = deactivate_data_call_time.total_seconds() |
| deactivate_data_call_time_list.append( |
| {'request': deactivate_data_call[last_message_id][ |
| 'request']['message'], |
| 'response': deactivate_data_call[last_message_id][ |
| 'response']['message'], |
| 'unsol_data_call_list_changed': deactivate_data_call[ |
| last_message_id]['response'][ |
| 'unsol_data_call_list_changed'], |
| 'start': deactivate_data_call_start_time, |
| 'end': deactivate_data_call_end_time, |
| 'duration': deactivate_data_call_time.total_seconds()}) |
| |
| last_message_id = None |
| |
| duration_list = [] |
| for item in deactivate_data_call_time_list: |
| if 'duration' in item: |
| duration_list.append(item['duration']) |
| |
| try: |
| avg_deactivate_data_call_time = statistics.mean(duration_list) |
| except: |
| avg_deactivate_data_call_time = None |
| |
| return ( |
| deactivate_data_call, |
| deactivate_data_call_time_list, |
| avg_deactivate_data_call_time) |
| |
| |
| def parse_deactivate_data_call_on_iwlan(ad): |
| """Search in logcat for lines containing data call deactivation procedure. |
| Calculate the data call deactivation time on iwlan. |
| |
| Args: |
| ad: Android object |
| |
| Returns: |
| deactivate_data_call: Dictionary containing data call deactivation |
| request and response messages for each data call. The format is |
| shown as below: |
| { |
| message_id: |
| { |
| 'request': |
| { |
| 'message': logcat message body of data call |
| deactivation request message |
| 'time_stamp': time stamp in text format |
| 'datetime_obj': datetime object of time stamp |
| } |
| 'response': |
| { |
| 'message': logcat message body of data call |
| deactivation response message |
| 'time_stamp': time stamp in text format |
| 'datetime_obj': datetime object of time stamp |
| 'send_ack_for_serial_time': time stamp of ACK |
| 'deactivate_data_call_time': time between data call |
| deactivation request and ACK |
| } |
| } |
| } |
| |
| deactivate_data_call_time_list: List. This is a summary of necessary |
| messages of data call deactivation procedure The format is shown |
| as below: |
| [ |
| { |
| 'request': logcat message body of data call |
| deactivation request message |
| 'response': logcat message body of data call |
| deactivation response message |
| 'start': time stamp of data call deactivation request |
| 'end': time stamp of the ACK |
| 'duration': time between data call deactivation |
| request and ACK |
| } |
| ] |
| |
| avg_deactivate_data_call_time: average of data call deactivation time |
| """ |
| ad.log.info('====== Start to search logcat ====== ') |
| logcat = ad.search_logcat(r'%s\|%s' % ( |
| IWLAN_DATA_SERVICE, WHI_IWLAN_DATA_SERVICE)) |
| |
| found_iwlan_data_service = 1 |
| if not logcat: |
| found_iwlan_data_service = 0 |
| |
| if not found_iwlan_data_service: |
| ( |
| deactivate_data_call, |
| deactivate_data_call_time_list, |
| avg_deactivate_data_call_time) = parse_deactivate_data_call(ad) |
| |
| return ( |
| deactivate_data_call, |
| deactivate_data_call_time_list, |
| avg_deactivate_data_call_time) |
| |
| for msg in logcat: |
| ad.log.info(msg["log_message"]) |
| |
| deactivate_data_call = {} |
| deactivate_data_call_time_list = [] |
| last_message_id = None |
| |
| whi_msg_index = None |
| for line in logcat: |
| serial = None |
| if IWLAN_DEACTIVATE_DATA_CALL_REQUEST in line['log_message']: |
| match_res = re.findall( |
| r'%s:\s(\d+)' % IWLAN_DATA_SERVICE, line['log_message']) |
| if match_res: |
| try: |
| serial = match_res[0] |
| except: |
| serial = None |
| |
| if not serial: |
| continue |
| |
| msg_id = serial |
| last_message_id = serial |
| if msg_id not in deactivate_data_call: |
| deactivate_data_call[msg_id] = {} |
| |
| deactivate_data_call[msg_id]['request'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj']} |
| else: |
| if re.search(WHI_IWLAN_DEACTIVATE_DATA_CALL_REQUEST, line[ |
| 'log_message']): |
| if whi_msg_index is None: |
| whi_msg_index = 0 |
| else: |
| whi_msg_index = whi_msg_index + 1 |
| |
| if str(whi_msg_index) not in deactivate_data_call: |
| deactivate_data_call[str(whi_msg_index)] = {} |
| |
| deactivate_data_call[str(whi_msg_index)]['request'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj']} |
| |
| if IWLAN_DEACTIVATE_DATA_CALL_RESPONSE in line['log_message']: |
| if 'response' not in deactivate_data_call[last_message_id]: |
| deactivate_data_call[msg_id]['response'] = {} |
| |
| deactivate_data_call[msg_id]['response'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj'], |
| 'send_ack_for_serial_time': None, |
| 'deactivate_data_call_time': None} |
| |
| else: |
| if re.search(WHI_IWLAN_DEACTIVATE_DATA_CALL_RESPONSE, line[ |
| 'log_message']): |
| if whi_msg_index is None: |
| continue |
| |
| if 'response' in deactivate_data_call[str(whi_msg_index)]: |
| ad.log.error('Duplicated deactivate data call response' |
| 'is found or the request message is lost.') |
| continue |
| |
| deactivate_data_call[str(whi_msg_index)]['response'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj'], |
| 'deactivate_data_call_time': None} |
| |
| deactivate_data_call_start_time = deactivate_data_call[ |
| str(whi_msg_index)]['request']['datetime_obj'] |
| deactivate_data_call_end_time = line['datetime_obj'] |
| deactivate_data_call_time = ( |
| deactivate_data_call_end_time - deactivate_data_call_start_time) |
| deactivate_data_call[str(whi_msg_index)]['response'][ |
| 'deactivate_data_call_time'] = deactivate_data_call_time.total_seconds() |
| deactivate_data_call_time_list.append( |
| {'request': deactivate_data_call[str(whi_msg_index)][ |
| 'request']['message'], |
| 'response': deactivate_data_call[str(whi_msg_index)][ |
| 'response']['message'], |
| 'start': deactivate_data_call_start_time, |
| 'end': deactivate_data_call_end_time, |
| 'duration': deactivate_data_call_time.total_seconds()}) |
| |
| if IWLAN_SEND_ACK in line['log_message']: |
| match_res = re.findall( |
| r'%s:\s(\d+)' % IWLAN_DATA_SERVICE, line['log_message']) |
| if match_res: |
| try: |
| serial = match_res[0] |
| except: |
| serial = None |
| |
| if not serial: |
| continue |
| |
| msg_id = serial |
| |
| if msg_id not in deactivate_data_call: |
| continue |
| |
| if 'response' not in deactivate_data_call[msg_id]: |
| continue |
| |
| deactivate_data_call[msg_id]['response'][ |
| 'send_ack_for_serial_time'] = line['datetime_obj'] |
| |
| deactivate_data_call_start_time = deactivate_data_call[msg_id][ |
| 'request']['datetime_obj'] |
| deactivate_data_call_end_time = line['datetime_obj'] |
| deactivate_data_call_time = ( |
| deactivate_data_call_end_time - deactivate_data_call_start_time) |
| deactivate_data_call[msg_id]['response'][ |
| 'deactivate_data_call_time'] = deactivate_data_call_time.total_seconds() |
| deactivate_data_call_time_list.append( |
| {'request': deactivate_data_call[msg_id]['request'][ |
| 'message'], |
| 'response': deactivate_data_call[msg_id]['response'][ |
| 'message'], |
| 'start': deactivate_data_call_start_time, |
| 'end': deactivate_data_call_end_time, |
| 'duration': deactivate_data_call_time.total_seconds()}) |
| |
| last_message_id = None |
| |
| duration_list = [] |
| for item in deactivate_data_call_time_list: |
| if 'duration' in item: |
| duration_list.append(item['duration']) |
| |
| try: |
| avg_deactivate_data_call_time = statistics.mean(duration_list) |
| except: |
| avg_deactivate_data_call_time = None |
| |
| return ( |
| deactivate_data_call, |
| deactivate_data_call_time_list, |
| avg_deactivate_data_call_time) |
| |
| |
| def parse_ims_reg( |
| ad, |
| search_intervals=None, |
| rat='4g', |
| reboot_or_apm='reboot', |
| slot=None): |
| """Search in logcat for lines containing messages about IMS registration. |
| |
| Args: |
| ad: Android object |
| search_intervals: List. Only lines with time stamp in given time |
| intervals will be parsed. |
| E.g., [(begin_time1, end_time1), (begin_time2, end_time2)] |
| Both begin_time and end_time should be datetime object. |
| rat: "4g" for IMS over LTE or "iwlan" for IMS over Wi-Fi |
| reboot_or_apm: specify the scenario "reboot" or "apm" |
| slot: 0 for pSIM and 1 for eSIM |
| |
| Returns: |
| (ims_reg, parsing_fail, avg_ims_reg_duration) |
| |
| ims_reg: List of dictionaries containing found lines for start and |
| end time stamps. Each dict represents a cycle of the test. |
| |
| [ |
| {'start': message on start time stamp, |
| 'end': message on end time stamp, |
| 'duration': time difference between start and end} |
| ] |
| parsing_fail: List of dictionaries containing the cycle number and |
| missing messages of each failed cycle |
| |
| [ |
| 'attempt': failed cycle number |
| 'missing_msg' missing messages which should be found |
| ] |
| avg_ims_reg_duration: average of the duration in ims_reg |
| |
| """ |
| if slot is None: |
| slot = get_slot_index_from_voice_sub_id(ad) |
| ad.log.info('Default voice slot: %s', slot) |
| else: |
| if get_subid_from_slot_index(ad.log, ad, slot) == INVALID_SUB_ID: |
| ad.log.error('Slot %s is invalid.', slot) |
| raise signals.TestFailure('Failed', |
| extras={'fail_reason': 'Slot %s is invalid.' % slot}) |
| |
| ad.log.info('Assigned slot: %s', slot) |
| |
| start_command = { |
| 'reboot': { |
| '0': {'4g': ON_ENABLE_APN_IMS_SLOT0, |
| 'iwlan': ON_ENABLE_APN_IMS_HANDOVER_SLOT0 + '\|' + ON_ENABLE_APN_IMS_SLOT0}, |
| '1': {'4g': ON_ENABLE_APN_IMS_SLOT1, |
| 'iwlan': ON_ENABLE_APN_IMS_HANDOVER_SLOT1 + '\|' + ON_ENABLE_APN_IMS_SLOT1} |
| }, |
| 'apm':{ |
| '0': {'4g': RADIO_ON_4G_SLOT0, 'iwlan': RADIO_ON_IWLAN}, |
| '1': {'4g': RADIO_ON_4G_SLOT1, 'iwlan': RADIO_ON_IWLAN} |
| }, |
| 'wifi_off':{ |
| '0': {'4g': WIFI_OFF, 'iwlan': WIFI_OFF}, |
| '1': {'4g': WIFI_OFF, 'iwlan': WIFI_OFF} |
| }, |
| } |
| |
| end_command = { |
| '0': {'4g': ON_IMS_MM_TEL_CONNECTED_4G_SLOT0, |
| 'iwlan': ON_IMS_MM_TEL_CONNECTED_IWLAN_SLOT0}, |
| '1': {'4g': ON_IMS_MM_TEL_CONNECTED_4G_SLOT1, |
| 'iwlan': ON_IMS_MM_TEL_CONNECTED_IWLAN_SLOT1} |
| } |
| |
| ad.log.info('====== Start to search logcat ======') |
| logcat = ad.search_logcat('%s\|%s' % ( |
| start_command[reboot_or_apm][str(slot)][rat], |
| end_command[str(slot)][rat])) |
| |
| if not logcat: |
| raise signals.TestFailure('Failed', |
| extras={'fail_reason': 'No line matching the given pattern can ' |
| 'be found in logcat.'}) |
| |
| for msg in logcat: |
| ad.log.info(msg["log_message"]) |
| |
| ims_reg = [] |
| ims_reg_duration_list = [] |
| parsing_fail = [] |
| |
| start_command['reboot'] = { |
| '0': {'4g': ON_ENABLE_APN_IMS_SLOT0, |
| 'iwlan': ON_ENABLE_APN_IMS_HANDOVER_SLOT0 + '|' + ON_ENABLE_APN_IMS_SLOT0}, |
| '1': {'4g': ON_ENABLE_APN_IMS_SLOT1, |
| 'iwlan': ON_ENABLE_APN_IMS_HANDOVER_SLOT1 + '|' + ON_ENABLE_APN_IMS_SLOT1} |
| } |
| |
| keyword_dict = { |
| 'start': start_command[reboot_or_apm][str(slot)][rat], |
| 'end': end_command[str(slot)][rat] |
| } |
| |
| for attempt, interval in enumerate(search_intervals): |
| if isinstance(interval, list): |
| try: |
| begin_time, end_time = interval |
| except Exception as e: |
| ad.log.error(e) |
| continue |
| |
| ad.log.info('Parsing begin time: %s', begin_time) |
| ad.log.info('Parsing end time: %s', end_time) |
| |
| temp_keyword_dict = copy.deepcopy(keyword_dict) |
| for line in logcat: |
| if begin_time and line['datetime_obj'] < begin_time: |
| continue |
| |
| if end_time and line['datetime_obj'] > end_time: |
| break |
| |
| for key in temp_keyword_dict: |
| if temp_keyword_dict[key] and not isinstance( |
| temp_keyword_dict[key], dict): |
| res = re.findall( |
| temp_keyword_dict[key], line['log_message']) |
| if res: |
| ad.log.info('Found: %s', line['log_message']) |
| temp_keyword_dict[key] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['datetime_obj']} |
| break |
| |
| for key in temp_keyword_dict: |
| if temp_keyword_dict[key] == keyword_dict[key]: |
| ad.log.error( |
| '"%s" is missing in cycle %s.', |
| keyword_dict[key], |
| attempt) |
| parsing_fail.append({ |
| 'attempt': attempt, |
| 'missing_msg': keyword_dict[key]}) |
| try: |
| ims_reg_duration = ( |
| temp_keyword_dict['end'][ |
| 'time_stamp'] - temp_keyword_dict[ |
| 'start'][ |
| 'time_stamp']).total_seconds() |
| ims_reg_duration_list.append(ims_reg_duration) |
| ims_reg.append({ |
| 'start': temp_keyword_dict['start'][ |
| 'message'], |
| 'end': temp_keyword_dict['end'][ |
| 'message'], |
| 'duration': ims_reg_duration}) |
| except Exception as e: |
| ad.log.error(e) |
| |
| try: |
| avg_ims_reg_duration = statistics.mean(ims_reg_duration_list) |
| except: |
| avg_ims_reg_duration = None |
| |
| return ims_reg, parsing_fail, avg_ims_reg_duration |
| |
| |
| def parse_mo_sms(logcat): |
| """Search in logcat for lines containing messages about SMS sending on |
| LTE. |
| |
| Args: |
| logcat: List containing lines of logcat |
| |
| Returns: |
| send_sms: Dictionary containing found lines for each SMS |
| request and response messages together with their time stamps. |
| { |
| 'message_id':{ |
| 'request':{ |
| 'message': logcat message body of SMS request |
| 'time_stamp': time stamp in text format |
| 'datetime_obj': datetime object of the time stamp |
| }, |
| 'response':{ |
| 'message': logcat message body of SMS response |
| 'time_stamp': time stamp in text format |
| 'datetime_obj': datetime object of the time stamp |
| 'sms_delivery_time': time between SMS request and |
| response |
| } |
| } |
| } |
| |
| summary: the format is listed below: |
| { |
| 'request': logcat message body of SMS request |
| 'response': logcat message body of SMS response |
| 'unsol_response_new_sms': unsolicited response message upon |
| SMS receiving on MT UE |
| 'sms_body': message body of SMS |
| 'mo_start': time stamp of MO SMS request message |
| 'mo_end': time stamp of MO SMS response message |
| 'mo_signal_duration': time between MO SMS request and response |
| 'delivery_time': time between MO SMS request and |
| unsol_response_new_sms on MT UE |
| } |
| |
| avg_setup_time: average of mo_signal_duration |
| """ |
| send_sms = {} |
| summary = [] |
| sms_body = DEFAULT_MO_SMS_BODY |
| msg_id = None |
| if not logcat: |
| return False |
| |
| for line in logcat: |
| res = re.findall(MO_SMS_LOGCAT_PATTERN, line['log_message']) |
| if res: |
| try: |
| sms_body = res[0] |
| except: |
| sms_body = 'Cannot find MO SMS body' |
| |
| if line['message_id']: |
| msg_id = line['message_id'] |
| if SEND_SMS_REQUEST in line[ |
| 'log_message'] and SEND_SMS_EXPECT_MORE not in line[ |
| 'log_message']: |
| if msg_id not in send_sms: |
| send_sms[msg_id] = {} |
| |
| send_sms[msg_id]['sms_body'] = sms_body |
| sms_body = DEFAULT_MO_SMS_BODY |
| send_sms[msg_id]['request'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj']} |
| |
| if SEND_SMS_RESPONSE in line[ |
| 'log_message'] and SEND_SMS_EXPECT_MORE not in line[ |
| 'log_message']: |
| if msg_id not in send_sms: |
| continue |
| |
| if 'request' not in send_sms[msg_id]: |
| continue |
| |
| if "error" in line['log_message']: |
| continue |
| |
| send_sms[msg_id]['response'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj'], |
| 'sms_delivery_time': None} |
| |
| mo_sms_start_time = send_sms[msg_id]['request'][ |
| 'datetime_obj'] |
| mo_sms_end_time = line['datetime_obj'] |
| sms_delivery_time = mo_sms_end_time - mo_sms_start_time |
| send_sms[msg_id]['response'][ |
| 'sms_delivery_time'] = sms_delivery_time.total_seconds() |
| summary.append( |
| {'request': send_sms[msg_id]['request']['message'], |
| 'response': send_sms[msg_id]['response']['message'], |
| 'unsol_response_new_sms': None, |
| 'sms_body': send_sms[msg_id]['sms_body'], |
| 'mo_start': mo_sms_start_time, |
| 'mo_end': mo_sms_end_time, |
| 'mo_signal_duration': sms_delivery_time.total_seconds(), |
| 'delivery_time': None}) |
| |
| duration_list = [] |
| for item in summary: |
| if 'mo_signal_duration' in item: |
| duration_list.append(item['mo_signal_duration']) |
| |
| try: |
| avg_setup_time = statistics.mean(duration_list) |
| except: |
| avg_setup_time = None |
| |
| return send_sms, summary, avg_setup_time |
| |
| |
| def parse_mo_sms_iwlan(logcat): |
| """Search in logcat for lines containing messages about SMS sending on |
| iwlan. |
| |
| Args: |
| logcat: List containing lines of logcat |
| |
| Returns: |
| send_sms: Dictionary containing found lines for each SMS |
| request and response messages together with their time stamps. |
| { |
| 'message_id':{ |
| 'request':{ |
| 'message': logcat message body of SMS request |
| 'time_stamp': time stamp in text format |
| 'datetime_obj': datetime object of the time stamp |
| }, |
| 'response':{ |
| 'message': logcat message body of SMS response |
| 'time_stamp': time stamp in text format |
| 'datetime_obj': datetime object of the time stamp |
| 'sms_delivery_time': time between SMS request and |
| response |
| } |
| } |
| } |
| |
| summary: List containing dictionaries for each SMS. The format is |
| listed below: |
| [ |
| { |
| 'request': logcat message body of SMS request |
| 'response': logcat message body of SMS response |
| 'sms_body': message body of SMS |
| 'mo_start': time stamp of MO SMS request message |
| 'mo_end': time stamp of MO SMS response message |
| 'mo_signal_duration': time between MO SMS request and |
| response |
| 'delivery_time': time between MO SMS request and |
| MT SMS received message |
| } |
| ] |
| |
| avg_setup_time: average of mo_signal_duration |
| """ |
| send_sms = {} |
| summary = [] |
| sms_body = DEFAULT_MO_SMS_BODY |
| msg_id = None |
| |
| if not logcat: |
| return False |
| |
| for line in logcat: |
| res = re.findall(MO_SMS_LOGCAT_PATTERN, line['log_message']) |
| if res: |
| try: |
| sms_body = res[0] |
| except: |
| sms_body = 'Cannot find MO SMS body' |
| |
| if SEND_SMS_REQUEST_OVER_IMS in line['log_message']: |
| if msg_id is None: |
| msg_id = '0' |
| else: |
| msg_id = str(int(msg_id) + 1) |
| |
| if msg_id not in send_sms: |
| send_sms[msg_id] = {} |
| |
| send_sms[msg_id]['sms_body'] = sms_body |
| sms_body = DEFAULT_MO_SMS_BODY |
| send_sms[msg_id]['request'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj']} |
| |
| if SEND_SMS_RESPONSE_OVER_IMS in line['log_message']: |
| |
| if msg_id not in send_sms: |
| continue |
| |
| if 'request' not in send_sms[msg_id]: |
| continue |
| |
| if "error" in line['log_message']: |
| continue |
| |
| send_sms[msg_id]['response'] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj'], |
| 'sms_delivery_time': None} |
| |
| mo_sms_start_time = send_sms[msg_id]['request'][ |
| 'datetime_obj'] |
| mo_sms_end_time = line['datetime_obj'] |
| sms_delivery_time = mo_sms_end_time - mo_sms_start_time |
| send_sms[msg_id]['response'][ |
| 'sms_delivery_time'] = sms_delivery_time.total_seconds() |
| summary.append( |
| {'request': send_sms[msg_id]['request']['message'], |
| 'response': send_sms[msg_id]['response']['message'], |
| 'unsol_response_new_sms': None, |
| 'sms_body': send_sms[msg_id]['sms_body'], |
| 'mo_start': mo_sms_start_time, |
| 'mo_end': mo_sms_end_time, |
| 'mo_signal_duration': sms_delivery_time.total_seconds(), |
| 'delivery_time': None}) |
| |
| duration_list = [] |
| for item in summary: |
| if 'mo_signal_duration' in item: |
| duration_list.append(item['mo_signal_duration']) |
| |
| try: |
| avg_setup_time = statistics.mean(duration_list) |
| except: |
| avg_setup_time = None |
| |
| return send_sms, summary, avg_setup_time |
| |
| |
| def parse_mt_sms(logcat): |
| """Search in logcat for lines containing messages about SMS receiving on |
| LTE. |
| |
| Args: |
| logcat: List containing lines of logcat |
| |
| Returns: |
| received_sms_list: List containing dictionaries for each received |
| SMS. The format is listed below: |
| [ |
| { |
| 'message': logcat message body of unsolicited response |
| message |
| 'sms_body': message body of SMS |
| 'time_stamp': time stamp of unsolicited response message in |
| text format |
| 'datetime_obj': datetime object of the time stamp |
| 'sms_delivery_time': time between SMS request and |
| response |
| } |
| ] |
| """ |
| received_sms_list = [] |
| if not logcat: |
| return False |
| |
| for line in logcat: |
| if UNSOL_RESPONSE_NEW_SMS in line['log_message']: |
| |
| # if received_sms_list: |
| # if received_sms_list[-1]['sms_body'] is None: |
| # del received_sms_list[-1] |
| |
| received_sms_list.append( |
| {'message': line['log_message'], |
| 'sms_body': DEFAULT_MT_SMS_BODY, |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj']}) |
| else: |
| res = re.findall(MT_SMS_CONTENT_PATTERN, line['log_message']) |
| |
| if res: |
| try: |
| sms_body = res[0] |
| except: |
| sms_body = 'Cannot find MT SMS body' |
| |
| if received_sms_list[-1]['sms_body'] == DEFAULT_MT_SMS_BODY: |
| received_sms_list[-1]['sms_body'] = sms_body |
| continue |
| |
| return received_sms_list |
| |
| |
| def parse_mt_sms_iwlan(logcat): |
| """Search in logcat for lines containing messages about SMS receiving on |
| iwlan. |
| |
| Args: |
| logcat: List containing lines of logcat |
| |
| Returns: |
| received_sms_list: List containing dictionaries for each received |
| SMS. The format is listed below: |
| [ |
| { |
| 'message': logcat message body of SMS received message |
| 'sms_body': message body of SMS |
| 'time_stamp': time stamp of SMS received message in |
| text format |
| 'datetime_obj': datetime object of the time stamp |
| } |
| ] |
| """ |
| received_sms_list = [] |
| if not logcat: |
| return False |
| |
| for line in logcat: |
| if re.findall( |
| SMS_RECEIVED_OVER_IMS_SLOT0 + '|' + SMS_RECEIVED_OVER_IMS_SLOT1, |
| line['log_message']): |
| received_sms_list.append( |
| {'message': line['log_message'], |
| 'sms_body': DEFAULT_MT_SMS_BODY, |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj']}) |
| else: |
| res = re.findall(MT_SMS_CONTENT_PATTERN, line['log_message']) |
| |
| if res: |
| try: |
| sms_body = res[0] |
| except: |
| sms_body = 'Cannot find MT SMS body' |
| |
| if received_sms_list[-1]['sms_body'] == DEFAULT_MT_SMS_BODY: |
| received_sms_list[-1]['sms_body'] = sms_body |
| continue |
| |
| return received_sms_list |
| |
| |
| def parse_sms_delivery_time(log, ad_mo, ad_mt, rat='4g'): |
| """Calculate the SMS delivery time (time between MO SMS request and MT |
| unsolicited response message or MT SMS received message) from logcat |
| of both MO and MT UE. |
| |
| Args: |
| ad_mo: MO Android object |
| ad_mt: MT Android object |
| rat: '4g' for LTE and 'iwlan' for iwlan |
| |
| Returns: |
| None |
| """ |
| ad_mo.log.info('====== Start to search logcat ====== ') |
| mo_logcat = ad_mo.search_logcat( |
| r'%s\|%s\|%s\|%s' % ( |
| SMS_SEND_TEXT_MESSAGE, |
| SEND_SMS, |
| SEND_SMS_REQUEST_OVER_IMS, |
| SEND_SMS_RESPONSE_OVER_IMS)) |
| ad_mt.log.info('====== Start to search logcat ====== ') |
| mt_logcat = ad_mt.search_logcat( |
| r'%s\|%s\|%s' % ( |
| UNSOL_RESPONSE_NEW_SMS, SMS_RECEIVED, SMS_RECEIVED_OVER_IMS)) |
| |
| for msg in mo_logcat: |
| ad_mo.log.info(msg["log_message"]) |
| for msg in mt_logcat: |
| ad_mt.log.info(msg["log_message"]) |
| |
| if rat == 'iwlan': |
| _, mo_sms_summary, avg = parse_mo_sms_iwlan(mo_logcat) |
| received_sms_list = parse_mt_sms_iwlan(mt_logcat) |
| else: |
| _, mo_sms_summary, avg = parse_mo_sms(mo_logcat) |
| received_sms_list = parse_mt_sms(mt_logcat) |
| |
| sms_delivery_time = [] |
| for mo_sms in mo_sms_summary: |
| for mt_sms in received_sms_list: |
| if mo_sms['sms_body'] == mt_sms['sms_body']: |
| mo_sms['delivery_time'] = ( |
| mt_sms['datetime_obj'] - mo_sms['mo_start']).total_seconds() |
| mo_sms['unsol_response_new_sms'] = mt_sms['message'] |
| sms_delivery_time.append(mo_sms['delivery_time']) |
| |
| try: |
| avg_sms_delivery_time = statistics.mean(sms_delivery_time) |
| except: |
| avg_sms_delivery_time = None |
| |
| ad_mo.log.info('====== MO SMS summary ======') |
| for item in mo_sms_summary: |
| ad_mo.log.info('------------------') |
| print_nested_dict(ad_mo, item) |
| ad_mt.log.info('====== Received SMS list ======') |
| for item in received_sms_list: |
| ad_mt.log.info('------------------') |
| print_nested_dict(ad_mt, item) |
| |
| ad_mo.log.info('%s SMS were actually sent.', len(mo_sms_summary)) |
| ad_mt.log.info('%s SMS were actually received.', len(received_sms_list)) |
| ad_mo.log.info('Average MO SMS setup time: %.2f sec.', avg) |
| log.info( |
| 'Average SMS delivery time: %.2f sec.', avg_sms_delivery_time) |
| |
| |
| def parse_mms(ad_mo, ad_mt): |
| """Search in logcat for lines containing messages about SMS sending and |
| receiving. Calculate MO & MT MMS setup time. |
| |
| Args: |
| ad_mo: MO Android object |
| ad_mt: MT Android object |
| |
| Returns: |
| send_mms: Dictionary containing each sent MMS. The format is shown |
| as below: |
| { |
| mms_msg_id: |
| { |
| MMS_START_NEW_NW_REQUEST: |
| { |
| 'time_stamp': time stamp of MMS request on MO UE in |
| text format |
| 'datetime_obj': datetime object of time stamp |
| }, |
| MMS_200_OK: |
| { |
| 'time_stamp': time stamp of '200 OK' for MMS request |
| in text format |
| 'datetime_obj': datetime object of time stamp |
| 'setup_time': MO MMS setup time. Time between MMS |
| request and 200 OK |
| } |
| } |
| |
| } |
| |
| mo_avg_setup_time: average of MO MMS setup time |
| |
| receive_mms： Dictionary containing each received MMS. The format is |
| shown as below: |
| { |
| mms_msg_id: |
| { |
| MMS_START_NEW_NW_REQUEST: |
| { |
| 'time_stamp': time stamp of MMS request on MT UE in |
| text format |
| 'datetime_obj': datetime object of time stamp |
| }, |
| MMS_200_OK: |
| { |
| 'time_stamp': time stamp of '200 OK' for MMS request |
| in text format |
| 'datetime_obj': datetime object of time stamp |
| 'setup_time': MT MMS setup time. Time between MMS |
| request and 200 OK |
| } |
| } |
| |
| } |
| |
| mt_avg_setup_time: average of MT MMS setup time |
| """ |
| send_mms = {} |
| receive_mms = {} |
| mo_setup_time_list = [] |
| mt_setup_time_list = [] |
| |
| ad_mo.log.info('====== Start to search logcat ====== ') |
| mo_logcat = ad_mo.search_logcat(MMS_SERVICE) |
| for msg in mo_logcat: |
| ad_mo.log.info(msg["log_message"]) |
| |
| ad_mt.log.info('====== Start to search logcat ====== ') |
| mt_logcat = ad_mt.search_logcat(MMS_SERVICE) |
| for msg in mt_logcat: |
| ad_mt.log.info(msg["log_message"]) |
| |
| if not mo_logcat or not mt_logcat: |
| return False |
| |
| for line in mo_logcat: |
| find_res = re.findall( |
| MMS_SEND_REQUEST_ID_PATTERN, line['log_message']) |
| |
| message_id = None |
| try: |
| message_id = find_res[0] |
| except: |
| pass |
| |
| if message_id: |
| mms_msg_id = message_id |
| if mms_msg_id not in send_mms: |
| send_mms[mms_msg_id] = {} |
| if MMS_START_NEW_NW_REQUEST in line['log_message']: |
| send_mms[mms_msg_id][MMS_START_NEW_NW_REQUEST] = { |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj']} |
| |
| if MMS_200_OK in line['log_message']: |
| send_mms[mms_msg_id][MMS_200_OK] = { |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj'], |
| 'setup_time': None} |
| |
| if MMS_START_NEW_NW_REQUEST in send_mms[mms_msg_id]: |
| setup_time = line['datetime_obj'] - send_mms[mms_msg_id][ |
| MMS_START_NEW_NW_REQUEST]['datetime_obj'] |
| send_mms[mms_msg_id][MMS_200_OK][ |
| 'setup_time'] = setup_time.total_seconds() |
| mo_setup_time_list.append(setup_time.total_seconds()) |
| |
| for line in mt_logcat: |
| find_res = re.findall( |
| MMS_DOWNLOAD_REQUEST_ID_PATTERN, line['log_message']) |
| |
| message_id = None |
| try: |
| message_id = find_res[0] |
| except: |
| pass |
| |
| if message_id: |
| mms_msg_id = message_id |
| if mms_msg_id not in receive_mms: |
| receive_mms[mms_msg_id] = {} |
| if MMS_START_NEW_NW_REQUEST in line['log_message']: |
| receive_mms[mms_msg_id][MMS_START_NEW_NW_REQUEST] = { |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj']} |
| |
| if MMS_200_OK in line['log_message']: |
| receive_mms[mms_msg_id][MMS_200_OK] = { |
| 'time_stamp': line['time_stamp'], |
| 'datetime_obj': line['datetime_obj'], |
| 'setup_time': None} |
| |
| if MMS_START_NEW_NW_REQUEST in receive_mms[mms_msg_id]: |
| setup_time = line['datetime_obj'] - receive_mms[ |
| mms_msg_id][MMS_START_NEW_NW_REQUEST]['datetime_obj'] |
| receive_mms[mms_msg_id][MMS_200_OK][ |
| 'setup_time'] = setup_time.total_seconds() |
| mt_setup_time_list.append(setup_time.total_seconds()) |
| |
| try: |
| mo_avg_setup_time = statistics.mean(mo_setup_time_list) |
| except: |
| mo_avg_setup_time = None |
| |
| try: |
| mt_avg_setup_time = statistics.mean(mt_setup_time_list) |
| except: |
| mt_avg_setup_time = None |
| |
| return send_mms, mo_avg_setup_time, receive_mms, mt_avg_setup_time |
| |
| |
| def parse_cst_reg(ad, slot, search_intervals=None): |
| """ Check if IMS CST and WFC is registered at given slot by parsing logcat. |
| |
| Args: |
| ad: Android object |
| slot: 0 for pSIM and 1 for eSIM |
| search_intervals: List. Only lines with time stamp in given time |
| intervals will be parsed. |
| E.g., [(begin_time1, end_time1), (begin_time2, end_time2)] |
| Both begin_time and end_time should be datetime object. |
| |
| Returns: List of attampt number and error messages of not found pattern |
| of failing cycles |
| """ |
| cst = { |
| 'ims_registered': { |
| '0': IMS_REGISTERED_CST_SLOT0, |
| '1': IMS_REGISTERED_CST_SLOT1 |
| }, |
| 'iwlan': { |
| '0': ON_IMS_MM_TEL_CONNECTED_IWLAN_SLOT0, |
| '1': ON_IMS_MM_TEL_CONNECTED_IWLAN_SLOT1 |
| } |
| } |
| ad.log.info('====== Start to search logcat ====== ') |
| logcat = ad.search_logcat( |
| '%s\|%s' % ( |
| cst['ims_registered'][str(slot)], cst['iwlan'][str(slot)])) |
| |
| for line in logcat: |
| msg = line["log_message"] |
| ad.log.info(msg) |
| |
| parsing_fail = [] |
| keyword_dict = { |
| 'ims_registered': cst['ims_registered'][str(slot)], |
| 'iwlan': cst['iwlan'][str(slot)] |
| } |
| for attempt, interval in enumerate(search_intervals): |
| if isinstance(interval, list): |
| try: |
| begin_time, end_time = interval |
| except Exception as e: |
| ad.log.error(e) |
| continue |
| |
| ad.log.info('Parsing begin time: %s', begin_time) |
| ad.log.info('Parsing end time: %s', end_time) |
| |
| temp_keyword_dict = copy.deepcopy(keyword_dict) |
| for line in logcat: |
| if begin_time and line['datetime_obj'] < begin_time: |
| continue |
| |
| if end_time and line['datetime_obj'] > end_time: |
| break |
| |
| for key in temp_keyword_dict: |
| if temp_keyword_dict[key] and not isinstance( |
| temp_keyword_dict[key], dict): |
| res = re.findall( |
| temp_keyword_dict[key], line['log_message']) |
| if res: |
| ad.log.info('Found: %s', line['log_message']) |
| temp_keyword_dict[key] = { |
| 'message': line['log_message'], |
| 'time_stamp': line['datetime_obj']} |
| break |
| |
| for key in temp_keyword_dict: |
| if temp_keyword_dict[key] == keyword_dict[key]: |
| ad.log.error( |
| '"%s" is missing in cycle %s.', |
| keyword_dict[key], |
| attempt+1) |
| parsing_fail.append({ |
| 'attempt': attempt+1, |
| 'missing_msg': keyword_dict[key]}) |
| |
| return parsing_fail |
| |
| |
| def check_ims_cst_reg(ad, slot, search_interval=None): |
| """ Check if IMS CST is registered at given slot by parsing logcat. |
| |
| Args: |
| ad: Android object |
| slot: 0 for pSIM and 1 for eSIM |
| search_intervals: List. Only lines with time stamp in given time |
| intervals will be parsed. |
| E.g., [(begin_time1, end_time1), (begin_time2, end_time2)] |
| Both begin_time and end_time should be datetime object. |
| |
| Returns: True for successful registration. Otherwise False |
| """ |
| ims_cst_reg = { |
| '0': IMS_REGISTERED_CST_SLOT0, |
| '1': IMS_REGISTERED_CST_SLOT1 |
| } |
| logcat = ad.search_logcat('%s' % ims_cst_reg[str(slot)]) |
| if isinstance(search_interval, list): |
| try: |
| begin_time, end_time = search_interval |
| except Exception as e: |
| ad.log.error(e) |
| |
| for line in logcat: |
| if begin_time and line['datetime_obj'] < begin_time: |
| continue |
| |
| if end_time and line['datetime_obj'] > end_time: |
| break |
| |
| res = re.findall(ims_cst_reg[str(slot)], line['log_message']) |
| if res: |
| ad.log.info( |
| 'IMS CST is registered due to following message ' |
| 'found: %s', line['log_message']) |
| return True |
| return False |
