Index: tst_cloud_sync_device_registration/test.py =================================================================== diff -u -r16b51bc5a664cfdac954366e8b371e2ebed1ccf2 -r043c2695cbc8e90cfe6ebde197300de94a80fbdb --- tst_cloud_sync_device_registration/test.py (.../test.py) (revision 16b51bc5a664cfdac954366e8b371e2ebed1ccf2) +++ tst_cloud_sync_device_registration/test.py (.../test.py) (revision 043c2695cbc8e90cfe6ebde197300de94a80fbdb) @@ -35,112 +35,102 @@ MESSAGE_ID_RESPONSE = "1002" EXPECTED_SOFTWARE_VERSION = "06300216" -def retrive_log_data_second_line(): +def retrive_log_data_first_line(): """ Method to read latest input buf file and getting log data from it - @return: message id - """ - - file_name = utility.get_extracted_file_inp_buf() - 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 retrive_log_data_first_line(): - """ - 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() - + file_name = utility.get_extracted_input_buf_file() try: - with open(file_name) as f: - contents = f.read() - test.log(str(contents)) - with open(file_name,mode = 'r') as f: - contents = csv.reader(f) + 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)): - test.log(str(row)) 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.") + test.fail("Log file is not created or log file is not created based on standard log naming format.") -def system_date_time(): +def retrive_log_data_second_line(): """ - Method to get systems date and time - @return: date and time + Method to read latest input buf file and getting log data from it + @return: message id """ - system_date_time = time.strftime('%Y-%m-%d %H:%M', time.localtime(time.time())) - return system_date_time + 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)) + date_time_format = time.strftime('%Y-%m-%d %H:%M', time.localtime(epoch)) ###change return date_time_format -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_response): +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 @return: None """ - test.startSection("Verification of cloud sync latest input buf file data") - EXPECTED_EPOCH = system_date_time() + EXPECTED_EPOCH = epoch_value_conversion(time.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(message_id_response, MESSAGE_ID_REQUEST, "The message id request should be: "+MESSAGE_ID_REQUEST) - test.compare(actual_message_id, MESSAGE_ID_RESPONSE, "The message id response should be: "+MESSAGE_ID_RESPONSE) + 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) - test.endSection() 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(5) - actual_epoch, actual_message_id, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version = retrive_log_data_first_line() - message_id_response = retrive_log_data_second_line() + 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() 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, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_response) - - utils.waitForGUI(5) + verify_inp_buf_data(actual_epoch, actual_message_id_response, actual_parameters, actual_sw_version, message_id_request) + test.endSection() + 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(5) - actual_epoch, actual_message_id, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version = retrive_log_data_first_line() - message_id_response = retrive_log_data_second_line() + 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() 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, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_response) - - utils.waitForGUI(5) - + verify_inp_buf_data(actual_epoch, actual_message_id_response, actual_parameters, actual_sw_version, message_id_request) + test.endSection() + + 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(5) - actual_epoch, actual_message_id, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version = retrive_log_data_first_line() - message_id_response = retrive_log_data_second_line() + 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() 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, actual_parameters, actual_hd_version_serial, actual_dg_version_serial, actual_sw_version, message_id_response) - - utils.waitForGUI(5) + verify_inp_buf_data(actual_epoch, actual_message_id_response, actual_parameters, actual_sw_version, message_id_request) + test.endSection() - utils.tstDone()