# -*- 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() 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_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 main(): utility.append_cloudsync_credentials_file() utils.tstStart(__file__) startApplication(config.AUT_NAME) utils.waitForGUI(1.5) 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()