Index: shared/scripts/configuration/config.py =================================================================== diff -u -r26be96f1f86d5823902beed32660f12cdb87d881 -re620a9fd664c96a736c0d7ca2ffe03d98a913189 --- shared/scripts/configuration/config.py (.../config.py) (revision 26be96f1f86d5823902beed32660f12cdb87d881) +++ shared/scripts/configuration/config.py (.../config.py) (revision e620a9fd664c96a736c0d7ca2ffe03d98a913189) @@ -14,10 +14,29 @@ ############################################################################ import os -from configuration.strings import * AUT_NAME = "denaliSquish" COMMON_PATH = os.environ['HOME']+"/Projects" - +COLOR_CODES = {"Aqua": "#81ffff", "Blue": "#017dea", "Blue 2": "#1090d5", "Green": "#29fd2d", "Grey": "#d3d3d3", + "Lavender": "#db98f5", "Light Blue": "#acd7f1", "Light Teal": "#29f1da","Lime": "#b7fc36", + "Magenta":"#fd28fd", "Orange": "#f2721c", "Peach":"#f1979a", "Red": "#c53b33", "Rose":"#fc178d", + "Slate blue":"#7f7ffa", "Violet": "#6435c9", "White": "#ffffff", "Yellow": "#fcfc4d"} + +#standby mode +GOODMORNING_START_TIME_SEC = 0 +GOODEVENING_START_TIME_SEC = 43200 + +BLOOD_PRIMING_TEXT = "Blood Priming" +SALINE_UNIT = "mL" +BLOOD_PRIMING_DEFAULT_VALUE = "0 mL" +EPOCH_INDEX = 0 +SEQUENCE_INDEX = 1 +CRC_INDEX = 2 +CODE_INDEX = 3 +PARAMETER_COUNT = 4 +OPMODE_INDEX = 5 +SUBMODE = 6 + + Index: shared/scripts/configuration/utility.py =================================================================== diff -u -rcd63acd00fab420ba8683b71843b75b0f8bf503a -re620a9fd664c96a736c0d7ca2ffe03d98a913189 --- shared/scripts/configuration/utility.py (.../utility.py) (revision cd63acd00fab420ba8683b71843b75b0f8bf503a) +++ shared/scripts/configuration/utility.py (.../utility.py) (revision e620a9fd664c96a736c0d7ca2ffe03d98a913189) @@ -30,8 +30,6 @@ date = datetime.now() return str(date.strftime(date_format)) - - def append_cloudsync_credentials_file(): try: @@ -45,61 +43,54 @@ except OSError as error: test.log("Directory can not be created") -def get_extracted_input_buf_file(): +def get_cloud_sync_input_file(): """ - This function is the handler for getting file from cloudsync folder. - - This handler will go inside cloudsync folder and looks for newly added input buf file based on current time. - if it satisfied that condition, it will return the exact path of newly created input buf file. - - Application input buf file is automatically created in '/home/denali/Desktop/sd-card/cloudsync/ {current_date}_inp.buf' - - @return latest_file - (string) returns latest file that append in cloudsync folder + This function is the handler for getting file from log folder. + Application log file is automatically created on '/home/denali/Desktop/sd-card/cloudsync/ {current_date}_inp.log' + @return latest_file - (string) returns latest file that append on log folder from sd-data """ try: current_date = get_current_date_and_time(date_format = "%Y_%m_%d") - latest_file = '/home/denali/Desktop/sd-card/cloudsync/'+current_date+'_inp.buf' + latest_file = CLOUD_SYNC_LOG_LOCATION+current_date+'_inp.buf' return latest_file except: - return False - - -def get_cloud_sync_output_file(): - """ - This function is the handler for getting file from log folder. - Application log file is automatically created on '/home/denali/Desktop/sd-card/cloudsync/ {current_date}_out.log' - @return latest_file - (string) returns latest file that append on log folder from sd-data - """ - try: - current_date = get_current_date_and_time(date_format = "%Y_%m_%d") - latest_file = CLOUD_SYNC_LOG_LOCATION+current_date+'_out.buf' - return latest_file - except: - return False + return False - -def retrive_log_data(readline_count): +def retrive_log_data(readline_count = 16): """ - This function is the handler for getting file from cloudsync folder. - Application log data is automatically appended on '/home/denali/Desktop/sd-card/cloudsync/ {current_date}_inp.buf' - @cloudsync_data - (list) returns latest file that append in cloudsync folder + This function is the handler for getting file from log folder. + Application log data is automatically appended on '/home/denali/Desktop/sd-card/cloudsync/ {current_date}_out.log' + @return latest_file - (list) returns latest file that append on log folder from sd-data """ cloudsync_data = [] count = 0 - file_name = get_extracted_input_buf_file() + file_name = get_cloud_sync_input_file() + try: with open(file_name,mode = 'r') as filereader: contents = csv.reader(filereader) - try: - for reader in reversed(list(contents)): + try: + for reader in reversed(list(contents)): if readline_count == count: - return cloudsync_data - cloudsync_data.append(reader) + return cloudsync_data + cloudsync_data.append(reader) count = count + 1 except: - test.fail("application log data is corrupted") + test.fail("application log data is corrupted") except: test.fail("Log file is not created or log file is not created based on standard log naming format.") - - \ No newline at end of file +def get_epoch_value_consistancy(current_epoch_value, expected_epoch_value): + """ + This function is verify consistancy of epoch value. + @input current_epoch_value (float) - current epoch time + @input expected_epoch_value (str) - epoch time from the cloud-sync log. + @return (bool) - True/False + """ + expected_epoch_value = int(expected_epoch_value) + maximum_epoch_value = int(current_epoch_value + 25) + minimum_epoch_value = int(current_epoch_value - 25) + + if expected_epoch_value > minimum_epoch_value and expected_epoch_value < maximum_epoch_value : + return True + return False \ No newline at end of file Index: tst_cloud_sync_hd_opmodes/test.py =================================================================== diff -u -rcd63acd00fab420ba8683b71843b75b0f8bf503a -re620a9fd664c96a736c0d7ca2ffe03d98a913189 --- tst_cloud_sync_hd_opmodes/test.py (.../test.py) (revision cd63acd00fab420ba8683b71843b75b0f8bf503a) +++ tst_cloud_sync_hd_opmodes/test.py (.../test.py) (revision e620a9fd664c96a736c0d7ca2ffe03d98a913189) @@ -17,131 +17,152 @@ import names +import time + from dialin.ui.hd_simulator import HDSimulator from dialin.ui import utils from configuration import utility from configuration import config from dialin.common.hd_defs import HDOpModes, HDOpSubModes - hd_simulator = HDSimulator() -MESSAGE_FAULT_MODE = '1639391827,1,0,1006,2,0,0' MESSAGE_SERVICE_MODE = '1639391827,1,0,1006,2,1,0' MESSAGE_INIT_MODE = '1639391827,1,0,1006,2,2,0' MESSAGE_STAND_BY_MODE = '1639391827,1,0,1006,2,3,0' MESSAGE_TPAR_MODE = '1639391827,1,0,1006,2,4,0' MESSAGE_PRETREATMENT_MODE = '1639391827,1,0,1006,2,5,0' -MESSAGE_PRETREATMENT_MODE = '1639391827,1,0,1006,2,6,0' +MESSAGE_TREATMENT_MODE = '1639391827,1,0,1006,2,6,0' MESSAGE_POST_TREATMENT_MODE = '1639391827,1,0,1006,2,7,0' MESSAGE_ILLEGAL_TRANSITION_MODE = '1639391827,1,0,1006,2,8,0' DEVICE_CREDENTIALS_RESPONSE = '1639391827,1,0,1004,1,/tmp/credentials/' -def navigate_to_hd_opmodes(): +def verify_hd_opmodes(): """ Method to navigate different hd opmodes to verify message id's logged in cloudsync """ - test.startSection("Navigating to different hd opmodes to verify message id's") - test.log("Fault mode") - hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_FAUL.value, sub_mode=HDOpSubModes.STANDBY_START_STATE.value) - utils.waitForGUI(1) + test.startSection("Navigating to different hd opmodes to verify message id's") + test.startSection("Verifying message id, opmode, submode for intialisation screen") hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_INIT.value, sub_mode=HDOpSubModes.STANDBY_START_STATE.value) utils.waitForGUI(1) - test.log("Standby mode") + test.log("Initial mode") + actual_inp_buf_data = utility.retrive_log_data(2) + test.log(str(actual_inp_buf_data[0])) + verify_log_response(actual_inp_buf_data[0],MESSAGE_INIT_MODE) + verify_log_response_credentials(actual_inp_buf_data[1],DEVICE_CREDENTIALS_RESPONSE) + test.endSection() + + test.startSection("Verifying message id, opmode, submode for Standby mode screen") hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_STAN.value, sub_mode=HDOpSubModes.STANDBY_START_STATE.value) utils.waitForGUI(1) - test.log("Treatment mode") + test.log("Stand By mode") + actual_inp_buf_data = utility.retrive_log_data(4) + test.log(str(actual_inp_buf_data[2])) + verify_log_response(actual_inp_buf_data[2],MESSAGE_STAND_BY_MODE) + verify_log_response_credentials(actual_inp_buf_data[3],DEVICE_CREDENTIALS_RESPONSE) + test.endSection() + + test.startSection("Verifying message id, opmode, submode for Treatment parameter mode screen") hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_TPAR.value, sub_mode=HDOpSubModes.STANDBY_START_STATE.value) utils.waitForGUI(1) - test.log("Pre- Treatment mode") + test.log("Treatment parameter mode") + actual_inp_buf_data = utility.retrive_log_data() + test.log(str(actual_inp_buf_data[4])) + verify_log_response(actual_inp_buf_data[4],MESSAGE_TPAR_MODE) + verify_log_response_credentials(actual_inp_buf_data[5],DEVICE_CREDENTIALS_RESPONSE) + test.endSection() + + test.startSection("Verifying message id, opmode, submode for Pre -treatment mode screen") hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_PRET.value, sub_mode=HDOpSubModes.STANDBY_START_STATE.value) utils.waitForGUI(1) - test.log("Treatment mode") + test.log("Pre - Treatment mode") + actual_inp_buf_data = utility.retrive_log_data() + test.log(str(actual_inp_buf_data[6])) + verify_log_response(actual_inp_buf_data[6],MESSAGE_PRETREATMENT_MODE) + verify_log_response_credentials(actual_inp_buf_data[7],DEVICE_CREDENTIALS_RESPONSE) + test.endSection() + + test.startSection("Verifying message id, opmode, submode for Treatment mode screen") hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_TREA.value, sub_mode=HDOpSubModes.STANDBY_START_STATE.value) utils.waitForGUI(1) - test.log("Post treatment Screen") + test.log("Treatment mode") + actual_inp_buf_data = utility.retrive_log_data() + test.log(str(actual_inp_buf_data[8])) + verify_log_response(actual_inp_buf_data[8],MESSAGE_TREATMENT_MODE) + verify_log_response_credentials(actual_inp_buf_data[9],DEVICE_CREDENTIALS_RESPONSE) + test.endSection() + + test.startSection("Verifying message id, opmode, submode for Post treatment mode screen") hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_POST.value, sub_mode=HDOpSubModes.STANDBY_START_STATE.value) utils.waitForGUI(1) - test.log("Service Screen") + test.log("Post - treatment mode") + actual_inp_buf_data = utility.retrive_log_data() + test.log(str(actual_inp_buf_data[10])) + verify_log_response(actual_inp_buf_data[10],MESSAGE_POST_TREATMENT_MODE) + verify_log_response_credentials(actual_inp_buf_data[11],DEVICE_CREDENTIALS_RESPONSE) + test.endSection() + + test.startSection("Verifying message id, opmode, submode for Service mode screen") hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_SERV.value, sub_mode=HDOpSubModes.STANDBY_START_STATE.value) utils.waitForGUI(1) - test.log("Illegal trasition") + test.log("Service mode") + actual_inp_buf_data = utility.retrive_log_data() + test.log(str(actual_inp_buf_data[12])) + verify_log_response(actual_inp_buf_data[12],MESSAGE_SERVICE_MODE) + verify_log_response_credentials(actual_inp_buf_data[13],DEVICE_CREDENTIALS_RESPONSE) + test.endSection() + + test.startSection("Verifying message id, opmode, submode for Illegal transition screen") hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_NLEG.value, sub_mode=HDOpSubModes.STANDBY_START_STATE.value) mouseClick(waitForObjectExists(names.device_settings_screen)) - test.log ("Service mode") + test.log ("Illegal Transition") + actual_inp_buf_data = utility.retrive_log_data() + test.log(str(actual_inp_buf_data[14])) + verify_log_response(actual_inp_buf_data[14],MESSAGE_ILLEGAL_TRANSITION_MODE) + verify_log_response_credentials(actual_inp_buf_data[15],DEVICE_CREDENTIALS_RESPONSE) test.endSection() + + test.endSection() -def verify_code_from_treatment_response(): +def verify_log_response(actual_cloudsync_data, expected_cloudsync_data): """ - This function is used for verify the code text from treatment log. + This function is used for verifying the expected and actual cloud-sync log data. """ - 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, 1): - if indicator != POST_TREATMENT_INSTRUCTION_COUNT: - mouseClick(waitForObject(names.o_patient_Disconnection_right_arrow)) - utils.waitForGUI(2) - mouseClick(names.o_patientDisconnectionConfirm_CONFIRM_Text) - message_text = MESSAGE_FAULT_MODE.split(',') + 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() - utils.waitForGUI(3) #delay for fetching the effect - text_code_label = str(waitForObjectExists(names.o_treatmentReviewConfirm_Code_Tx_code_Label).text) - test.compare(message_text[5], text_code_label.split(': ')[1], "CS2UI message should be verified") - - -def get_data_from_cloud_sync_input_file(message): + 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.OPMODE_INDEX, config.SUBMODE): + test.compare(expected_cloudsync_data[index], actual_cloudsync_data[index], "cloud sync data %s is verified" %expected_cloudsync_data[index]) + test.endSection() + +def verify_log_response_credentials(actual_cloudsync_data, expected_cloudsync_data): """ - 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 read on input.buf file. + This function is used for verifying the expected and actual cloud-sync log data. """ - cloud_out_log = utility.get_cloud_sync_input_file() + 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() - with open(cloud_out_log, "r") as filereader: - filereader.read(message) - utils.waitForGUI(5) - - -def verify_log_response(actual_cloudsync_data, expected_cloudsync_data): + 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") - test.startSection("Verification of cloud response for code %s" %actual_cloudsync_data[3]) - expected_cloudsync_data = expected_cloudsync_data.split(',') - for index in (0, 2, 3, 4, 5): + for index in (config.CRC_INDEX, config.CODE_INDEX, config.PARAMETER_COUNT, config.OPMODE_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) - - navigate_to_hd_opmodes() - #utility.get_cloud_sync_input_file() -# test.log("input.buf") - inp_buf_data = utility.retrive_log_data(16) - test.log(str(inp_buf_data)) - message_id_1 = inp_buf_data[0][3] - num_of_parameters_1 = inp_buf_data[0][4] - opmode_1 = inp_buf_data[0][5] - submode_1 = inp_buf_data[0][6] + utils.waitForGUI(3) + verify_hd_opmodes() - message_id_2 = inp_buf_data[1][3] - num_of_parameters_2 = inp_buf_data[1][4] - constant_2 = inp_buf_data[1][5] - - message_id_3 = inp_buf_data[2][3] - num_of_parameters_3 = inp_buf_data[2][4] - opmode_3 = inp_buf_data[2][5] - submode_3= inp_buf_data[2][6] - test.log(str(message_id_1)) -# num_of_parameters_1,opmode_1)) -# test.log(str(message_id_2,num_of_parameters_2,constant_2)) -# test.log(str(message_id_3,num_of_parameters_3,opmode_3)) -# test.log(str(submode_1,submode_3)) -# test.log("file created") -# get_data_from_cloud_sync_input_file(message = MESSAGE_ILLEGAL_TRANSITION_MODE) utils.tstDone() \ No newline at end of file