Index: tst_cloud_sync - treatment_screen/0 =================================================================== diff -u --- tst_cloud_sync - treatment_screen/0 (revision 0) +++ tst_cloud_sync - treatment_screen/0 (revision e72ee29e03cef1663b40e2e1915c440df696daff) @@ -0,0 +1 @@ +1.pem \ No newline at end of file Index: tst_cloud_sync - treatment_screen/1 =================================================================== diff -u --- tst_cloud_sync - treatment_screen/1 (revision 0) +++ tst_cloud_sync - treatment_screen/1 (revision e72ee29e03cef1663b40e2e1915c440df696daff) @@ -0,0 +1 @@ +2.pem \ No newline at end of file Index: tst_cloud_sync - treatment_screen/2 =================================================================== diff -u --- tst_cloud_sync - treatment_screen/2 (revision 0) +++ tst_cloud_sync - treatment_screen/2 (revision e72ee29e03cef1663b40e2e1915c440df696daff) @@ -0,0 +1 @@ +3.pem \ No newline at end of file Index: tst_cloud_sync - treatment_screen/test.py =================================================================== diff -u -rf7bfeef1136eab7daee4c21ec1e6ee97f3412503 -re72ee29e03cef1663b40e2e1915c440df696daff --- tst_cloud_sync - treatment_screen/test.py (.../test.py) (revision f7bfeef1136eab7daee4c21ec1e6ee97f3412503) +++ tst_cloud_sync - treatment_screen/test.py (.../test.py) (revision e72ee29e03cef1663b40e2e1915c440df696daff) @@ -17,6 +17,7 @@ import csv +import os from dialin.ui.hd_simulator import HDSimulator from dialin.ui import utils from configuration import utility @@ -26,232 +27,102 @@ hd_simulator = HDSimulator() -LOCATION = '/home/denali/Projects/application/resources/settings/Messages/Unhandled.conf' +CLOUD_CREDENTIALS_LOCATION = '/home/denali/Desktop/cloudsync/credentials' +CLOUD_SYNC_LOG_LOCATION = '/home/denali/Desktop/sd-card/cloudsync' +MESSAGE_1 = '1639391827,1,0,2008,1,Tx_code' -DUMMY_FLOAT_VALUE_1 = 1.250 -DUMMY_FLOAT_VALUE_2 = 5.855 -DUMMY_INT_VALUE_1 = 25 -DUMMY_INT_VALUE_2 = 32 -DUMMY_UNSIGNED_INT_VALUE = 5 -UNHANDLED_MESSAGE = { - 'MESSAGE_1' : "\n[0xBB00]\nHD_custom_data_verification_1\nU32=unsigned_32bit\nU32=argument_2\n", - 'MESSAGE_2' : "\n[0xAA00]\nHD_custom_data_verification_2\nF32=float_argument\nS32=argument_2\n" - } -ERROR_MESSAGE = { - "0x1300" : "Incorrect data length (9 of 32) for received Message with ID '0x1300'", - "0x0900" : "Incorrect data length (9 of 20) for received Message with ID '0x0900'" - } +PEM_FILES = ['1.pem', '2.pem', '3.pem' ] - -def verify_log(api_content): +def get_cloud_sync_input_file(): """ - This function is capable to verify the log data from sd-card - - @param api_content - (string) combination of message id, message text and arguments separated with '\n' - @param message_extracted - (list) list contains extracted data (ack_req, ack_bak, message, message_id). + 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 """ - - utils.waitForGUI(5) - test.log(str(api_content)) - message_extracted = utility.get_current_log_details(message_ack = True, message_text = '~'+api_content) + 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 - parameter_index = 0 - #loop to remove default arguments present in API, here it is '?' - while('?' in message_extracted[parameter_index]): - message_extracted[parameter_index].remove('?') - return message_extracted - - -def convert_message_id_from_unhandled_file_format_to_standared_format(message_id): +def get_cloud_sync_output_file(): """ - This function is capable to remove square bracket and to convert message id into firmware support format. - - @param message_id - (string) message id from unhandled.conf file. - @return expected_meassage_id - (string) message id in the standared format. + 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 """ - formated_meassage_id = message_id.replace('[', "") - formated_meassage_id = formated_meassage_id.replace(']', "") - if formated_meassage_id[-2:] == '00' and len(formated_meassage_id) == 6: - expected_meassage_id = formated_meassage_id.replace("00", "") - return expected_meassage_id - else: - return formated_meassage_id + 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 get_error_message_from_log(): - """ - This function is capable to verify the error logs from sd-card +def append_cloudsync_credentials_file(): - @param api_content - (string) combination of message id, message text and arguments separated with '\n' - @param expected_api_argument - (list) expected arguments for API. - """ - utils.waitForGUI(5) - count = 0 - file_name = utility.get_extracted_error_file() try: - with open(file_name, 'r') as csv_file: - try: - for row in reversed(list(csv.reader(csv_file))): - if row[0].isalpha(): - pass - else: - if count == 30: - test.fail("handler unable to find message text from error file.") - break - row_length = sum(1 for values in row) - try: - if "'" in row[1]: - return row[1] - if "Unhandled Message ID (HD)" in row[1]: - return True - except: - pass - count+=1 + os.makedirs(CLOUD_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 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 error log data is corrupted") + test.fail("application log data is corrupted") except: - test.fail("err log file is not created or log file is not created based on standard log naming format.") - - -def append_unhandled_configuration_file(): + 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 custom API's on unhandled.conf folder. - - unhandled.conf location = '/home/denali/Projects/application/resources/settings/Messages/Unhandled.conf' - - Through this method filereader act as a handler and it will write custom unhandled messages on Unhandled.conf - NB: it is a single time callable method should only use before starting the application. it is mandatory - to use deappend_unhandled_configuration_file() at the end of the testcase. + 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(UNHANDLED_MESSAGE['MESSAGE_1']) - filereader.write(UNHANDLED_MESSAGE['MESSAGE_2']) + filereader.write(UNHANDLED_MESSAGE[message]) utils.waitForGUI(5) -def remove_unhandled_configuration_file(): - """ - This function is capable to remove custom API's which is written during method "append_unhandled_configuration_file()". + - unhandled.conf location = '/home/denali/Projects/application/resources/settings/Messages/Unhandled.conf' - NB: it is a single time callable method should use only after calling "append_unhandled_configuration_file()" method. - """ - count = 0 - for value in UNHANDLED_MESSAGE['MESSAGE_2']: - if value == '\n': - count+=1 - for value in UNHANDLED_MESSAGE['MESSAGE_1']: - if value == '\n': - count+=1 - with open(LOCATION, "r") as filereader: - lines = filereader.readlines() - with open(LOCATION, "w") as filereader: - filereader.writelines(lines[:-count]) - - def main(): - append_unhandled_configuration_file() + + append_cloudsync_credentials_file() utils.tstStart(__file__) - startApplication(config.AUT_NAME) + startApplication(config.AUT_NAME) + utils.waitForGUI(1.5) - #checking if the message is not in the unhandled (err) - test.startSection("verify the message is not in unhandled (err)") - hd_simulator.cmd_send_hd_general_response(message_id = 275, accepted = 1, reason = 1) - unhandled_status = get_error_message_from_log() - test.verify(unhandled_status == True, "message is not in unhandled.conf file. verified error message from .err file") + cloud_out_log = get_cloud_sync_output_file() + set_data_in_cloud_sync_output_file(cloud_out_log, MESSAGE_1) + retrive_log_data - #checking if the message is not in the unhandled (err) - hd_simulator.cmd_send_hd_general_response(message_id = 276, accepted = 1, reason = 1) - unhandled_status = get_error_message_from_log() - test.verify(unhandled_status == True, "message is not in unhandled.conf file. verified error message from .err file") - test.endSection() - - #checking the id exists and parameters passed is not correct regarding the definition in the conf. (err) - test.startSection("verify the id exists and parameters passed is not correct regarding the definition in the conf. (err)") - #0x1300 - hd_simulator.cmd_send_hd_general_response(message_id = 19, accepted = 1, reason = 1) - error_message = get_error_message_from_log() - test.compare(error_message, ERROR_MESSAGE["0x1300"], "error code should be verified in log file") - - #checking the id exists and parameters passed is not correct regarding the definition in the conf. (err) - #0x0900 - hd_simulator.cmd_send_hd_general_response(message_id = 9, accepted = 0, reason = 0) - error_message = get_error_message_from_log() - test.compare(error_message, ERROR_MESSAGE["0x0900"], "error code should be verified in log file") - test.endSection() - - - #checking the number of parameters are correct but the type is not correct. (err) - test.startSection("verify the number of parameters are correct but the type is not correct. (err)") - #0x0A00 - payload = float_to_bytearray(DUMMY_FLOAT_VALUE_1) #expected negative scenario verification - hd_simulator.cmd_send_hd_general_response(message_id = 10, accepted = 1, reason = 1, is_pure_data = False, has_parameters = True, parameters_payload = payload) - log_data = verify_log("HD_RTC_Epoch_Data") - test.verify(DUMMY_FLOAT_VALUE_1 not in log_data, "parameters type should show deviation in negative scenario.") - - #checking the number of parameters are correct but the type is not correct. (err) - #0x7D00 - payload = integer_to_bytearray(DUMMY_INT_VALUE_1) - payload += float_to_bytearray(DUMMY_FLOAT_VALUE_1) - payload += float_to_bytearray(DUMMY_FLOAT_VALUE_2) - payload += float_to_bytearray(DUMMY_FLOAT_VALUE_1) - payload += float_to_bytearray(DUMMY_INT_VALUE_2) #expected negative scenario verification - hd_simulator.cmd_send_hd_general_response(message_id = 125, accepted = 1, reason = 1, is_pure_data = False, has_parameters = True, parameters_payload = payload) - - log_data = verify_log("HD_Alarm_Information") - test.verify([DUMMY_INT_VALUE_1, DUMMY_FLOAT_VALUE_1, DUMMY_FLOAT_VALUE_2, DUMMY_FLOAT_VALUE_1, DUMMY_INT_VALUE_1] not in log_data, "parameters type should show deviation in negative scenario.") - test.endSection() - - - #checking everything is passed correctly. (log) - test.startSection("verify the log is passed correctly.") - #0xBB00 - messageID_index = 1 - message_index = 2 - payload = integer_to_bytearray(DUMMY_INT_VALUE_1) - payload += integer_to_bytearray(DUMMY_INT_VALUE_2) - hd_simulator.cmd_send_hd_general_response(message_id = 187, accepted = 1, reason = 1, is_pure_data = False, has_parameters = True, parameters_payload = payload) - - - unhandled_message = UNHANDLED_MESSAGE['MESSAGE_1'].split('\n') - #unhandled_message[message_index] is the expected message in unhandled.conf after API call - #unhandled_message[messageID_index] is the expected message id in unhandled.conf after API call - log_data = verify_log(unhandled_message[message_index]) - message_id = convert_message_id_from_unhandled_file_format_to_standared_format(unhandled_message[messageID_index]) - test.verify(message_id.lower() in log_data, "message id should be "+str(unhandled_message[messageID_index])) - test.verify([DUMMY_INT_VALUE_1, DUMMY_INT_VALUE_2] in log_data, "API arguments should be "+str([DUMMY_INT_VALUE_1, DUMMY_INT_VALUE_2])) - test.verify(config.ACK_REQ_STATUS in log_data, "ack request is verified") - test.verify(config.ACK_BAK_STATUS in log_data, "ack back is verified") - - - #checking everything is passed correctly. (log) - #0xAA00 - payload = float_to_bytearray(DUMMY_FLOAT_VALUE_1) - payload += unsigned_integer_to_bytearray(DUMMY_UNSIGNED_INT_VALUE) - hd_simulator.cmd_send_hd_general_response(message_id = 170, accepted = 1, reason = 0, is_pure_data = False, has_parameters = True, parameters_payload = payload) - - unhandled_message =UNHANDLED_MESSAGE['MESSAGE_2'].split('\n') - #unhandled_message[2] is the expected message in unhandled.conf after API call - #unhandled_message[messageID_index] is the expected message id in unhandled.conf after API call - - log_data = verify_log(unhandled_message[message_index]) - message_id = convert_message_id_from_unhandled_file_format_to_standared_format(unhandled_message[messageID_index]) - test.verify(message_id.lower() in log_data, "message id should be "+str(unhandled_message[messageID_index])) - test.verify([DUMMY_FLOAT_VALUE_1, DUMMY_UNSIGNED_INT_VALUE] in log_data, "API arguments should be "+str([DUMMY_FLOAT_VALUE_1, DUMMY_UNSIGNED_INT_VALUE, 0])) - test.verify(config.ACK_REQ_STATUS in log_data, "ack request is verified") - test.verify(config.ACK_BAK_STATUS in log_data, "ack back is verified") - test.endSection() - - remove_unhandled_configuration_file() - utils.tstDone() \ No newline at end of file