Index: tst_cloud_sync_device_registration/test.py =================================================================== diff -u -r7ea06f6694a6c28c6c071cd8b7b8c75f72fb41ac -rf17d0b3a1bd0e3dec8ead6d11be0a96514a1d875 --- tst_cloud_sync_device_registration/test.py (.../test.py) (revision 7ea06f6694a6c28c6c071cd8b7b8c75f72fb41ac) +++ tst_cloud_sync_device_registration/test.py (.../test.py) (revision f17d0b3a1bd0e3dec8ead6d11be0a96514a1d875) @@ -16,106 +16,109 @@ # NOTE: # This test verifies the logs for the messages provided import builtins -import csv from dialin.ui import HDSimulator from dialin.ui import DGSimulator from dialin.ui import utils from configuration import utility from configuration import config -from builtins import int as pyInt -from builtins import str as pyStr import time -from datetime import datetime hd_simulator = HDSimulator() dg_simulator = DGSimulator() +EXPECTED_NUMBER_OF_PARAMETERS = "3" +MESSAGE_ID_REQUEST = "1001" +MESSAGE_ID_RESPONSE = "1002" +EXPECTED_SOFTWARE_VERSION = "06300216" def retrive_log_data(): + """ + Method to read latest input buf file and getting log data from it + @return: epoch value, message id, number of parameters, HD serial version, DG serial version, software version + """ 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("input buf data is corrupted") + with open(file_name,mode = 'r') as inp_buf: + last_line_one = inp_buf.readline(-1) + last_line_one = last_line_one.strip() + last_line_one = last_line_one.split(",") + test.log(str(last_line_one)) + + last_line_two = inp_buf.readline(-2) + last_line_two = last_line_two.strip() + last_line_two = last_line_two.split(",") + + test.log(str(last_line_two)) + return (last_line_one[0], last_line_one[3], last_line_one[4], last_line_one[5], last_line_one[6], last_line_one[7], last_line_two[3]) except: test.fail("Input buf file is not created or it is not created based on standard naming format.") -def system_epoch_value(): - - time_formatted = time.strftime('%Y-%m-%d %H:%M', time.localtime(time.time())) - test.log(str(time_formatted)) - return time_formatted +def system_date_time(): + """ + Method to get systems date and time + @return: date and time + """ + system_date_time = time.strftime('%Y-%m-%d %H:%M', time.localtime(time.time())) + return system_date_time def epoch_value_conversion(epoch): - - time_formatted_one = time.strftime('%Y-%m-%d %H:%M', time.localtime(epoch)) - test.log(str(time_formatted_one)) - return time_formatted_one + """ + Method to convert input buf files epoch value into date and time format + @return: epoch value in date and time format + """ + date_time_format = time.strftime('%Y-%m-%d %H:%M', time.localtime(epoch)) + return date_time_format -def verify_inp_buf_data(): - - test.startSection("Verification of cloud sync latest inuput buf file data") - EPOCH_ONE = system_epoch_value() - epoch,event, ZERO, registration_request, THREE, HD_VERSION_SERIAL, DG_VERSION_SERIAL, SW_VERSION = retrive_log_data() - epoch = builtins.int(epoch) - epoch = epoch_value_conversion(epoch) - test.compare(EPOCH_ONE, epoch) -# test.compare(event) -# test.compare(ZERO) -# test.compare(registration_request) -# test.compare(THREE) -# test.compare(config.CLOUD_SYNC[index], HD_VERSION_SERIAL) -# test.compare(config.CLOUD_SYNC[index], DG_VERSION_SERIAL) -# test.compare(SW_VERSION) +def verify_inp_buf_data(actual_epoch, actual_message_id, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_request): + """ + Method to verify and compare epoch value, number of parameters and software version from input buf file + @return: None + """ + test.startSection("Verification of cloud sync latest input buf file data") + EXPECTED_EPOCH = system_date_time() + actual_epoch = builtins.int(actual_epoch) + actual_epoch = epoch_value_conversion(actual_epoch) + test.compare(actual_epoch, EXPECTED_EPOCH, "The epoch should be: "+EXPECTED_EPOCH) + test.compare(actual_message_id, MESSAGE_ID_REQUEST, "The message id request should be: "+MESSAGE_ID_REQUEST) + test.compare(message_id_request, MESSAGE_ID_RESPONSE, "The message id response should be: "+MESSAGE_ID_RESPONSE) + test.compare(actual_parameters, EXPECTED_NUMBER_OF_PARAMETERS, "The number of parameters should be: "+EXPECTED_NUMBER_OF_PARAMETERS) + test.compare(actual_sw_version, EXPECTED_SOFTWARE_VERSION, "The software version should be: "+EXPECTED_SOFTWARE_VERSION) test.endSection() - - def main(): utils.tstStart(__file__) startApplication(config.AUT_NAME) for index in range (3): hd_simulator.cmd_send_serial_hd_data(serial = config.CLOUD_SYNC["HD_VERSION_SERIAL"][index]) utils.waitForGUI(5) - verify_inp_buf_data() + actual_epoch, actual_message_id, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_request = retrive_log_data() + test.log(str(message_id_request)) + test.compare(actual_hd_version_serial, config.CLOUD_SYNC["HD_VERSION_SERIAL"][index], "The HD serial version should be: "+config.CLOUD_SYNC["HD_VERSION_SERIAL"][index]) + verify_inp_buf_data(actual_epoch, actual_message_id, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_request) - - #utils.waitForGUI(5) -# -# for index in range (3): -# dg_simulator.cmd_send_serial_dg_data(serial = config.CLOUD_SYNC["DG_VERSION_SERIAL"][index]) -# utils.waitForGUI(5) -# epoch, event, ZERO, registration_request, THREE, HD_VERSION_SERIAL, DG_VERSION_SERIAL, SW_VERSION = retrive_log_data() -# test.log(str(epoch)) -# test.log(str(event)) -# test.log(str(ZERO)) -# test.log(str(registration_request)) -# test.log(str(THREE)) -# test.log(str(HD_VERSION_SERIAL)) -# test.log(str(DG_VERSION_SERIAL)) -# test.log(str(SW_VERSION)) -# -# utils.waitForGUI(5) -# for index in range (3): -# hd_simulator.cmd_send_serial_hd_data(serial = config.CLOUD_SYNC["HD_VERSION_SERIAL"][index]) -# dg_simulator.cmd_send_serial_dg_data(serial = config.CLOUD_SYNC["DG_VERSION_SERIAL"][index]) -# utils.waitForGUI(5) -# epoch,event, ZERO, registration_request, THREE, HD_VERSION_SERIAL, DG_VERSION_SERIAL, SW_VERSION = retrive_log_data() -# test.log(str(epoch)) -# test.log(str(event)) -# test.log(str(ZERO)) -# test.log(str(registration_request)) -# test.log(str(THREE)) -# test.log(str(HD_VERSION_SERIAL)) -# test.log(str(DG_VERSION_SERIAL)) -# test.log(str(SW_VERSION)) + + for index in range (3): + dg_simulator.cmd_send_serial_dg_data(serial = config.CLOUD_SYNC["DG_VERSION_SERIAL"][index]) + utils.waitForGUI(5) + actual_epoch, actual_message_id, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_request = retrive_log_data() + test.compare(actual_dg_version_serial, config.CLOUD_SYNC["DG_VERSION_SERIAL"][index], "The DG serial version should be: "+config.CLOUD_SYNC["DG_VERSION_SERIAL"][index]) + verify_inp_buf_data(actual_epoch, actual_message_id, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_request) + #utils.waitForGUI(5) + + for index in range (3): + hd_simulator.cmd_send_serial_hd_data(serial = config.CLOUD_SYNC["HD_VERSION_SERIAL"][index]) + dg_simulator.cmd_send_serial_dg_data(serial = config.CLOUD_SYNC["DG_VERSION_SERIAL"][index]) + utils.waitForGUI(5) + actual_epoch, actual_message_id, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_request = retrive_log_data() + test.compare(actual_hd_version_serial, config.CLOUD_SYNC["HD_VERSION_SERIAL"][index], "The HD serial version should be: "+config.CLOUD_SYNC["HD_VERSION_SERIAL"][index]) + test.compare(actual_dg_version_serial, config.CLOUD_SYNC["DG_VERSION_SERIAL"][index], "The DG serial version should be: "+config.CLOUD_SYNC["DG_VERSION_SERIAL"][index]) + verify_inp_buf_data(actual_epoch, actual_message_id, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_request) + + #utils.waitForGUI(5) + + utils.tstDone()