# -*- 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_post_treatment_disconnection # date 2022/05/20 # author Joseph Varghese # author Akshay Dhawan # author Shweta Policepatil # author Amol Shinde # NOTE: # This test contradicts verification of post treatment section. import names import csv import os import test import builtins from builtins import int as pyInt from pathlib import Path from dialin.ui import utils from datetime import * from configuration import config from configuration import application_init as application_init from dialin.ui.hd_simulator import HDSimulator from configuration import utility from dialin.common.hd_defs import HDOpModes, HDStandbyStates, PostTreatmentStates hd_simulator = HDSimulator() SCREEN_OBJ1 = names.o_PostTreatmentStack_patientDisconnectionConfirm_TreatmentFlowBase SCREEN_OBJ3 = names.o_PostTreatmentStack_disposablesRemovalConfirm_TreatmentFlowBase num_of_instructionss = 0 rightarrow_obj_names = [names.o_PostTreatmentStack_patientDisconnectionConfirm_TreatmentFlowBase] leftarrow_obj_names = [names.o_PostTreatmentStack_patientDisconnectionConfirm_TreatmentFlowBase] pth = Path(os.getcwd()) def get_rightarrow_obj(screen_obj): """ To obtain object for right arrow based on container @param (str) screen_obj: provides the container @returns a real name object """ names.o_right_arrow["container"] = screen_obj return names.o_right_arrow def navigation_to_post_treatment(): """ Method for navigating to post treatment disconnection submode. """ hd_simulator.cmd_send_hd_operation_mode(op_mode = HDOpModes.MODE_POST.value, sub_mode=PostTreatmentStates.HD_POST_TREATMENT_PATIENT_DISCONNECTION_STATE.value) for iteration in range(2): if object.exists(get_rightarrow_obj(rightarrow_obj_names[0])): mouseClick(waitForObject(get_rightarrow_obj(rightarrow_obj_names[0]))) def post_treatment_review_text_obj(text): """ Method to set object property as dynamic. @param text : (str) treatment parameter text to be appended with object properties """ names.o_PostTreatmentStack_treatmentReviewConfirm_PostTreatmentReview_ONE["text"] = text return names.o_PostTreatmentStack_treatmentReviewConfirm_PostTreatmentReview_ONE def review_text(text, occurrence=None): """ Method to set object property based on text @param text : (str) treatment parameter text """ if isinstance(text, str): if ("occurrence" in names.o_review_text): # occurrence key exists, remove the key # Need to do this because we're re-using the o_review_text for labels and units del names.o_review_text["occurrence"] if occurrence is not None: names.o_review_text["occurrence"] = occurrence names.o_review_text["text"] = text return names.o_review_text else: test.log("Invalid \"text\": " +text+" for object.") names.o_review_text["text"] = None def verify_right_instruction_navigation_patient_disconnection(num_of_instructions): """. Method to verify right arrow functionality @param num_of_instructions - (int) count the number of instructions """ test.startSection("verifying right arrow functionality for Patient Disconnection Screen") utility.verify_missing_object(names.o_patient_Disconnection_left_arrow) for indicator in range(1, num_of_instructions, 1): verify_bullet_navigation(indicator, num_of_instructions, SCREEN_OBJ1) if indicator != num_of_instructions: mouseClick(waitForObject(names.o_patient_Disconnection_right_arrow)) test.compare(str(waitForObjectExists(names.o_patient_disconnection_text).text), config.PATIENT_DISCONNECTION_TEXT, "Patient Disconnection text must be {}".format(config.PATIENT_DISCONNECTION_TEXT)) utility.verify_missing_object(names.o_patient_Disconnection_right_arrow) test.endSection() def verify_left_instruction_navigation_patient_disconnection(num_of_instructions): """ Method to verify left arrow functionality and verify the status of back, next and confirm button if available @param num_of_instructions - (int) count the number of instructions, """ test.startSection("verifying left arrow functionality for Patient Disconnection Screen") utility.verify_missing_object(names.o_patient_Disconnection_right_arrow) for indicator in range(num_of_instructions, 0, -1): verify_bullet_navigation(indicator, num_of_instructions, SCREEN_OBJ1) if indicator != 1: mouseClick(waitForObject(names.o_patient_Disconnection_left_arrow)) test.compare(str(waitForObjectExists(names.o_patient_disconnection_text).text), config.PATIENT_DISCONNECTION_TEXT, "Patient Disconnection text must be {}".format(config.PATIENT_DISCONNECTION_TEXT)) utility.verify_missing_object(names.o_patient_Disconnection_left_arrow) test.endSection() def verify_right_instruction_navigation_disposables(num_of_instructions): """ Method to verify right arrow functionality and verify status confirm button if available @param num_of_instructions - (int) count the number of instructions """ test.startSection("verifying right arrow functionality and Confirm Button for Disposables Screen") utility.verify_missing_object(names.o_disposables_removal_left_arrow) for indicator in range(1, num_of_instructions, 1): verify_bullet_navigation(indicator, num_of_instructions, SCREEN_OBJ3) test.verify(not waitForObjectExists(names.o_disposables_removal_confirm_button).enabled, "Confirm button should be disabled") if indicator != num_of_instructions: mouseClick(waitForObject(names.o_disposables_removal_right_arrow)) test.compare(findObject(names.o_disposables_removal_text).text, config.DISPOSABLES_REMOVAL_TEXT, "DISPOSABLES_REMOVAL_TEXT must be NEXT"+config.DISPOSABLES_REMOVAL_TEXT) utility.verify_missing_object(names.o_disposables_removal_right_arrow) test.verify(waitForObjectExists(names.o_disposables_removal_confirm_button).enabled ," confirm button must be active") test.verify(waitForObjectExists(names.o_disposables_removal_backButton).enabled, " Back button must be active") test.endSection() def verify_left_instruction_navigation_disposables(num_of_instructions): """ Method to verify left arrow functionality and verify status of back button. @param num_of_instructions - (int) count the number of instructions """ test.startSection("verifying left arrow functionality and Back Button for Disposables Screen") utility.verify_missing_object(names.o_disposables_removal_right_arrow) for indicator in range(num_of_instructions, 0, -1): verify_bullet_navigation(indicator, num_of_instructions, SCREEN_OBJ3) if indicator != 1: mouseClick(waitForObject(names.o_disposables_removal_left_arrow)) test.compare(waitForObjectExists(names.o_disposables_removal_text).text, config.DISPOSABLES_REMOVAL_TEXT, "DISPOSABLES_REMOVAL_TEXT must be "+config.DISPOSABLES_REMOVAL_TEXT) utility.verify_missing_object(names.o_disposables_removal_left_arrow) test.verify(waitForObjectExists(names.o_disposables_removal_backButton).enabled, " Back button must be active") test.endSection() def verify_bullet_navigation(num, num_of_instructions, screen_obj): """ Method to verify status of bullets based on number of instruction screen @param num - (int) number of indicator @param num_of_instructions- (int) count the number of instructions @param object - screen_obj - (str) Screen object """ test.startSection("instruction bullet verification for screens") for instruction in range(1, num_of_instructions): NUM_OF_POSTTREATMENT_BULLETS = len(config.POST_TREATMENT_SCREENS) bullet_children = object.children(waitForObjectExists(utility.get_bullet_object(screen_obj,(NUM_OF_POSTTREATMENT_BULLETS + instruction) - 1))) bullet_circle_color = bullet_children[0].color.name bullet_border_color = bullet_children[0].border.color.name if instruction <= num: test.compare(bullet_circle_color, config.COMPLETE_COLOR) test.compare(bullet_border_color,config.COMPLETE_COLOR) test.log(str(instruction) + " Complete bullet") else: test.compare(bullet_circle_color, config.CURRENT_COLOR) test.compare(bullet_border_color,config.INCOMPLETE_COLOR) test.log(str(instruction) + " Incomplete bullet") test.endSection() def verify_patient_disconnection_screens(): """ Method to verify parameters under 'Patient Disconnection Screens' """ test.startSection("verifying Patient Disconnection screen") PATIENT_DISCONNECTION_STEP = 0 utility.verify_page_step_indicator(SCREEN_OBJ1, PATIENT_DISCONNECTION_STEP, config.POST_TREATMENT_SCREENS) NUM_OF_PATIENT_DISCONNECTION_SCREENS = 2 verify_right_instruction_navigation_patient_disconnection(NUM_OF_PATIENT_DISCONNECTION_SCREENS) verify_left_instruction_navigation_patient_disconnection(NUM_OF_PATIENT_DISCONNECTION_SCREENS) verify_right_instruction_navigation_patient_disconnection(NUM_OF_PATIENT_DISCONNECTION_SCREENS) test.compare(str(waitForObjectExists(names.o_patient_disconnection_confirm_button).text), config.PATIENT_DISCONNECTION_CONFIRM_BUTTON_TEXT,"Confirm button text must be {}".format(config.PATIENT_DISCONNECTION_CONFIRM_BUTTON_TEXT)) test.verify(waitForObjectExists(names.o_patient_disconnection_confirm_button).enabled, "Confirm button must be enabled") mouseClick(waitForObject(names.o_patient_disconnection_confirm_button)) test.endSection() def verify_treatment_review_screen(): """ Method to verify parameters under 'Treatment Review Screens' and also to verify treatment message on clicking export button """ test.startSection("Verifying Treatment Review Screen") SCREEN_OBJ2 = names.o_PostTreatmentStack_treatmentReviewConfirm_PostTreatmentReview REVIEW_TREATMENT_STEP = 1 utility.verify_page_step_indicator(SCREEN_OBJ2, REVIEW_TREATMENT_STEP, config.POST_TREATMENT_SCREENS) test.verify(waitForObjectExists(names.o_treatmentReviewConfirm_Export_Text).enabled, " Export button should be enabled") test.compare(str(waitForObjectExists(names.o_treatmentReviewConfirm_Export_Text).text), config.EXPORT_TEXT,"Export button text must be {}".format(config.EXPORT_TEXT)) test.verify(waitForObjectExists(names.o_treatmentReviewConfirm_Export_Text).enabled, " Export button should be enabled") test.verify(waitForObject(names.o_eject_button).enabled, "Eject button must be enable") mouseClick(waitForObject(names.o_PostTreatment_Export_mousearea)) test.compare(str(waitForObjectExists(names.o_treatmentlog_msg_export_btn).text), config.EXPORT_LOG_MSG,"Treatment log message on clicking Export button text must be {}".format(config.EXPORT_LOG_MSG)) test.verify(waitForObjectExists(names.o_treatmentlog_msg_export_btn).visible, "Treatment log message should be visible") utils.waitForGUI(1) # mouseClick(waitForObject(names.o_eject_button)) # test.verify(not waitForObjectExists(names.o_treatmentReviewConfirm_Export_Text).enabled, " Export button should be disabled") verify_post_treatment_review_parameters() verify_post_treatment_review_log_data() test.verify(waitForObject(names.o_treatment_review_next_button).enabled, "NEXT button must be enabled") mouseClick(waitForObject(names.o_treatment_review_next_button)) test.endSection() def verify_disposables_screen(): """ Method to verify parameters under 'Disposables screens' """ test.startSection("verification of Disposable Screen") mouseClick(waitForObject(names.o_next_button_review)) utils.waitForGUI(1) # HD Response to confirm/next click hd_simulator.cmd_send_confirm_post_tx_next(accepted = 1, reason=0) utils.waitForGUI(1) #Delay given for screen navigation mouseClick(waitForObjectExists(names.o_disposablesRemovalConfirm_BACK_Text)) utils.waitForGUI(0.3) #Delay given for screen navigation mouseClick(waitForObject(names.o_treatment_review_next_button)) hd_simulator.cmd_send_confirm_post_tx_next(accepted = 1, reason=0) utils.waitForGUI(1) #Delay given for screen navigation DISPOSABLES_STEP = 2 utility.verify_page_step_indicator(SCREEN_OBJ3, DISPOSABLES_STEP, config.POST_TREATMENT_SCREENS) names.o_bullet_object.pop("occurrence") NUM_OF_DISPOSABLE_INSTALLATION_SCREENS = 10 verify_right_instruction_navigation_disposables(NUM_OF_DISPOSABLE_INSTALLATION_SCREENS) test.verify(waitForObjectExists(names.o_disposables_removal_confirm_button).enabled, " confirm button must be active") verify_left_instruction_navigation_disposables(NUM_OF_DISPOSABLE_INSTALLATION_SCREENS) test.verify(not waitForObjectExists(names.o_disposables_removal_confirm_button).enabled, " confirm button must not to be active") verify_right_instruction_navigation_disposables(NUM_OF_DISPOSABLE_INSTALLATION_SCREENS) test.verify(waitForObjectExists(names.o_disposables_removal_confirm_button).enabled ," confirm button must be active") mouseClick(waitForObjectExists(names.o_disposables_removal_confirm_button)) test.endSection() _unit_label_count = { "mL/min" : 0, "mL" : 0 } def get_unit_occurence(whichUnit): if whichUnit in _unit_label_count: _unit_label_count[whichUnit] += 1 return _unit_label_count[whichUnit] return -1 def verify_post_treatment_review_parameters(): """ Method to verify post treatment review parameters and unit by default. """ test.startSection("verification of post treatment review parameters with Units") treatment_review_text = waitForObjectExists(post_treatment_review_text_obj(config.TREATMENT_REVIEW_TITLE_TEXT)) test.compare(treatment_review_text.text, config.TREATMENT_REVIEW_TITLE_TEXT, "{} screen is displayed".format(config.TREATMENT_REVIEW_TITLE_TEXT)) # the code generation is not necessary to check since the CS is not running therefore code won't get updated. # test.compare(str(waitForObjectExists(names.o_code_text).text), config.CODE_TEXT, "Code text must be {}".format(config.CODE_TEXT)) utils.waitForGUI(0.1) # resetting the unit key count for key in _unit_label_count.keys(): _unit_label_count[key] = 0 for parameter in config.POST_TREATMENT_REVIEW_SCREEN_UNITS.keys(): utility.scroll_to_zone(review_text(parameter), names.o_review_area) parameter_text = waitForObjectExists(review_text(parameter)) test.log("verification of parameter -> " + str(parameter)) test.compare(parameter_text.text, parameter, "{} should be available under 'Treatment Review' screen".format(parameter)) unit = config.POST_TREATMENT_REVIEW_SCREEN_UNITS[parameter] if unit is not "": # some params don't have units, check before verifying units label occurence = get_unit_occurence(unit) if occurence >= 2 : unit_dict = review_text(unit, occurrence=occurence) else: unit_dict = review_text(unit) unit_text = utility.get_object_from_names(unit_dict, f"Unit ({unit}) Object of {parameter} is missing" + str(unit_dict)) if unit_text is not None: test.log("verification of unit for data -> " + str(parameter)) test.compare(unit_text.text, unit, "{} should be available under 'Treatment Review' screen".format(unit)) test.endSection() def getting_patient_id_from_create_custom_treatment(patient_id): """ Method to fetch patient ID from create custom treatment in post treatment review """ test.startSection("Method for navigate to create custom treatment to set all parameters ") hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_TPAR.value, sub_mode = HDStandbyStates.STANDBY_START_STATE.value) mouseClick(waitForObject(names.o_input_patient_id)) type(waitForObject(names.o_input_patient_id), patient_id) mouseClick(waitForObject(names.o_continue_button)) test.endSection() def verify_post_treatment_review_log_data(): """ Method to verify all the log values on review submode of post treatment Here we are verifying 5 set of values to the review section based in edge condition. This section covers log verification. """ test.startSection("Method to check all the values of parameters in review screen") for index in range (2): getting_patient_id_from_create_custom_treatment(config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Patient ID"][index]) navigation_to_post_treatment() mouseClick(waitForObject(names.o_patientDisconnectionConfirm_Button)) review_parameter_value = review_text(text = "Patient ID") utility.scroll_to_zone(review_parameter_value, names.o_review_area, direction = "Top") hd_simulator.cmd_send_post_treatment_log_response(accepted = 1, reason = 0, device_id = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Device ID"][index], treatment_duration = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Treatment Duration"][index], bood_flow_rate = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Blood Flow Rate"][index], dialysate_flow_rate = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Dialysate Flow Rate"][index], acid_concentrate_type = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Acid ConcentrateType"][index], bicarbonate_concentrate_type = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Bicarbonate Concentrate Type"][index], potassium_concentration = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Potassium Concentration"][index], calcium_concentration = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Calcium Concentration"][index], bicarbonate_concentration = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Bicarbonate Concentration"][index], sodium_concentration = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Sodium Concentration"][index], dialysate_temperature = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Dialysate Temperature"][index], dialyzer_type = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Dialyzer Type"][index], heparin_type = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Heparin Type"][index], heparin_bolus_volume = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Heparin Bolus Volume"][index], heparin_dispense_rate = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Heparin Dispense Rate"][index], heparin_pre_stop = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Heparin Stop"][index], treatment_start_date_time = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Treatment Start DateTime"][index], treatment_end_date_time = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Treatment End DateTime"][index], actual_treatment_duration = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Actual Treatment Duration"][index], dialysate_volume_used = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Dialysate Volume Used"][index], origin_uf_volume = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Prescribed UF Volume"][index], target_uf_volume = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Target UF Volume"][index], actual_uf_volume = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Actual UF Volume"][index], saline_bolus_volume = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Saline Bolus Volume"][index], heparin_delivered_volume = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Heparin Delivered Volume"][index], water_sample_test_result = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Water Sample Test Result"][index], origin_uf_rate = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Prescribed UF Rate"][index], target_uf_rate = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Target UF Rate"][index], actual_uf_rate = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Actual UF Rate"][index], average_blood_flow = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Average Blood Flow"][index], average_dialysate_flow = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Average Dialysate Flow"][index], average_dialysate_temp = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Average Dialysate Temp"][index], average_arterial_pressure = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Average Arterial Pressure"][index], average_venous_pressure = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE["Average Venous Pressure"][index] ) utils.waitForGUI(1) hd_simulator.cmd_send_treatment_log_data(blood_flow_rate = config.TREATMENT_DATA_PARAMETER["Blood flow rate"][index], dialysate_flow_rate = config.TREATMENT_DATA_PARAMETER["Dialysate flow rate"][index], uf_rate = config.TREATMENT_DATA_PARAMETER["Uf rate"][index], arterial_pressure = config.TREATMENT_DATA_PARAMETER["Arterial pressure"][index], venous_pressure = config.TREATMENT_DATA_PARAMETER["Venous pressure"][index]) hd_simulator.cmd_send_treatment_log_alarm(alarm_id = config.TREATMENT_ALARM_PARAMETER["Alarm Id"][index], parameter1 = config.TREATMENT_ALARM_PARAMETER["Parameter1"][index], parameter2 = config.TREATMENT_ALARM_PARAMETER["Parameter2"][index]) hd_simulator.cmd_send_treatment_log_event(event_id = config.TREATMENT_EVENT_PARAMETER["Event id"][index], old_value = config.TREATMENT_EVENT_PARAMETER["Old Value"][index], new_value = config.TREATMENT_EVENT_PARAMETER["New Value"][index]) mouseClick(waitForObject(names.o_PostTreatment_Export_mousearea)) # test.compare(str(waitForObjectExists(names.o_treatmentlog_msg_export_btn).text), config.EXPORT_LOG_MSG,"Treatment log message on clicking Export button text must be {}".format(config.EXPORT_LOG_MSG)) # test.verify(waitForObjectExists(names.o_treatmentlog_msg_export_btn).visible, "Treatment log message should be visible") utils.waitForGUI(3) # wait for export to happen test.compare(str(waitForObjectExists(names.o_code_text_after_passing_log_values).text), config.CODE_TEXT_AFTER_LOGGING, "Code text must be {}".format(config.CODE_TEXT_AFTER_LOGGING)) test.startSection("verification of post treatment review values for iteration -> "+str(index+1)) test.log("###verification of post treatment UI data based on config data") for parameters_value in config.POST_TREATMENT_REVIEW_PARAMETER_RANGE.keys(): parameter_set = config.POST_TREATMENT_REVIEW_PARAMETER_RANGE[parameters_value] # get values from treatment log file parameter_value, parameter_unit = verify_parameter_from_post_treatment_log(parameters_value) #TODO: Following six parameters are not available on UI but API command is requiring this argument for logging. if parameters_value == "Device ID": continue elif parameters_value == "Average Blood Flow": continue elif parameters_value == "Average Dialysate Flow": continue elif parameters_value == "Average Dialysate Temp": continue elif parameters_value == "Average Arterial Pressure": continue elif parameters_value == "Average Venous Pressure": continue elif parameters_value == "Prescribed UF Rate": review_parameter_value = review_text(text = "Prescribed UF Rate") utility.scroll_to_zone(targetObject = review_parameter_value, screen_object = names.o_review_area) elif parameters_value == "Heparin Type" \ or parameters_value == " Delivered Volume" : # review_parameter_value = review_text(text = config.HEPARIN_TYPE) # utility.scroll_to_zone(targetObject = review_parameter_value, screen_object = names.o_review_area) # parameter_text = waitForObject(review_parameter_value) # test.log("verification of values for parameter ->" + parameters_value) # test.compare(config.HEPARIN_TYPE, parameter_text.text, "parameter value should be "+str(parameter_set[index])) if parameter_set[index] is 0: string_cmp = "NONE" continue elif parameters_value == "Heparin Bolus Volume" \ or parameters_value == "Heparin Dispense Rate" \ or parameters_value == "Heparin Stop" \ or parameters_value == "Heparin Delivered Volume" : # review_parameter_value = review_text(text = config.HEPARIN_TYPE) # utility.scroll_to_zone(targetObject = review_parameter_value, screen_object = names.o_review_area) # parameter_text = waitForObject(review_parameter_value) # test.log("verification of values for parameter ->" + parameters_value) # test.compare(config.HEPARIN_TYPE, parameter_text.text, "parameter value should be "+str(parameter_set[index])) if parameter_set[index] is 0: string_cmp = "OFF" continue elif parameters_value == "Bicarbonate Concentrate Type": review_parameter_value = review_text(text = config.BICARBONATE_CONCENTRATE) parameter_text = waitForObject(review_parameter_value) test.log("verification of values for parameter ->" + parameters_value) test.compare(config.BICARBONATE_CONCENTRATE, parameter_text.text, "parameter value should be "+str(parameter_set[index])) continue elif parameters_value == "Acid ConcentrateType": review_parameter_value = review_text(text = config.ACID_CONCENTRATE[index]) parameter_text = waitForObject(review_parameter_value) test.log("verification of values for parameter ->" + parameters_value) test.compare(config.ACID_CONCENTRATE[index], parameter_text.text, "parameter value should be "+str(parameter_set[index])) continue elif parameters_value == "Dialyzer Type": review_parameter_value = review_text(text = config.DIALYZER_TYPE[index]) parameter_text = waitForObject(review_parameter_value) test.log("verification of values for parameter ->" + parameters_value) test.compare(config.DIALYZER_TYPE[index], parameter_text.text, "parameter value should be "+str(parameter_set[index])) continue elif parameters_value == 'Water Sample Test Result': string_cmp = "Pass" if parameter_set[index] is 0: string_cmp = "Fail" test.log("verification of values for parameter ->" + parameters_value) review_parameter_value = review_text(text = string_cmp) test.log(str(review_parameter_value)) parameter_text = waitForObjectExists(review_parameter_value) test.log(f"{string_cmp} {str(parameter_text.text)}") test.compare(string_cmp, str(parameter_text.text), "parameter value should be "+str(string_cmp)) continue elif parameters_value == "Patient ID": # patient ID has no unit parameter_value, parameter_unit = verify_parameter_from_post_treatment_log(parameters_value) test.compare(parameter_value, parameter_set[index], "parameters value should be ->" + str(parameter_set[index])) continue if parameters_value == 'Treatment Start DateTime': start_date_time = datetime.fromtimestamp(parameter_set[index]).strftime('%Y/%m/%d %H:%M') review_parameter_value = review_text(text = str(start_date_time)) parameter_text = waitForObjectExists(review_parameter_value) test.compare(start_date_time, str(parameter_text.text), "parameter value should be "+str(parameter_set[index])) continue if parameters_value == 'Treatment End DateTime': end_date_time = datetime.fromtimestamp(parameter_set[index]).strftime('%Y/%m/%d %H:%M') review_parameter_value = review_text(text = str(end_date_time)) parameter_text = waitForObjectExists(review_parameter_value) test.compare(end_date_time, str(parameter_text.text), "parameter value should be "+str(parameter_set[index])) continue else: if isinstance(parameter_set[index], float): parameter_set[index] = ('%.3f' %parameter_set[index]) test.log("verification of values for parameter " + str(parameter_to_scroll)) parameter_to_scroll = str(parameter_set[index]) if parameters_value == "Treatment Duration": time_duration = builtins.int(utility.convert_seconds_into_min_and_sec(seconds=parameter_to_scroll, time_format="%M")) if time_duration < 1 : parameter_value = "0" else: parameter_value = str(time_duration) test.compare(parameter_to_scroll, str(parameter_set[index]), "raw treatment time parameters value should be ->" + str(parameter_set[index])) continue if parameters_value == 'Actual Treatment Duration': time_duration = builtins.int(utility.convert_seconds_into_min_and_sec(seconds=parameter_to_scroll, time_format="%M")) test.log(str(time_duration)) parameter_value = str(time_duration) test.compare(parameter_to_scroll, str(parameter_set[index]), "raw actual time parameters value should be ->" + str(parameter_set[index])) continue test.log("verification of post treatment log file data") test.log(f"{parameters_value} = {parameter_value} {parameter_unit}") result = test.compare(parameter_unit, config.POST_TREATMENT_REVIEW_SCREEN_UNITS[parameters_value], "parameters unit should be ->"+ config.POST_TREATMENT_REVIEW_SCREEN_UNITS[parameters_value]) if result is not True: test.compare(True, True) test.compare(parameter_value, str(parameter_set[index]), "parameters value should be ->" + str(parameter_set[index])) test.endSection() test.endSection() def verify_treatment_log_data(): """ Method to verify treatment log data """ test.startSection("Verification of treatment data") log_data = get_message_from_log("[Treatment Data]",parameter_count=9) for index in range (1): test.compare(log_data[1], "{val:.3f}".format(val=config.TREATMENT_DATA_PARAMETER["Blood flow rate"][index]), "parameters value should be ->" + str(config.TREATMENT_DATA_PARAMETER["Blood flow rate"][index])) test.compare(log_data[2], "{val:.3f}".format(val=config.TREATMENT_DATA_PARAMETER["Dialysate flow rate"][index]), "parameters value should be ->" + str(config.TREATMENT_DATA_PARAMETER["Dialysate flow rate"][index])) test.compare(log_data[3], str(config.TREATMENT_DATA_PARAMETER["Uf rate"][index]), "parameters value should be ->" + str(config.TREATMENT_DATA_PARAMETER["Uf rate"][index])) test.compare(log_data[4], str(config.TREATMENT_DATA_PARAMETER["Arterial pressure"][index]), "parameters value should be ->" + str(config.TREATMENT_DATA_PARAMETER["Arterial pressure"][index])) test.compare(log_data[5], str(config.TREATMENT_DATA_PARAMETER["Venous pressure"][index]), "parameters value should be ->" + str(config.TREATMENT_DATA_PARAMETER["Venous pressure"][index])) test.endSection() def verify_treatment_alarm_data(): """ Method to verify treatment alarm data """ test.startSection("Verification of treatment alarm data") log_data = get_message_from_log("[Treatment Alarms]",parameter_count=4) for index in range (1): test.compare(log_data[1].strip(), str(config.TREATMENT_ALARM_PARAMETER["Alarm Id String"][index]), "parameters value should be ->" + str(config.TREATMENT_ALARM_PARAMETER["Alarm Id"][index])) test.compare(log_data[2].strip(), str(config.TREATMENT_ALARM_PARAMETER["Parameter1"][index]), "parameters value should be ->" + str(config.TREATMENT_ALARM_PARAMETER["Parameter1"][index])) test.compare(log_data[3].strip(), str(config.TREATMENT_ALARM_PARAMETER["Parameter2"][index]), "parameters value should be ->" + str(config.TREATMENT_ALARM_PARAMETER["Parameter2"][index])) test.endSection() def verify_treatment_event_data(): """ Method to verify treatment event data """ test.startSection("Verification of treatment event data") log_data = get_message_from_log("[Treatment Events]",parameter_count=4) for index in range (1): test.compare(log_data[1].strip(), str(config.TREATMENT_EVENT_PARAMETER["Event id String"][index]), "parameters value should be ->" + str(config.TREATMENT_EVENT_PARAMETER["Event id"][index])) test.compare(log_data[2].strip(), str(config.TREATMENT_EVENT_PARAMETER["Old Value"][index]), "parameters value should be ->" + str(config.TREATMENT_EVENT_PARAMETER["Old Value"][index])) test.compare(log_data[3].strip(), str(config.TREATMENT_EVENT_PARAMETER["New Value"][index]), "parameters value should be ->" + str(config.TREATMENT_EVENT_PARAMETER["New Value"][index])) test.endSection() def get_message_from_log(parameter_text, parameter_count): """ This method intended to extract the message from treatment log. For row[index], index represent column to be extracted. @param parameter_text - (string) message text to be identified. @param parameter_count - number of arguments displayed. @return row - (list) API arguments displayed in log. """ count = 0 try: log_location = str(utility.get_extracted_file_from_post_treatment()) with open(log_location, 'r') as csv_file: try: for row in list(csv.reader(csv_file)): if row[0].isalpha(): pass else: row_length = sum(1 for values in row) if row[0]!= None and row[0] == parameter_text: count = 1 if count == 1 and row_length == parameter_count: return row else: pass except: test.fail("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_parameter_from_post_treatment_log(msg_text): """ To obtain the details of parameter from post treatment log file. @param msg_text: parameter to be extracted. @returns message value and message unit """ try: log_location = str(utility.get_extracted_file_from_post_treatment()) if log_location is not False: with open(log_location, 'r') as csv_file: try: for row in csv_file: columns = row.split(',') if columns[0] == msg_text and len(columns) == 3: return (columns[1].strip(),columns[2].strip()) elif columns[0] == msg_text and len(columns) == 2: return (columns[1].strip(),columns[1].strip()) else: pass except: test.fail("Treatment log data is corrupted") else: test.fail("Log file is not created or log file is not created based on standard log naming format.") except: test.fail("Log file is not created or log file is not created based on standard log naming format.") return (-1, -1) def main(): utils.tstStart(__file__) application_init.setup_post_log_successful_start(); # remove the old test execution result log files. utility.remove_files_in_folder(config.POST_TREATMENT_LOG_LOCATION) startApplication(config.AUT_NAME+ " -l") hd_simulator.cmd_send_hd_operation_mode(op_mode=HDOpModes.MODE_POST.value, sub_mode=PostTreatmentStates.HD_POST_TREATMENT_PATIENT_DISCONNECTION_STATE.value) verify_patient_disconnection_screens() verify_treatment_review_screen() verify_disposables_screen() verify_treatment_log_data() verify_treatment_alarm_data() verify_treatment_event_data() utils.tstDone()