# -*- 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") try: test.compare(verify_status(get_error_message_from_log("521")),ERROR_MESSAGE["521"], "message is not in unhandled.conf file. verified error message from .err file") except: test.log("CS Out buffer error not logged ") 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 rename_file(path,filename): """ This method Rename the original folder name to dummy name. """ folder_path = os.listdir(path) for files in folder_path: if files == filename+"_1": if os.path.exists(path+filename+"_1"): shutil.rmtree(path+filename+"_1") test.log(str(filename+" name changed to "+files)) if files == filename: if os.path.exists(path+filename): utils.waitForGUI(1) os.rename(path+filename,path+filename+"_1") test.log(str(files+" name changed to "+filename+"_1")) def rename_old_name(path,filename): """ This method Rename the dummy name to original folder name . """ folder_path = os.listdir(path) for files in folder_path: if files == filename: if os.path.exists(path+filename): shutil.rmtree(path+filename) test.log(str(files+"_1 name changed to "+filename)) utils.waitForGUI(5) if files != filename: if (files == filename+"_1"): os.rename(path+files,path+filename) test.log(str(files+" name changed to "+filename)) 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: rename_file(config.SD_CARD_CLOUDSYNC_FOLDER_PATH,"cloudsync") utils.waitForGUI(5) except: test.log(str("cloudsync main desktop folder name not changed")) finally: rename_old_name(config.SD_CARD_CLOUDSYNC_FOLDER_PATH,"cloudsync") try: rename_file(config.SD_CARD_LOCATION,"sd-card") utils.waitForGUI(5) except: test.log(str("sd-card folder name not changed")) finally: rename_old_name(config.SD_CARD_LOCATION,"sd-card") try: rename_file(config.USB_DISK_FOLDER_LOCATION,"usb-disk") utils.waitForGUI(5) except: test.log(str("usb_disk folder name not changed")) finally: rename_old_name(config.USB_DISK_FOLDER_LOCATION,"usb-disk") try: rename_file(config.APPLICATION_FOLDER_LOCATION,"application") utils.waitForGUI(5) verify_wifi_scan() except: test.log(str("Setings wifi screen not visible")) finally: rename_old_name(config.APPLICATION_FOLDER_LOCATION,"application") try: 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: 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(): utils.tstStart(__file__) try: 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: rename_old_name(config.CLOUDSYNC_FOLDER_PATH,"cloudsync") verify_cloudsync_treatment() rename_folder_names() utils.tstDone()