| #!/usr/bin/env python3 |
| # |
| # Copyright 2019 - The Android Open Source Project |
| # |
| # 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. |
| """ |
| Prerequisites: |
| Windows 10 |
| Bluetooth PTS installed |
| Recommended: Running cmder as Admin: https://cmder.net/ |
| |
| ### BEGIN SETUP STEPS### |
| 1. Install latest version of Python for windows: |
| https://www.python.org/downloads/windows/ |
| |
| Tested successfully on Python 3.7.3.: |
| https://www.python.org/ftp/python/3.7.3/python-3.7.3.exe |
| |
| 2. Launch Powershell and setup PATH: |
| Setx PATH “%PATH%;C:/Users/<username>/AppData/Local/Programs/Python/Python37-32/Scripts” |
| |
| 3. Launch Cmder as Admin before running any PTS related ACTS tests. |
| |
| |
| ### END SETUP STEPS### |
| |
| |
| Bluetooth PTS controller. |
| Mandatory parameters are log_directory and sig_root_directory. |
| |
| ACTS Config setup: |
| "BluetoothPtsDevice": { |
| "log_directory": "C:\\Users\\fsbtt\\Documents\\Profile Tuning Suite\\Test_Dir", |
| "sig_root_directory": "C:\\Program Files (x86)\\Bluetooth SIG" |
| } |
| |
| """ |
| from acts import signals |
| from datetime import datetime |
| from threading import Thread |
| |
| import ctypes |
| import logging |
| import os |
| import subprocess |
| import time |
| import xml.etree.ElementTree as ET |
| |
| from xml.dom import minidom |
| from xml.etree.ElementTree import Element |
| |
| |
| class BluetoothPtsDeviceConfigError(signals.ControllerError): |
| pass |
| |
| |
| class BluetoothPtsSnifferError(signals.ControllerError): |
| pass |
| |
| |
| MOBLY_CONTROLLER_CONFIG_NAME = "BluetoothPtsDevice" |
| ACTS_CONTROLLER_REFERENCE_NAME = "bluetooth_pts_device" |
| |
| # Prefix to identify final verdict string. This is a PTS specific log String. |
| VERDICT = 'VERDICT/' |
| |
| # Verdict strings that are specific to PTS. |
| VERDICT_STRINGS = { |
| 'RESULT_PASS': 'PASS', |
| 'RESULT_FAIL': 'FAIL', |
| 'RESULT_INCONC': 'INCONC', |
| 'RESULT_INCOMP': |
| 'INCOMP', # Initial final verdict meaning that test has not completed yet. |
| 'RESULT_NONE': |
| 'NONE', # Error verdict usually indicating internal PTS error. |
| } |
| |
| # Sniffer ready log message. |
| SNIFFER_READY = 'SNIFFER/Save and clear complete' |
| |
| # PTS Log Types as defined by PTS: |
| LOG_TYPE_GENERAL_TEXT = 0 |
| LOG_TYPE_FIRST = 1 |
| LOG_TYPE_START_TEST_CASE = 1 |
| LOG_TYPE_TEST_CASE_ENDED = 2 |
| LOG_TYPE_START_DEFAULT = 3 |
| LOG_TYPE_DEFAULT_ENDED = 4 |
| LOG_TYPE_FINAL_VERDICT = 5 |
| LOG_TYPE_PRELIMINARY_VERDICT = 6 |
| LOG_TYPE_TIMEOUT = 7 |
| LOG_TYPE_ASSIGNMENT = 8 |
| LOG_TYPE_START_TIMER = 9 |
| LOG_TYPE_STOP_TIMER = 10 |
| LOG_TYPE_CANCEL_TIMER = 11 |
| LOG_TYPE_READ_TIMER = 12 |
| LOG_TYPE_ATTACH = 13 |
| LOG_TYPE_IMPLICIT_SEND = 14 |
| LOG_TYPE_GOTO = 15 |
| LOG_TYPE_TIMED_OUT_TIMER = 16 |
| LOG_TYPE_ERROR = 17 |
| LOG_TYPE_CREATE = 18 |
| LOG_TYPE_DONE = 19 |
| LOG_TYPE_ACTIVATE = 20 |
| LOG_TYPE_MESSAGE = 21 |
| LOG_TYPE_LINE_MATCHED = 22 |
| LOG_TYPE_LINE_NOT_MATCHED = 23 |
| LOG_TYPE_SEND_EVENT = 24 |
| LOG_TYPE_RECEIVE_EVENT = 25 |
| LOG_TYPE_OTHERWISE_EVENT = 26 |
| LOG_TYPE_RECEIVED_ON_PCO = 27 |
| LOG_TYPE_MATCH_FAILED = 28 |
| LOG_TYPE_COORDINATION_MESSAGE = 29 |
| |
| PTS_DEVICE_EMPTY_CONFIG_MSG = "Configuration is empty, abort!" |
| |
| |
| def create(config): |
| if not config: |
| raise errors.PTS_DEVICE_EMPTY_CONFIG_MSG |
| return get_instance(config) |
| |
| |
| def destroy(pts): |
| try: |
| pts[0].clean_up() |
| except: |
| pts[0].log.error("Failed to clean up properly.") |
| |
| |
| def get_info(pts_devices): |
| """Get information from the BluetoothPtsDevice object. |
| |
| Args: |
| pts_devices: A list of BluetoothPtsDevice objects although only one |
| will ever be specified. |
| |
| Returns: |
| A dict, representing info for BluetoothPtsDevice object. |
| """ |
| return { |
| "address": pts_devices[0].address, |
| "sniffer_ready": pts_devices[0].sniffer_ready, |
| "ets_manager_library": pts_devices[0].ets_manager_library, |
| "log_directory": pts_devices[0].log_directory, |
| "pts_installation_directory": |
| pts_devices[0].pts_installation_directory, |
| } |
| |
| |
| def get_instance(config): |
| """Create BluetoothPtsDevice instance from a dictionary containing |
| information related to PTS. Namely the SIG root directory as |
| sig_root_directory and the log directory represented by the log_directory. |
| |
| Args: |
| config: A dict that contains BluetoothPtsDevice device info. |
| |
| Returns: |
| A list of BluetoothPtsDevice objects. |
| """ |
| result = [] |
| try: |
| log_directory = config.pop("log_directory") |
| except KeyError: |
| raise BluetoothPtsDeviceConfigError( |
| "Missing mandatory log_directory in config.") |
| try: |
| sig_root_directory = config.pop("sig_root_directory") |
| except KeyError: |
| example_path = \ |
| "C:\\\\Program Files (x86)\\\\Bluetooth SIG" |
| raise BluetoothPtsDeviceConfigError( |
| "Missing mandatory sig_root_directory in config. Example path: {}". |
| format(example_path)) |
| |
| # "C:\\Program Files (x86)\\Bluetooth SIG\\Bluetooth PTS\\bin\\ETSManager.dll" |
| ets_manager_library = "{}\\Bluetooth PTS\\bin\\ETSManager.dll".format( |
| sig_root_directory) |
| # "C:\\Program Files (x86)\\Bluetooth SIG\\Bluetooth PTS\\bin" |
| pts_installation_directory = "{}\\Bluetooth PTS\\bin".format( |
| sig_root_directory) |
| # "C:\\Program Files (x86)\\Bluetooth SIG\\Bluetooth Protocol Viewer" |
| pts_sniffer_directory = "{}\\Bluetooth Protocol Viewer".format( |
| sig_root_directory) |
| result.append( |
| BluetoothPtsDevice(ets_manager_library, log_directory, |
| pts_installation_directory, pts_sniffer_directory)) |
| return result |
| |
| |
| class BluetoothPtsDevice: |
| """Class representing an Bluetooth PTS device and associated functions. |
| |
| Each object of this class represents one BluetoothPtsDevice in ACTS. |
| """ |
| |
| _next_action = -1 |
| _observers = [] |
| address = "" |
| current_implicit_send_description = "" |
| devices = [] |
| extra_answers = [] |
| log_directory = "" |
| log = None |
| ics = None |
| ixit = None |
| profile_under_test = None |
| pts_library = None |
| pts_profile_mmi_request = "" |
| pts_test_result = VERDICT_STRINGS['RESULT_INCOMP'] |
| sniffer_ready = False |
| test_log_directory = "" |
| test_log_prefix = "" |
| |
| def __init__(self, ets_manager_library, log_directory, |
| pts_installation_directory, pts_sniffer_directory): |
| self.log = logging.getLogger() |
| if ets_manager_library is not None: |
| self.ets_manager_library = ets_manager_library |
| self.log_directory = log_directory |
| if pts_installation_directory is not None: |
| self.pts_installation_directory = pts_installation_directory |
| if pts_sniffer_directory is not None: |
| self.pts_sniffer_directory = pts_sniffer_directory |
| # Define callback functions |
| self.USEAUTOIMPLSENDFUNC = ctypes.CFUNCTYPE(ctypes.c_bool) |
| self.use_auto_impl_send_func = self.USEAUTOIMPLSENDFUNC( |
| self.UseAutoImplicitSend) |
| |
| self.DONGLE_MSG_FUNC = ctypes.CFUNCTYPE(ctypes.c_bool, ctypes.c_char_p) |
| self.dongle_msg_func = self.DONGLE_MSG_FUNC(self.DongleMsg) |
| |
| self.DEVICE_SEARCH_MSG_FUNC = ctypes.CFUNCTYPE(ctypes.c_bool, |
| ctypes.c_char_p, |
| ctypes.c_char_p, |
| ctypes.c_char_p) |
| self.dev_search_msg_func = self.DEVICE_SEARCH_MSG_FUNC( |
| self.DeviceSearchMsg) |
| |
| self.LOGFUNC = ctypes.CFUNCTYPE(ctypes.c_bool, ctypes.c_char_p, |
| ctypes.c_char_p, ctypes.c_char_p, |
| ctypes.c_int, ctypes.c_void_p) |
| self.log_func = self.LOGFUNC(self.Log) |
| |
| self.ONIMPLSENDFUNC = ctypes.CFUNCTYPE(ctypes.c_char_p, |
| ctypes.c_char_p, ctypes.c_int) |
| self.onimplsend_func = self.ONIMPLSENDFUNC(self.ImplicitSend) |
| |
| # Helps with PTS reliability. |
| os.chdir(self.pts_installation_directory) |
| # Load EtsManager |
| self.pts_library = ctypes.cdll.LoadLibrary(self.ets_manager_library) |
| self.log.info("ETS Manager library {0:s} has been loaded".format( |
| self.ets_manager_library)) |
| # If post-logging is turned on all callbacks to LPLOG-type function |
| # will be executed after test execution is complete. It is recommended |
| # that post-logging is turned on to avoid simultaneous invocations of |
| # LPLOG and LPAUTOIMPLICITSEND callbacks. |
| self.pts_library.SetPostLoggingEx(True) |
| |
| self.xml_root = Element("ARCHIVE") |
| version = Element("VERSION") |
| version.text = "2.0" |
| self.xml_root.append(version) |
| self.xml_pts_pixit = Element("PicsPixit") |
| self.xml_pts_pixit.text = "" |
| self.xml_pts_running_log = Element("LOG") |
| self.xml_pts_running_log.text = "" |
| self.xml_pts_running_summary = Element("SUMMARY") |
| self.xml_pts_running_summary.text = "" |
| |
| def clean_up(self): |
| # Since we have no insight to the actual PTS library, |
| # catch all Exceptions and log them. |
| try: |
| self.log.info("Cleaning up Stack...") |
| self.pts_library.ExitStackEx(self.profile_under_test) |
| except Exception as err: |
| self.log.error( |
| "Failed to clean up BluetoothPtsDevice: {}".format(err)) |
| try: |
| self.log.info("Unregistering Profile...") |
| self.pts_library.UnregisterProfileEx.argtypes = [ctypes.c_char_p] |
| self.pts_library.UnregisterProfileEx( |
| self.profile_under_test.encode()) |
| self.pts_library.UnRegisterGetDevInfoEx() |
| except Exception as err: |
| self.log.error( |
| "Failed to clean up BluetoothPtsDevice: {}".format(err)) |
| try: |
| self.log.info("Cleaning up Sniffer") |
| self.pts_library.SnifferTerminateEx() |
| except Exception as err: |
| self.log.error( |
| "Failed to clean up BluetoothPtsDevice: {}".format(err)) |
| self.log.info("Cleanup Done.") |
| |
| def write_xml_pts_pixit_values_for_current_test(self): |
| """ Writes the current PICS and IXIT values to the XML result. |
| """ |
| self.xml_pts_pixit.text = "ICS VALUES:\n\n" |
| for key, value in self.ics.items(): |
| self.xml_pts_pixit.text += "{} {}\n".format( |
| key.decode(), value.decode()) |
| self.xml_pts_pixit.text += "\nIXIT VALUES:\n\n" |
| for key, (_, value) in self.ixit.items(): |
| self.xml_pts_pixit.text += "{} {}\n".format( |
| key.decode(), value.decode()) |
| |
| def set_ics_and_ixit(self, ics, ixit): |
| self.ics = ics |
| self.ixit = ixit |
| |
| def set_profile_under_test(self, profile): |
| self.profile_under_test = profile |
| |
| def setup_pts(self): |
| """Prepares PTS to run tests. This needs to be called in test classes |
| after ICS, IXIT, and setting Profile under test. |
| Specifically BluetoothPtsDevice functions: |
| set_profile_under_test |
| set_ics_and_ixit |
| """ |
| |
| # Register layer to test with callbacks |
| self.pts_library.RegisterProfileWithCallbacks.argtypes = [ |
| ctypes.c_char_p, self.USEAUTOIMPLSENDFUNC, self.ONIMPLSENDFUNC, |
| self.LOGFUNC, self.DEVICE_SEARCH_MSG_FUNC, self.DONGLE_MSG_FUNC |
| ] |
| res = self.pts_library.RegisterProfileWithCallbacks( |
| self.profile_under_test.encode(), self.use_auto_impl_send_func, |
| self.onimplsend_func, self.log_func, self.dev_search_msg_func, |
| self.dongle_msg_func) |
| |
| self.log.info( |
| "Profile has been registered with result {0:d}".format(res)) |
| |
| # GetDeviceInfo module is for discovering devices and PTS Dongle address |
| # Initialize GetDeviceInfo and register it with callbacks |
| # First parameter is PTS executable directory |
| self.pts_library.InitGetDevInfoWithCallbacks.argtypes = [ |
| ctypes.c_char_p, self.DEVICE_SEARCH_MSG_FUNC, self.DONGLE_MSG_FUNC |
| ] |
| res = self.pts_library.InitGetDevInfoWithCallbacks( |
| self.pts_installation_directory.encode(), self.dev_search_msg_func, |
| self.dongle_msg_func) |
| self.log.info( |
| "GetDevInfo has been initialized with result {0:d}".format(res)) |
| # Initialize PTS dongle |
| res = self.pts_library.VerifyDongleEx() |
| self.log.info( |
| "PTS dongle has been initialized with result {0:d}".format(res)) |
| |
| # Find PTS dongle address |
| self.pts_library.GetDongleBDAddress.restype = ctypes.c_ulonglong |
| self.address = self.pts_library.GetDongleBDAddress() |
| self.address_str = "{0:012X}".format(self.address) |
| self.log.info("PTS BD Address 0x{0:s}".format(self.address_str)) |
| |
| # Initialize Bluetooth Protocol Viewer communication module |
| self.pts_library.SnifferInitializeEx() |
| |
| # If Bluetooth Protocol Viewer is not running, start it |
| if not self.is_sniffer_running(): |
| self.log.info("Starting Protocol Viewer") |
| args = [ |
| "{}\Executables\Core\FTS.exe".format( |
| self.pts_sniffer_directory), |
| '/PTS Protocol Viewer=Generic', |
| '/OEMTitle=Bluetooth Protocol Viewer', '/OEMKey=Virtual' |
| ] |
| subprocess.Popen(args) |
| sniffer_timeout = 10 |
| while not self.is_sniffer_running(): |
| time.sleep(sniffer_timeout) |
| |
| # Register to recieve Bluetooth Protocol Viewer notofications |
| self.pts_library.SnifferRegisterNotificationEx() |
| self.pts_library.SetParameterEx.argtypes = [ |
| ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p |
| ] |
| |
| for ics_name in self.ics: |
| res = self.pts_library.SetParameterEx( |
| ics_name, b'BOOLEAN', self.ics[ics_name], |
| self.profile_under_test.encode()) |
| if res: |
| self.log.info("ICS {0:s} set successfully".format( |
| str(ics_name))) |
| else: |
| self.log.error("Setting ICS {0:s} value failed".format( |
| str(ics_name))) |
| |
| for ixit_name in self.ixit: |
| res = self.pts_library.SetParameterEx( |
| ixit_name, (self.ixit[ixit_name])[0], |
| (self.ixit[ixit_name])[1], self.profile_under_test.encode()) |
| if res: |
| self.log.info("IXIT {0:s} set successfully".format( |
| str(ixit_name))) |
| else: |
| self.log.error("Setting IXIT {0:s} value failed".format( |
| str(ixit_name))) |
| |
| # Prepare directory to store Bluetooth Protocol Viewer output |
| if not os.path.exists(self.log_directory): |
| os.makedirs(self.log_directory) |
| |
| address_b = self.address_str.encode("utf-8") |
| self.pts_library.InitEtsEx.argtypes = [ |
| ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p |
| ] |
| |
| implicit_send_path = "{}\\implicit_send3.dll".format( |
| self.pts_installation_directory).encode() |
| res = self.pts_library.InitEtsEx(self.profile_under_test.encode(), |
| self.log_directory.encode(), |
| implicit_send_path, address_b) |
| self.log.info("ETS has been initialized with result {0:s}".format( |
| str(res))) |
| |
| # Initialize Host Stack DLL |
| self.pts_library.InitStackEx.argtypes = [ctypes.c_char_p] |
| res = self.pts_library.InitStackEx(self.profile_under_test.encode()) |
| self.log.info("Stack has been initialized with result {0:s}".format( |
| str(res))) |
| |
| # Select to receive Log messages after test is done |
| self.pts_library.SetPostLoggingEx.argtypes = [ |
| ctypes.c_bool, ctypes.c_char_p |
| ] |
| self.pts_library.SetPostLoggingEx(True, |
| self.profile_under_test.encode()) |
| |
| # Clear Bluetooth Protocol Viewer. Dongle message callback will update |
| # sniffer_ready automatically. No need to fail setup if the timeout |
| # is exceeded since the logs will still be available just not starting |
| # from a clean slate. Just post a warning. |
| self.sniffer_ready = False |
| self.pts_library.SnifferClearEx() |
| end_time = time.time() + 10 |
| while not self.sniffer_ready and time.time() < end_time: |
| time.sleep(1) |
| if not self.sniffer_ready: |
| self.log.warning("Sniffer not cleared. Continuing.") |
| |
| def is_sniffer_running(self): |
| """ Looks for running Bluetooth Protocol Viewer process |
| |
| Returns: |
| Returns True if finds one, False otherwise. |
| """ |
| prog = [ |
| line.split() |
| for line in subprocess.check_output("tasklist").splitlines() |
| ] |
| [prog.pop(e) for e in [0, 1, 2]] |
| for task in prog: |
| task_name = task[0].decode("utf-8") |
| if task_name == "Fts.exe": |
| self.log.info("Found FTS process successfully.") |
| # Sleep recommended by PTS. |
| time.sleep(1) |
| return True |
| return False |
| |
| def UseAutoImplicitSend(self): |
| """Callback method that defines Which ImplicitSend will be used. |
| |
| Returns: |
| True always to inform PTS to use the local implementation. |
| """ |
| return True |
| |
| def DongleMsg(self, msg_str): |
| """ Receives PTS dongle messages. |
| |
| Specifically this receives the Bluetooth Protocol Viewer completed |
| save/clear operations. |
| |
| Returns: |
| True if sniffer is ready, False otherwise. |
| """ |
| msg = (ctypes.c_char_p(msg_str).value).decode("utf-8") |
| self.log.info(msg) |
| # Sleep recommended by PTS. |
| time.sleep(1) |
| if SNIFFER_READY in msg: |
| self.sniffer_ready = True |
| return True |
| |
| def DeviceSearchMsg(self, addr_str, name_str, cod_str): |
| """ Receives device search messages |
| |
| Each device may return multiple messages |
| Each message will contain device address and may contain device name and |
| COD. |
| |
| Returns: |
| True always and reports to the callback appropriately. |
| """ |
| addr = (ctypes.c_char_p(addr_str).value).replace(b'\xed', |
| b' ').decode("utf-8") |
| name = (ctypes.c_char_p(name_str).value).replace(b'\xed', |
| b' ').decode("utf-8") |
| cod = (ctypes.c_char_p(cod_str).value).replace(b'\xed', |
| b' ').decode("utf-8") |
| self.devices.append( |
| "Device address = {0:s} name = {1:s} cod = {2:s}".format( |
| addr, name, cod)) |
| return True |
| |
| def Log(self, log_time_str, log_descr_str, log_msg_str, log_type, project): |
| """ Receives PTS log messages. |
| |
| Returns: |
| True always and reports to the callback appropriately. |
| """ |
| log_time = (ctypes.c_char_p(log_time_str).value).decode("utf-8") |
| log_descr = (ctypes.c_char_p(log_descr_str).value).decode("utf-8") |
| log_msg = (ctypes.c_char_p(log_msg_str).value).decode("utf-8") |
| if "Verdict Description" in log_descr: |
| self.xml_pts_running_summary.text += "\t- {}".format(log_msg) |
| if "Final Verdict" in log_descr: |
| self.xml_pts_running_summary.text += "{}{}\n".format( |
| log_descr.strip(), log_msg.strip()) |
| full_log_msg = "{}{}{}".format(log_time, log_descr, log_msg) |
| self.xml_pts_running_log.text += "{}\n".format(str(full_log_msg)) |
| |
| if ctypes.c_int(log_type).value == LOG_TYPE_FINAL_VERDICT: |
| indx = log_msg.find(VERDICT) |
| if indx == 0: |
| if self.pts_test_result == VERDICT_STRINGS['RESULT_INCOMP']: |
| if VERDICT_STRINGS['RESULT_INCONC'] in log_msg: |
| self.pts_test_result = VERDICT_STRINGS['RESULT_INCONC'] |
| elif VERDICT_STRINGS['RESULT_FAIL'] in log_msg: |
| self.pts_test_result = VERDICT_STRINGS['RESULT_FAIL'] |
| elif VERDICT_STRINGS['RESULT_PASS'] in log_msg: |
| self.pts_test_result = VERDICT_STRINGS['RESULT_PASS'] |
| elif VERDICT_STRINGS['RESULT_NONE'] in log_msg: |
| self.pts_test_result = VERDICT_STRINGS['RESULT_NONE'] |
| return True |
| |
| def ImplicitSend(self, description, style): |
| """ ImplicitSend callback |
| |
| Implicit Send Styles: |
| MMI_Style_Ok_Cancel1 = 0x11041, Simple prompt | OK, Cancel buttons | Default: OK |
| MMI_Style_Ok_Cancel2 = 0x11141, Simple prompt | Cancel button | Default: Cancel |
| MMI_Style_Ok1 = 0x11040, Simple prompt | OK button | Default: OK |
| MMI_Style_Yes_No1 = 0x11044, Simple prompt | Yes, No buttons | Default: Yes |
| MMI_Style_Yes_No_Cancel1 = 0x11043, Simple prompt | Yes, No buttons | Default: Yes |
| MMI_Style_Abort_Retry1 = 0x11042, Simple prompt | Abort, Retry buttons | Default: Abort |
| MMI_Style_Edit1 = 0x12040, Request for data input | OK, Cancel buttons | Default: OK |
| MMI_Style_Edit2 = 0x12140, Select item from a list | OK, Cancel buttons | Default: OK |
| |
| Handling |
| MMI_Style_Ok_Cancel1 |
| OK = return "OK" |
| Cancel = return 0 |
| |
| MMI_Style_Ok_Cancel2 |
| OK = return "OK" |
| Cancel = return 0 |
| |
| MMI_Style_Ok1 |
| OK = return "OK", this version should not return 0 |
| |
| MMI_Style_Yes_No1 |
| Yes = return "OK" |
| No = return 0 |
| |
| MMI_Style_Yes_No_Cancel1 |
| Yes = return "OK" |
| No = return 0 |
| Cancel = has been deprecated |
| |
| MMI_Style_Abort_Retry1 |
| Abort = return 0 |
| Retry = return "OK" |
| |
| MMI_Style_Edit1 |
| OK = return expected string |
| Cancel = return 0 |
| |
| MMI_Style_Edit2 |
| OK = return expected string |
| Cancel = return 0 |
| |
| Receives ImplicitSend messages |
| Description format is as following: |
| {MMI_ID,Test Name,Layer Name}MMI Action\n\nDescription: MMI Description |
| """ |
| descr_str = (ctypes.c_char_p(description).value).decode("utf-8") |
| # Sleep recommended by PTS. |
| time.sleep(1) |
| indx = descr_str.find('}') |
| implicit_send_info = descr_str[1:(indx)] |
| self.current_implicit_send_description = descr_str[(indx + 1):] |
| items = implicit_send_info.split(',') |
| implicit_send_info_id = items[0] |
| implicit_send_info_test_case = items[1] |
| self.pts_profile_mmi_request = items[2] |
| self.log.info( |
| "OnImplicitSend() has been called with the following parameters:\n" |
| ) |
| self.log.info("\t\tproject_name = {0:s}".format( |
| self.pts_profile_mmi_request)) |
| self.log.info("\t\tid = {0:s}".format(implicit_send_info_id)) |
| self.log.info( |
| "\t\ttest_case = {0:s}".format(implicit_send_info_test_case)) |
| self.log.info("\t\tdescription = {0:s}".format( |
| self.current_implicit_send_description)) |
| self.log.info("\t\tstyle = {0:#X}".format(ctypes.c_int(style).value)) |
| self.log.info("") |
| try: |
| self.next_action = int(implicit_send_info_id) |
| except Exception as err: |
| self.log.error( |
| "Setting verdict to RESULT_FAIL, exception found: {}".format( |
| err)) |
| self.pts_test_result = VERDICT_STRINGS['RESULT_FAIL'] |
| res = b'OK' |
| if len(self.extra_answers) > 0: |
| res = self.extra_answers.pop(0).encode() |
| self.log.info("Sending Response: {}".format(res)) |
| return res |
| |
| def log_results(self, test_name): |
| """Log results. |
| |
| Saves the sniffer results in cfa format and clears the sniffer. |
| |
| Args: |
| test_name: string, name of the test run. |
| """ |
| self.pts_library.SnifferCanSaveEx.restype = ctypes.c_bool |
| canSave = ctypes.c_bool(self.pts_library.SnifferCanSaveEx()).value |
| self.pts_library.SnifferCanSaveAndClearEx.restype = ctypes.c_bool |
| canSaveClear = ctypes.c_bool( |
| self.pts_library.SnifferCanSaveAndClearEx()).value |
| file_name = "\\{}.cfa".format(self.test_log_prefix).encode() |
| path = self.test_log_directory.encode() + file_name |
| |
| if canSave == True: |
| self.pts_library.SnifferSaveEx.argtypes = [ctypes.c_char_p] |
| self.pts_library.SnifferSaveEx(path) |
| else: |
| self.pts_library.SnifferSaveAndClearEx.argtypes = [ctypes.c_char_p] |
| self.pts_library.SnifferSaveAndClearEx(path) |
| end_time = time.time() + 60 |
| while self.sniffer_ready == False and end_time > time.time(): |
| self.log.info("Waiting for sniffer to be ready...") |
| time.sleep(1) |
| if self.sniffer_ready == False: |
| raise BluetoothPtsSnifferError( |
| "Sniffer not ready after 60 seconds.") |
| |
| def execute_test(self, test_name, test_timeout=60): |
| """Execute the input test name. |
| |
| Preps PTS to run the test and waits up to 2 minutes for all steps |
| in the execution to finish. Cleanup of PTS related objects follows |
| any test verdict. |
| |
| Args: |
| test_name: string, name of the test to execute. |
| """ |
| today = datetime.now() |
| self.write_xml_pts_pixit_values_for_current_test() |
| # TODO: Find out how to grab the PTS version. Temporarily |
| # hardcoded to v.7.4.1.2. |
| self.xml_pts_pixit.text = ( |
| "Test Case Started: {} v.7.4.1.2, {} started on {}\n\n{}".format( |
| self.profile_under_test, test_name, |
| today.strftime("%A, %B %d, %Y, %H:%M:%S"), |
| self.xml_pts_pixit.text)) |
| |
| self.xml_pts_running_summary.text += "Test case : {} started\n".format( |
| test_name) |
| log_time_formatted = "{:%Y_%m_%d_%H_%M_%S}".format(datetime.now()) |
| formatted_test_name = test_name.replace('/', '_') |
| formatted_test_name = formatted_test_name.replace('-', '_') |
| self.test_log_prefix = "{}_{}".format(formatted_test_name, |
| log_time_formatted) |
| self.test_log_directory = "{}\\{}\\{}".format(self.log_directory, |
| self.profile_under_test, |
| self.test_log_prefix) |
| os.makedirs(self.test_log_directory) |
| curr_test = test_name.encode() |
| |
| self.pts_library.StartTestCaseEx.argtypes = [ |
| ctypes.c_char_p, ctypes.c_char_p, ctypes.c_bool |
| ] |
| res = self.pts_library.StartTestCaseEx( |
| curr_test, self.profile_under_test.encode(), True) |
| self.log.info("Test has been started with result {0:s}".format( |
| str(res))) |
| |
| # Wait till verdict is received |
| self.log.info("Begin Test Execution... waiting for verdict.") |
| end_time = time.time() + test_timeout |
| while self.pts_test_result == VERDICT_STRINGS[ |
| 'RESULT_INCOMP'] and time.time() < end_time: |
| time.sleep(1) |
| self.log.info("End Test Execution... Verdict {}".format( |
| self.pts_test_result)) |
| |
| # Clean up after test is done |
| self.pts_library.TestCaseFinishedEx.argtypes = [ |
| ctypes.c_char_p, ctypes.c_char_p |
| ] |
| res = self.pts_library.TestCaseFinishedEx( |
| curr_test, self.profile_under_test.encode()) |
| |
| self.log_results(test_name) |
| self.xml_pts_running_summary.text += "{} finished\n".format(test_name) |
| # Add the log results to the XML output |
| self.xml_root.append(self.xml_pts_pixit) |
| self.xml_root.append(self.xml_pts_running_log) |
| self.xml_root.append(self.xml_pts_running_summary) |
| rough_string = ET.tostring(self.xml_root, |
| encoding='utf-8', |
| method='xml') |
| reparsed = minidom.parseString(rough_string) |
| with open( |
| "{}\\{}.xml".format(self.test_log_directory, |
| self.test_log_prefix), "w") as writter: |
| writter.write( |
| reparsed.toprettyxml(indent=" ", encoding="utf-8").decode()) |
| |
| if self.pts_test_result is VERDICT_STRINGS['RESULT_PASS']: |
| return True |
| return False |
| |
| """Observer functions""" |
| |
| def bind_to(self, callback): |
| """ Callbacks to add to the observer. |
| This is used for DUTS automatic responses (ImplicitSends local |
| implementation). |
| """ |
| self._observers.append(callback) |
| |
| @property |
| def next_action(self): |
| return self._next_action |
| |
| @next_action.setter |
| def next_action(self, action): |
| self._next_action = action |
| for callback in self._observers: |
| callback(self._next_action) |
| |
| """End Observer functions""" |
