# -*- 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_UI_log_verification # date 2022/07/06 # author Amol Pandharinath Shinde # author Akshay Dhawan # # NOTE: # This test verifies the logs for the messages provided from dialin.ui.hd_simulator_alarms import HDAlarmsSimulator import logging from dialin.protocols.CAN import DenaliCanMessenger from dialin.ui import HDSimulator from dialin.ui import DGSimulator from dialin.ui import utils from configuration import utility from configuration import config import builtins from dialin.common.msg_ids import MsgIds from dialin.common.hd_defs import HDOpSubModes, HDOpModes from dialin.common import msg_defs from dialin.utils.conversions import integer_to_bytearray, float_to_bytearray, short_to_bytearray, unsigned_integer_to_bytearray, integer_to_bit_array hd_simulator = HDSimulator() dg_simulator = DGSimulator() logger= logging.getLogger("INFO") messenger= DenaliCanMessenger('can0') hd_alarm= HDAlarmsSimulator(can_interface=messenger, logger=logger) NUMBER_OF_OPMODE = 9 NUMBER_OF_TREATMENTMODES = 3 CONDITIONS = [0,1] SW_VERSION = 6292356 ZERO = 0 THREE = 3 epoch = 0 def retrive_log_data(): file_name = utility.get_extracted_file_inp_buf() contain = [] try: with open(file_name) as f: contents = f.read() test.log(str(contents)) try: for reader in reversed(list(contents)): test.log(str(reader)) return (row[0],row[1],row[2],row[3],row[4],row[5],row[6],row[7],row[8]) 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.") # try: # log_location = str(utility.get_extracted_file()) # with open(log_location, 'r') as csv_file: # # try: # for row in csv_file: # reader = csv.reader(csv_file) # for row in reader: # row_length = sum(1 for values in row) # for row1 in row: # # if row[0]!= None and row[0] == msg_text and row_length == 3: # return (row[1],row[2],row[3],row[4],row[5],row[6],row[7],row[8]) # except: # test.fail("Treatment 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_log(epoch, ZERO, registration_request, THREE, HD_VERSION_SERIAL, DG_VERSION_SERIAL, SW_VERSION): """ This function verifies the UI log's. @param msg_id - (string) expected message_id's. @param msg - (string) message to be displayed on log. @param param - (list) parameters for msg_id's. """ test.startSection("Verification of UI log based on message ID : " + str(msg_id)) utils.waitForGUI(2) message_extracted = utility.get_current_log_details_from_inp_buf(epoch, ZERO, registration_request, THREE, HD_VERSION_SERIAL, DG_VERSION_SERIAL, SW_VERSION) test.verify(epoch in message_extracted, "EPOCH value is verified") test.verify(registration_request in message_extracted, "Registration request number is verified") test.verify(THREE in message_extracted, "THREE is verified") test.verify(HD_VERSION_SERIAL in message_extracted, "HD Version Serial is verified") test.verify(DG_VERSION_SERIAL in message_extracted, "DG Version Serial is verified") test.verify(SW_VERSION in message_extracted, "Software Version is verified") # message_id_hex = builtins.hex(builtins.int(msg_id)) # message_id_str = builtins.str(message_id_hex) # test.verify(message_id_str in message_extracted, "message ID is verified") # if param != None: # test.verify(param in message_extracted, "parameters are verified") test.endSection() def main(): utils.tstStart(__file__) startApplication(config.AUT_NAME) for index in range (3): utils.waitForGUI(5) #0x9900 hd_simulator.cmd_send_serial_hd_data(serial = config.CLOUD_SYNC["HD_VERSION_SERIAL"][index]) # verify_log(epoch, ZERO, registration_request, THREE, HD_VERSION_SERIAL, DG_VERSION_SERIAL = None, SW_VERSION) epoch, ZERO, registration_request, THREE, HD_VERSION_SERIAL, DG_VERSION_SERIAL, SW_VERSION = retrive_log_data() test.log(str(epoch, ZERO, registration_request, THREE, HD_VERSION_SERIAL, DG_VERSION_SERIAL, SW_VERSION)) # utils.waitForGUI(5) # #0x8800 # dg_simulator.cmd_send_serial_dg_data(serial = 'DG1234567890123') # #verify_log(msg_id = MsgIds.MSG_ID_DG_SERIAL_NUMBER.value, msg = "Serial", param = ['DG1234567890123']) utils.waitForGUI(5) #0x8800 dg_simulator.cmd_send_serial_dg_data(serial = config.CLOUD_SYNC["DG_VERSION_SERIAL"][index]) verify_log() # utils.waitForGUI(5) # #0x9900 # hd_simulator.cmd_send_serial_hd_data(serial = 'HD1234567890123') # #verify_log(msg_id = MsgIds.MSG_ID_HD_SERIAL_NUMBER.value, msg = "Serial", param = ['HD1234567890123']) # # #0x8800 # dg_simulator.cmd_send_serial_dg_data(serial = 'DG1234567890123') # #verify_log(msg_id = MsgIds.MSG_ID_DG_SERIAL_NUMBER.value, msg = "Serial", param = ['DG1234567890123']) utils.waitForGUI(5) #0x9900 hd_simulator.cmd_send_serial_hd_data(serial = config.CLOUD_SYNC["HD_VERSION_SERIAL"][index]) #verify_log(msg_id = MsgIds.MSG_ID_HD_SERIAL_NUMBER.value, msg = "Serial", param = ['HD1234567890123']) #0x8800 dg_simulator.cmd_send_serial_dg_data(serial = config.CLOUD_SYNC["DG_VERSION_SERIAL"][index]) #verify_log(msg_id = MsgIds.MSG_ID_DG_SERIAL_NUMBER.value, msg = "Serial", param = ['DG1234567890123']) utils.tstDone()