# -*- 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 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 = { "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 ", "520" : "CS The provided Treatment Code is 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_TxCodeEmpty_520 = '1639391827,1,0,2008,1,' 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 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_error_message(): """ This method is used to verify error message log printing inside denaliSquish.error file. """ set_data_in_cloud_sync_output_file(eError_HeaderCount_501) utils.waitForGUI(2) test.compare(str(verify_status(get_error_message_from_log("501"))),ERROR_MESSAGE["501"], "message is not in unhandled.conf file. verified error message from .err 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 == 30: 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 main(): utility.append_cloudsync_credentials_file() utils.tstStart(__file__) startApplication(config.AUT_NAME) utils.waitForGUI(1.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) utils.tstDone()