# -*- 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_inhandled_message_log # date 2022/07/08 # author Shweta Policepatil # # # # NOTE: # This test verifies the cloud sync - Hd opmodes. import builtins import csv import datetime import os from builtins import int as pyInt from builtins import str as pyStr 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() CLOUD_CREDENTIALS_LOCATION = '/home/denali/Desktop/cloudsync/credentials' CLOUD_SYNC_LOG_LOCATION = '/home/denali/Desktop/sd-card/cloudsync' #MESSAGE_1 = '1639391827,1,0,2006,0' NUM_OF_HD_OPMODES = [0,1,2,3,4,5,6,7,8] NUM_OF_SUBMODES = 0 MESSAGE_2 = '1639391827,1,0,1006,2,0,0' PEM_FILES = ['1.pem', '2.pem', '3.pem' ] def get_cloud_sync_input_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}_inp.log' @return latest_file - (string) returns latest file that append on log folder from sd-data """ try: current_date = utility.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 = utility.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 def append_cloudsync_credentials_file(): try: os.makedirs(CLOUD_CREDENTIALS_LOCATION, exist_ok = True) test.log("Directory created successfully") for file_handler in range(len(PEM_FILES)): path = os.path.join(CLOUD_CREDENTIALS_LOCATION, PEM_FILES[file_handler]) with open(path, 'w') as file_reader: pass except OSError as error: test.log("Directory can not be created") def navigate_to_hd_opmode_(): """ 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") 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") 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") 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") 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") 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") 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) # test.endSection() def retrive_log_data(): 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],reader[7]) 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 set_data_in_cloud_sync_output_file(location, 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 """ with open(location, "a") as filereader: filereader.write(message) utils.waitForGUI(5) # def set_data_in_cloud_sync_input_file(location, 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 # """ # with open(location, "r") as filereader: # test.log(message) # filereader.read(message) # test.log(message) # utils.waitForGUI(5) def main(): append_cloudsync_credentials_file() utils.tstStart(__file__) startApplication(config.AUT_NAME) utils.waitForGUI(1.5) navigate_to_hd_opmode_() cloud_out_log = get_cloud_sync_output_file() set_data_in_cloud_sync_output_file(cloud_out_log, MESSAGE_2) # cloud_out_log = get_cloud_sync_input_file() # set_data_in_cloud_sync_input_file(cloud_out_log, MESSAGE_2) # utils.waitForGUI(1) # set_data_in_cloud_sync_input_file(cloud_out_log, MESSAGE_2) set_data_in_cloud_sync_output_file(cloud_out_log, MESSAGE_2) retrive_log_data() utils.tstDone()