Index: tst_cloud_sync_device_registration/test.py =================================================================== diff -u -r043c2695cbc8e90cfe6ebde197300de94a80fbdb -r15cca0a5aa54862e11fa0d416a1d8ae1ade38e8a --- tst_cloud_sync_device_registration/test.py (.../test.py) (revision 043c2695cbc8e90cfe6ebde197300de94a80fbdb) +++ tst_cloud_sync_device_registration/test.py (.../test.py) (revision 15cca0a5aa54862e11fa0d416a1d8ae1ade38e8a) @@ -24,8 +24,10 @@ from configuration import config from builtins import int as pyInt from builtins import str as pyStr +from time import * import time from datetime import datetime +from datetime import * hd_simulator = HDSimulator() dg_simulator = DGSimulator() @@ -35,102 +37,116 @@ MESSAGE_ID_RESPONSE = "1002" EXPECTED_SOFTWARE_VERSION = "06300216" -def retrive_log_data_first_line(): +def retrive_input_buf_file_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_input_buf_file() - try: - with open(file_name) as inp_buf: ##remove - contents = inp_buf.read() - with open(file_name,mode = 'r') as inp_buf: - contents = csv.reader(inp_buf) - try: - for row in reversed(list(contents)): - return (row[0],row[3],row[4],row[5],row[6],row[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.") + Method to get all the parameter values from input buf file + @return: actual_epoch, actual_message_id_response, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_request + """ + + first_input_buf_line = utility.retrive_log_data(readline_count = 2) + actual_epoch = first_input_buf_line[0][0] + actual_message_id_response = first_input_buf_line[0][3] + actual_parameters = first_input_buf_line[0][4] + actual_hd_version_serial = first_input_buf_line[0][5] + actual_dg_version_serial = first_input_buf_line[0][6] + actual_sw_version = first_input_buf_line[0][7] + message_id_request = first_input_buf_line[1][3] + return (actual_epoch, actual_message_id_response, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_request) -def retrive_log_data_second_line(): - """ - Method to read latest input buf file and getting log data from it - @return: message id - """ - file_name = utility.get_extracted_input_buf_file() - try: - with open(file_name,mode = 'r') as inp_buf: - last_line_two = inp_buf.readline(-2) - last_line_two = last_line_two.strip() - last_line_two = last_line_two.split(",") - return (last_line_two[3]) - except: - test.fail("Input buf file is not created or it is not created based on standard naming format.") - def epoch_value_conversion(epoch): """ Method to convert input buf files epoch value into date and time format @param epoch: (int) Date and Time in epoch format @return: epoch value in date and time format """ - date_time_format = time.strftime('%Y-%m-%d %H:%M', time.localtime(epoch)) ###change - return date_time_format + x = strftime('%Y-%m-%d %H:%M:%S', localtime(epoch)) + + #x = datetime.now() + y = x + timedelta(0,5) + z = x - timedelta(0,5) + test.log(str(x.time())) + test.log(str(y.time())) + test.log(str(z.time())) + + +# date_time_format = time.strftime('%Y-%m-%d %H:%M', time.localtime(epoch)) +# date_time_1 = date_time_format + datetime.timedelta(0,5) +# date_time_2 = date_time_format + datetime.timedelta(0,-5) +# test.log(str(date_time_1)) +# test.log(str(date_time_2)) +# return date_time_format def verify_inp_buf_data(actual_epoch, actual_message_id_response, actual_parameters, actual_sw_version, message_id_request): """ Method to verify and compare epoch value, number of parameters and software version from input buf file @param actual_epoch: (str) Date and Time in epoch format @param actual_message_id_response: (str) message id response from inp_buf file @param actual_parameters: (str) number of parameters - @param actual_sw_version: (int) software version - @param message_id_request: (int) message id request from inp_buf file + @param actual_sw_version: (str) software version + @param message_id_request: (str) message id request from inp_buf file @return: None """ - EXPECTED_EPOCH = epoch_value_conversion(time.time()) + + #if epoch_value isIn range() + #EXPECTED_EPOCH = epoch_value_conversion(1657604640) 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(message_id_request, MESSAGE_ID_REQUEST, "The message id request should be: "+MESSAGE_ID_REQUEST) test.compare(actual_message_id_response, 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) + +def verify_hd_version_registration(): + """ + Method to verify all the parameters generating in input buf file after passing hd version serial + """ -def main(): - utils.tstStart(__file__) - startApplication(config.AUT_NAME) - test.startSection("Verification of input buf file data for HD version serial") for value in range (3): hd_simulator.cmd_send_serial_hd_data(serial = config.CLOUD_SYNC["HD_VERSION_SERIAL"][value]) utils.waitForGUI(2) - actual_epoch, actual_message_id_response, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version = retrive_log_data_first_line() - message_id_request = retrive_log_data_second_line() + actual_epoch, actual_message_id_response, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_request = retrive_input_buf_file_data() test.compare(actual_hd_version_serial, config.CLOUD_SYNC["HD_VERSION_SERIAL"][value], "The HD serial version should be: "+config.CLOUD_SYNC["HD_VERSION_SERIAL"][value]) verify_inp_buf_data(actual_epoch, actual_message_id_response, actual_parameters, actual_sw_version, message_id_request) test.endSection() - + +def verify_dg_version_registration(): + """ + Method to verify all the parameters generating in input buf file after passing dg version serial + """ + test.startSection("Verification of input buf file data for DG version serial") for value in range (3): dg_simulator.cmd_send_serial_dg_data(serial = config.CLOUD_SYNC["DG_VERSION_SERIAL"][value]) utils.waitForGUI(2) - actual_epoch, actual_message_id_response, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version = retrive_log_data_first_line() - message_id_request = retrive_log_data_second_line() + actual_epoch, actual_message_id_response, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_request = retrive_input_buf_file_data() test.compare(actual_dg_version_serial, config.CLOUD_SYNC["DG_VERSION_SERIAL"][value], "The DG serial version should be: "+config.CLOUD_SYNC["DG_VERSION_SERIAL"][value]) verify_inp_buf_data(actual_epoch, actual_message_id_response, actual_parameters, actual_sw_version, message_id_request) - test.endSection() + test.endSection() +def verify_hd_dg_version_registration(): + """ + Method to verify all the parameters generating in input buf file after passing both hd and dg version serial + """ + test.startSection("Verification of input buf file data for HD and DG version serial") for value in range (3): hd_simulator.cmd_send_serial_hd_data(serial = config.CLOUD_SYNC["HD_VERSION_SERIAL"][value]) dg_simulator.cmd_send_serial_dg_data(serial = config.CLOUD_SYNC["DG_VERSION_SERIAL"][value]) utils.waitForGUI(2) - actual_epoch, actual_message_id_response, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version = retrive_log_data_first_line() - message_id_request = retrive_log_data_second_line() + actual_epoch, actual_message_id_response, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_request = retrive_input_buf_file_data() test.compare(actual_hd_version_serial, config.CLOUD_SYNC["HD_VERSION_SERIAL"][value], "The HD serial version should be: "+config.CLOUD_SYNC["HD_VERSION_SERIAL"][value]) test.compare(actual_dg_version_serial, config.CLOUD_SYNC["DG_VERSION_SERIAL"][value], "The DG serial version should be: "+config.CLOUD_SYNC["DG_VERSION_SERIAL"][value]) verify_inp_buf_data(actual_epoch, actual_message_id_response, actual_parameters, actual_sw_version, message_id_request) test.endSection() - + +def main(): + utils.tstStart(__file__) + startApplication(config.AUT_NAME) + + verify_hd_version_registration() + verify_dg_version_registration() + verify_hd_dg_version_registration() + utils.tstDone()