Index: shared/scripts/configuration/config.py =================================================================== diff -u -r8a47faf40db724c9f1360b5e5579530410ee86ca -r26be96f1f86d5823902beed32660f12cdb87d881 --- shared/scripts/configuration/config.py (.../config.py) (revision 8a47faf40db724c9f1360b5e5579530410ee86ca) +++ shared/scripts/configuration/config.py (.../config.py) (revision 26be96f1f86d5823902beed32660f12cdb87d881) @@ -21,4 +21,3 @@ COMMON_PATH = os.environ['HOME']+"/Projects" -CLOUD_SYNC_INP__BUF_LOCATION = '/home/denali/Desktop/sd-card/cloudsync/*.buf' Index: shared/scripts/configuration/utility.py =================================================================== diff -u -r8a47faf40db724c9f1360b5e5579530410ee86ca -r26be96f1f86d5823902beed32660f12cdb87d881 --- shared/scripts/configuration/utility.py (.../utility.py) (revision 8a47faf40db724c9f1360b5e5579530410ee86ca) +++ shared/scripts/configuration/utility.py (.../utility.py) (revision 26be96f1f86d5823902beed32660f12cdb87d881) @@ -13,27 +13,25 @@ ############################################################################ import os -import names +import csv import test -import object -import squish -import glob -from configuration import config -from builtins import int as pyInt - + CLOUD_CREDENTIALS_LOCATION = '/home/denali/Desktop/cloudsync/credentials' -CLOUD_SYNC_LOG_LOCATION = '/home/denali/Desktop/sd-card/cloudsync' +CLOUD_SYNC_LOG_LOCATION = '/home/denali/Desktop/sd-card/cloudsync/' + PEM_FILES = ['1.pem', '2.pem', '3.pem' ] + def get_current_date_and_time(date_format='%Y/%b/%d - %H:%M:%S'): date = datetime.now() return str(date.strftime(date_format)) + def append_cloudsync_credentials_file(): try: @@ -47,21 +45,56 @@ except OSError as error: test.log("Directory can not be created") -def get_extracted_file_inp_buf(): + +def get_cloud_sync_input_file(): """ - This function is the handler for getting file from 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 = 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 - This handler will go inside cloudsync folder and looks for newly added input buf file based on current time. - If it is satisfying that condition, it will return the exact path of newly created input buf file. - @return latest_file - (string) returns latest file that append on log folder from sd-data +def retrive_log_data(readline_count = 1): """ + 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}_input.log' + @return latest_file - (list) returns latest file that append on log folder from sd-data + """ + cloudsync_data = [] + count = 0 + file_name = get_cloud_sync_input_file() try: - list_of_files = glob.glob(config.CLOUD_SYNC_INP__BUF_LOCATION) - latest_file = max(list_of_files, key=os.path.getctime) - test.log(str(latest_file)) - return str(latest_file) + with open(file_name,mode = 'r') as filereader: + contents = csv.reader(filereader) + try: + for reader in reversed(list(contents)): + if readline_count == count: + return cloudsync_data + cloudsync_data.append(reader) + count = count + 1 + except: + test.fail("application log data is corrupted") except: - test.fail("input buf file is not created during application interaction") - return False - \ No newline at end of file + 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 Index: shared/scripts/names.py =================================================================== diff -u -ra20cd43aecdf785029b1d01c5e224b1e1728261e -r26be96f1f86d5823902beed32660f12cdb87d881 --- shared/scripts/names.py (.../names.py) (revision a20cd43aecdf785029b1d01c5e224b1e1728261e) +++ shared/scripts/names.py (.../names.py) (revision 26be96f1f86d5823902beed32660f12cdb87d881) @@ -27,3 +27,5 @@ t_Text = {"container": o_Gui_MainView, "text": "t", "type": "Text", "unnamed": 1, "visible": True} o_PreTreatmentBase_confirmButton_TouchRect = {"container": o_PreTreatmentCreateStack_PreTreatmentBase_TreatmentFlowBase, "objectName": "_confirmButton", "type": "TouchRect", "visible": True} + +device_settings_screen = {"container": o_Gui_MainView, "text": "Treatment", "type": "Text", "unnamed": 1, "visible": True} Index: tst_cloud_sync_hd_opmodes/test.py =================================================================== diff -u -r8a47faf40db724c9f1360b5e5579530410ee86ca -r26be96f1f86d5823902beed32660f12cdb87d881 --- tst_cloud_sync_hd_opmodes/test.py (.../test.py) (revision 8a47faf40db724c9f1360b5e5579530410ee86ca) +++ tst_cloud_sync_hd_opmodes/test.py (.../test.py) (revision 26be96f1f86d5823902beed32660f12cdb87d881) @@ -14,27 +14,20 @@ # # NOTE: # This test verifies the cloud sync - tst_cloud_sync_hd_opmode. -import os -import csv -from builtins import int as pyInt -from builtins import str as pyStr + + +import names 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 -from dialin.utils.conversions import float_to_bytearray, integer_to_bytearray,unsigned_integer_to_bytearray - hd_simulator = HDSimulator() -expected_crc = 0 -expected_code = 1006 -expected_parameter = [] -expected_opmodes = [0, 1, 2, 3, 4, 5, 6, 7] -expected_submode = 0 -PEM_FILES = ['1.pem', '2.pem', '3.pem' ] + + 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' @@ -44,25 +37,17 @@ MESSAGE_PRETREATMENT_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' -expected_crc = '0' -expected_code = '1006' -expected_parameter = '2' -expected_opmodes = [0, 1, 2, 3, 4, 5, 6, 7] -expected_submode = '0' -PEM_FILES = ['1.pem', '2.pem', '3.pem' ] - +DEVICE_CREDENTIALS_RESPONSE = '1639391827,1,0,1004,1,/tmp/credentials/' + def navigate_to_hd_opmodes(): """ - Method to navigate different hd opmodes to verify message id's logged in cloudsync """ + 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.log("Service 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("Initial 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") @@ -84,39 +69,50 @@ 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") - hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_NLEG.value, sub_mode=HDOpSubModes.STANDBY_START_STATE.value) -# utils.waitForGUI(1) -# hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.NUM_OF_MODES.value, sub_mode=HDOpSubModes.STANDBY_START_STATE.value) -# + 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.endSection() -def retrive_log_data(): +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, 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(',') - file_name = utility.get_extracted_file_inp_buf() - try: - with open(file_name,mode = 'r') as f: - contents = csv.reader(f) - try: - for reader in reversed(list(contents)): - test.log(str(reader)) - return (reader[0],reader[1],reader[2],reader[3],reader[4],reader[5],reader[6]) - except: - 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.") - -def verify_cloud_sync_log_data(crc,code,parameter,opmode,submode): + 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): """ + 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. """ - test.startSection("verifying cloud sync log data ") - test.compare(crc,expected_crc) - test.compare(code,expected_code) - test.compare(parameter,expected_parameter) -# for opmode in range (len(expected_opmodes)): -# test.compare(opmode,expected_opmodes[0]) - test.compare(opmode,expected_opmodes) - test.compare(submode,expected_submode) - test.endSection() + cloud_out_log = utility.get_cloud_sync_input_file() + + with open(cloud_out_log, "r") as filereader: + filereader.write(message) + utils.waitForGUI(5) + + +def verify_log_response(actual_cloudsync_data, expected_cloudsync_data): + + 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): + 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() @@ -125,9 +121,10 @@ utils.waitForGUI(1.5) navigate_to_hd_opmodes() - - epoch_time,serial_num,crc,code,parameter,opmode,submode = retrive_log_data() - verify_cloud_sync_log_data(crc,code,parameter,opmode,submode) + utility.get_cloud_sync_input_file() + test.log("input.buf") + utility.retrive_log_data() + test.log("file created") utils.tstDone() \ No newline at end of file