# -*- coding: utf-8 -*- ## # Copyright (c) 2019-2020 Diality Inc. - All Rights Reserved. # copyright # THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, # IN PART OR IN WHOLE, # WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. # # file tst_cloud_sync-treatment_screen # date 2022/07/09 # author Amritha debnath # joseph varghese # # # NOTE: # This test verifies the cloud sync - treatment screen. import csv import names import shutil import os import time from dialin.ui import utils from configuration import utility from configuration import config from dialin.ui.hd_simulator import HDSimulator from dialin.ui.dg_simulator import DGSimulator from dialin.common.hd_defs import HDOpModes, PreTreatmentSubModes, PostTreatmentStates from dialin.common.hd_defs import HDOpModes, HDOpSubModes from dialin.utils.conversions import float_to_bytearray, integer_to_bytearray,unsigned_integer_to_bytearray dg_simulator = DGSimulator() hd_simulator = HDSimulator() ERROR_MESSAGE = { "500" : "CS Unknown Error ", "501" : "CS Incorrect header ", "502" : "CS Incorrect timestamp ", "503" : "CS Incorrect sequence ", "504" : "CS Incorrect CRC ", "505" : "CS Incorrect ID ", "506" : "CS Invalid ID ", "507" : "CS Incorrect parameter count ", "508" : "CS Mismatch parameter count ", "509" : "CS Missing parameter ", "510" : "CS No history available ", "511" : "CS Duplicate data ", "512" : "CS The log folder cannot be created. ", "513" : "CS Error writing to the input file. ", "514" : "CS The credentials file does not exist. ", "515" : "CS The credentials folder make failed. ", "516" : "CS The credentials file copy failed. ", "517" : "CS The credentials file remove failed. ", "518" : "CS The credentials folder is empty. ", "519" : "CS No Treatment Code provided. ", "520" : "CS The provided Treatment Code is empty. ", "521" : "CS Out buffer empty " } DEVICE_FACTORY_RESET_REQUEST_1 = '' eError_HeaderCount_501 = '1639391827,2999,3,501' eError_Timestamp_502 = '1639391ddvgvg827,1,0,2999,0,502' eError_Sequence_503 = '1639391827,dcdc1,0,2999,0,503' eError_Crc_504 = '1639391827,1,0dfd,2999,0,504' eError_MessageID_505 = '1639391827,1,0,299dfd9,0,505' eError_InvalidID_506 = '1639391827,1,0,20,0' eError_ParamCount_507 = '1639391827,1,0,2008,Tx_code' eError_ParamMismatch_508 = '1,1639391827,0,2999,0,508' eError_ParamMissing_509 = '1639391827,1,0,2999,3,509,[8,2]' eError_NoHistory_510 = '1639391827,1,0,2006,0' eError_CredentialMake_514 = '1657706417,7,0,2003,3,,DG1234567890123,06300216' eError_TxCodeEmpty_520 = '1639391827,1,0,2008,1,' eError_OutFileEmpty_521 = '' POST_TREATMENT_INSTRUCTION_COUNT = 2 CODE_DISPLAY_REQUEST = '1639391827,1,0,2008,1,Tx_code' DEVICE_STATE_REQUEST = '1639391827,1,0,2006,0' DEVICE_STATE_RESPONSE = '1639391827,1,0,1006,2,3,0' DEVICE_CREDENTIALS_REQUEST = '1639391827,1,0,2004,0' DEVICE_CREDENTIALS_RESPONSE = '1639391827,1,0,1004,1,/tmp/credentials/' DEVICE_FACTORY_RESET_REQUEST = '1639391827,1,0,2005,0' DEVICE_FACTORY_RESET_CONFIRM = '1639391827,1,0,1005,1,0' DEVICE_INFO_REQUEST = '1639391827,1,9,2002,0' DG_VERSION_SERIAL_REQUEST = '1657706417,7,0,1001,3,,DG1234567890123,06300216' HD_VERSION_SERIAL_REQUEST = '1657706411,1,0,1001,3,HD1234567890123,,06300216' DEVICE_HD_INFO_RESPONSE = '1639391827,1,0,1002,3,HD1234567890123,,sw_version' DEVICE_DG_INFO_RESPONSE = '1639391827,1,0,1002,3,,DG1234567890123,sw_version' def settings_text_obj(text): names.o_settings_home_text_obj["text"] = text return names.o_settings_home_text_obj def verify_code_from_treatment_response(): """ This function is used for verify the code text from treatment log. """ hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_POST.value, sub_mode=PostTreatmentStates.HD_POST_TREATMENT_PATIENT_DISCONNECTION_STATE.value) for indicator in range(1, POST_TREATMENT_INSTRUCTION_COUNT): if indicator != POST_TREATMENT_INSTRUCTION_COUNT: mouseClick(waitForObject(names.o_patient_Disconnection_right_arrow)) utils.waitForGUI() mouseClick(names.o_patientDisconnectionConfirm_CONFIRM_Text) message_text = CODE_DISPLAY_REQUEST.split(',') utils.waitForGUI(1) #delay for fetching the effect text_code_label = str(waitForObjectExists(names.o_treatmentReviewConfirm_Code_Tx_code_Label).text) test.compare(message_text[config.PARAMETER_INDEX], text_code_label.split(': ')[1], "CS2UI message text for code 2008 should be verified") def set_data_in_cloud_sync_output_file(message): """ This function is capable to write message into cloud sync folder. Through this method file reader act as a handler and it will write custom messages into {current_date}_out.buf @input: (string) message - expected message to be write on out.buf file. """ cloud_out_log = utility.get_cloud_sync_output_file() with open(cloud_out_log, "w") as filereader: filereader.write(message) test.log("User written message - %s into out.buf file" %message) utils.waitForGUI(3) def verify_status(read_line): read_line = read_line.split('[') test.log(str(read_line[0])) return read_line[0] def verify_files_error_message(): """ This method is used to verify error message log printing inside denaliSquish.error file. """ set_data_in_cloud_sync_output_file(eError_NoHistory_510) utils.waitForGUI(2) test.compare(verify_status(get_error_message_from_log("510")),ERROR_MESSAGE["510"], "message is not in unhandled.conf file. verified error message from .err file") test.compare(str(verify_status(get_error_message_from_log("518"))),ERROR_MESSAGE["518"], "message is not in unhandled.conf file. verified error message from .err file") test.compare(verify_status(get_error_message_from_log("516")),ERROR_MESSAGE["516"], "message is not in unhandled.conf file. verified error message from .err file") set_data_in_cloud_sync_output_file(eError_HeaderCount_501) utils.waitForGUI(2) test.compare(verify_status(get_error_message_from_log("501")),ERROR_MESSAGE["501"], "message is not in unhandled.conf file. verified error message from .err file") def verify_error_message(): """ This method is used to verify error message log printing inside denaliSquish.error file. """ set_data_in_cloud_sync_output_file(eError_Timestamp_502) utils.waitForGUI(2) test.compare(verify_status(get_error_message_from_log("502")),ERROR_MESSAGE["502"], "message is not in unhandled.conf file. verified error message from .err file") set_data_in_cloud_sync_output_file(eError_Sequence_503) utils.waitForGUI(2) test.compare(verify_status(get_error_message_from_log("503")),ERROR_MESSAGE["503"], "message is not in unhandled.conf file. verified error message from .err file") set_data_in_cloud_sync_output_file(eError_Crc_504) utils.waitForGUI(2) test.compare(verify_status(get_error_message_from_log("504")),ERROR_MESSAGE["504"], "message is not in unhandled.conf file. verified error message from .err file") set_data_in_cloud_sync_output_file(eError_MessageID_505) utils.waitForGUI(2) test.compare(verify_status(get_error_message_from_log("505")),ERROR_MESSAGE["505"], "message is not in unhandled.conf file. verified error message from .err file") set_data_in_cloud_sync_output_file(eError_InvalidID_506) utils.waitForGUI(2) test.compare(verify_status(get_error_message_from_log("506")),ERROR_MESSAGE["506"], "message is not in unhandled.conf file. verified error message from .err file") set_data_in_cloud_sync_output_file(eError_ParamCount_507) utils.waitForGUI(2) test.compare(verify_status(get_error_message_from_log("507")),ERROR_MESSAGE["507"], "message is not in unhandled.conf file. verified error message from .err file") set_data_in_cloud_sync_output_file(eError_ParamMismatch_508) utils.waitForGUI(2) test.compare(verify_status(get_error_message_from_log("508")),ERROR_MESSAGE["508"], "message is not in unhandled.conf file. verified error message from .err file") set_data_in_cloud_sync_output_file(eError_ParamMissing_509) utils.waitForGUI(2) test.compare(verify_status(get_error_message_from_log("509")),ERROR_MESSAGE["509"], "message is not in unhandled.conf file. verified error message from .err file") set_data_in_cloud_sync_output_file(eError_NoHistory_510) utils.waitForGUI(2) test.compare(verify_status(get_error_message_from_log("510")),ERROR_MESSAGE["510"], "message is not in unhandled.conf file. verified error message from .err file") set_data_in_cloud_sync_output_file(eError_TxCodeEmpty_520) utils.waitForGUI(2) test.compare(verify_status(get_error_message_from_log("520")),ERROR_MESSAGE["520"], "message is not in unhandled.conf file. verified error message from .err file") def verify_log(api_content,error_message): """ This function is capable to verify the log data from sd-card @param api_content - (string) combination of message id, message text and arguments separated with '\n' @param message_extracted - (list) list contains extracted data (ack_req, ack_bak, message, message_id). """ test.log(str(api_content)) if api_content == error_message: return True else : return False def verify_log_response(actual_cloudsync_data, expected_cloudsync_data): """ This function is used for verifying the expected and actual cloud-sync log data. @input: (string) actual_cloudsync_data - cloud sync log from in device. @input: (string) expected_cloudsync_data - expected cloud sync log. """ test.startSection("Verification of cloud response for code %s" %actual_cloudsync_data[config.CODE_INDEX]) expected_cloudsync_data = expected_cloudsync_data.split(',') current_epoch_value = time.time() epoch_value_status = utility.get_epoch_value_consistancy(current_epoch_value, actual_cloudsync_data[config.EPOCH_INDEX]) test.verify(epoch_value_status, "epoch value should be in range and it is verified") for index in (config.CRC_INDEX, config.CODE_INDEX, config.PARAMETER_COUNT, config.PARAMETER_INDEX): test.compare(expected_cloudsync_data[index], actual_cloudsync_data[index], "cloud sync data %s is verified" %expected_cloudsync_data[index]) test.endSection() def get_error_message_from_log(error_num): """ This function is capable to verify the error logs from sd-card @param api_content - (string) combination of message id, message text and arguments separated with '\n' @param expected_api_argument - (list) expected arguments for API. """ utils.waitForGUI(5) count = 0 file_name = utility.get_extracted_error_file() try: with open(file_name, 'r') as csv_file: try: for row in reversed(list(csv.reader(csv_file))): if row[0].isalpha(): pass else: if count == 100: test.fail("handler unable to find message text from error file.") break row_length = sum(1 for values in row) try: if ("["+error_num) in row[1]: if row[1] != None: return row[1] except: pass count+=1 except: test.fail("application error log data is corrupted") except: test.fail("err log file is not created or log file is not created based on standard log naming format.") def verify_wifi_scan(): """ Method to navigate to "Settings" screen """ test.startSection("Navigating to 'Device Settings' screen") hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_STAN.value, sub_mode=HDOpModes.MODE_FAUL.value) utils.waitForGUI(1) mouseClick(waitForObjectExists(names.o_settings_Text)) device_settings_text = waitForObjectExists(settings_text_obj(config.DEVICE_SETTINGS_TEXT)) test.compare(device_settings_text.text, config.DEVICE_SETTINGS_TEXT, "{} screen is displayed".format(config.DEVICE_SETTINGS_TEXT)) utils.waitForGUI(0.2) mouseClick(waitForObjectExists(names.o_wifi_settings_main_button)) utils.waitForGUI(1) mouseClick(waitForObjectExists(names.o_scan_wifi_setting_button)) utils.waitForGUI(1) mouseClick(waitForObjectExists(names.o_wifi_settings_back_button)) utils.waitForGUI(1) test.endSection() def rename_folder_names(): """ This method is for renaming the folder name and wait application for verifying particular file. """ test.startSection("Renaming the cloudsync,sd-card, usb-disk, application folder name") try: utility.rename_file(config.SD_CARD_CLOUDSYNC_FOLDER_PATH,"cloudsync") utils.waitForGUI(5) except: test.log(str("cloudsync main desktop folder name not changed")) finally: utility.rename_old_name(config.SD_CARD_CLOUDSYNC_FOLDER_PATH,"cloudsync") try: utility.rename_file(config.SD_CARD_LOCATION,"sd-card") utils.waitForGUI(5) except: test.log(str("sd-card folder name not changed")) finally: utility.rename_old_name(config.SD_CARD_LOCATION,"sd-card") try: utility.rename_file(config.USB_DISK_FOLDER_LOCATION,"usb-disk") utils.waitForGUI(5) except: test.log(str("usb_disk folder name not changed")) finally: utility.rename_old_name(config.USB_DISK_FOLDER_LOCATION,"usb-disk") try: utility.rename_file(config.APPLICATION_FOLDER_LOCATION,"application") utils.waitForGUI(5) verify_wifi_scan() except: test.log(str("Setings wifi screen not visible")) finally: utility.rename_old_name(config.APPLICATION_FOLDER_LOCATION,"application") try: utility.rename_file(config.CANBUS_FOLDER_LOCATION,"canbus") HD_DG_VERSION_SERIAL_REQUEST = "1657706421,13,0,1001,3,HD1234567890123,DG1234567890123,06300216" hd_simulator.cmd_send_serial_hd_data(serial = HD_VERSION_SERIAL_REQUEST[5]) dg_simulator.cmd_send_serial_dg_data(serial = DG_VERSION_SERIAL_REQUEST[6]) utils.waitForGUI(5) verify_wifi_scan() except: test.log(str("Setings wifi screen not visible")) finally: utility.rename_old_name(config.CANBUS_FOLDER_LOCATION,"canbus") test.endSection() def verify_cloudsync_treatment(): """ This method is to verify cloudsync treatment log data. """ utility.append_cloudsync_credentials_file() utils.waitForGUI(5) verify_error_message() main_item = waitForObjectExists(names.o_mainItem_Item) expected_software_version = str(object.children(main_item)[5]) set_data_in_cloud_sync_output_file(CODE_DISPLAY_REQUEST) verify_code_from_treatment_response() set_data_in_cloud_sync_output_file(DEVICE_CREDENTIALS_REQUEST) device_credentials_response = utility.retrive_log_data() verify_log_response(device_credentials_response[0], DEVICE_CREDENTIALS_RESPONSE) set_data_in_cloud_sync_output_file(DEVICE_FACTORY_RESET_REQUEST) factory_reset_response = utility.retrive_log_data() verify_log_response(factory_reset_response[0], DEVICE_FACTORY_RESET_CONFIRM) HD_VERSION = HD_VERSION_SERIAL_REQUEST.split(",") hd_simulator.cmd_send_serial_hd_data(HD_VERSION[5]) set_data_in_cloud_sync_output_file(DEVICE_INFO_REQUEST) device_credentials_response = utility.retrive_log_data() verify_log_response(device_credentials_response[0], DEVICE_HD_INFO_RESPONSE) DG_VERSION = DG_VERSION_SERIAL_REQUEST.split(",") dg_simulator.cmd_send_serial_dg_data(DG_VERSION[6]) set_data_in_cloud_sync_output_file(DEVICE_INFO_REQUEST) device_credentials_response = utility.retrive_log_data() verify_log_response(device_credentials_response[0], DEVICE_DG_INFO_RESPONSE) hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_STAN.value, sub_mode=HDOpSubModes.STANDBY_WAIT_FOR_TREATMENT_STATE .value) set_data_in_cloud_sync_output_file(DEVICE_STATE_REQUEST) device_credentials_response = utility.retrive_log_data() verify_log_response(device_credentials_response[0], DEVICE_STATE_RESPONSE) def main(): """ Do not interrupt this script while it is executing. This script might create folder problems. Inside this script we are renaming names of following folders which are required for execution. 1)cloudsync present on desktop 2)cloudsync present inside sd-card 3)usb-disk 4)sd-card 5)application 6)canbus """ utils.tstStart(__file__) try: utility.rename_file(config.CLOUDSYNC_FOLDER_PATH,"cloudsync") utils.waitForGUI(0.5) startApplication(config.AUT_NAME) utils.waitForGUI(1.5) HD_DG_VERSION_SERIAL_REQUEST = "1657706421,13,0,1001,3,HD1234567890123,DG1234567890123,06300216" hd_simulator.cmd_send_serial_hd_data(serial = HD_VERSION_SERIAL_REQUEST[5]) dg_simulator.cmd_send_serial_dg_data(serial = DG_VERSION_SERIAL_REQUEST[6]) utils.waitForGUI(1.5) verify_files_error_message() except: test.log(str("Cloudsync desktop folder name not changed")) finally: utility.rename_old_name(config.CLOUDSYNC_FOLDER_PATH,"cloudsync") verify_cloudsync_treatment() rename_folder_names() utils.tstDone()