Merge pull request #487 from AutomationSolutionz/exit_tc_2

[Add] Test case exit action is added

Author: Muntasib-creator

Framework/Built_In_Automation/Sequential_Actions/action_declarations/common.py

@@ -12,6 +12,7 @@
     { "name": "delete shared variables",                     "function": "delete_all_shared_variables",           "screenshot": "none" },
     { "name": "settings",                                    "function": "sequential_actions_settings",           "screenshot": "none" },
     { "name": "step exit",                                   "function": "step_exit",                             "screenshot": "none" },
+    { "name": "testcase exit",                               "function": "testcase_exit",                             "screenshot": "none" },
     { "name": "save time",                                   "function": "Save_Current_Time",                     "screenshot": "none" },
     { "name": "create or append list into list",             "function": "insert_list_into_another_list",         "screenshot": "none" },
     { "name": "validate order",                              "function": "validate_list_order",                   "screenshot": "none" },

Framework/Built_In_Automation/Sequential_Actions/common_functions.py

@@ -546,26 +546,56 @@ def step_exit(data_set):
     try:
         action_value = ""
         msg = ""
-        for row in data_set:
-            if row[0] == "step exit" and row[1] == "action":
-                action_value = row[2]
-            elif row[0].strip().lower() == "message":
-                msg = row[2]
+        for left, mid, right in data_set:
+            if left.lower().strip() == "step exit":
+                action_value = right.strip().lower()
+            elif left.strip().lower() == "message":
+                msg = right
+
     except:
         return CommonUtil.Exception_Handler(sys.exc_info())
 
-    if (
-        action_value in failed_tag_list
-    ):  # Convert user specified pass/fail into standard result
-        if msg:
-            CommonUtil.ExecLog(sModuleInfo, f'{msg}', 3)
+    if msg:
+        CommonUtil.ExecLog(sModuleInfo, msg, 3 if "fail" in action_value else 1)
+    if "fail" in action_value:
         return "zeuz_failed"
-    elif action_value in skipped_tag_list:
-        return "skipped"
-    elif action_value in passed_tag_list:
-        if msg:
-            CommonUtil.ExecLog(sModuleInfo, f'{msg}', 1)
+    elif "pass" in action_value:
+        return "passed"
+    else:
+        CommonUtil.ExecLog(sModuleInfo, "Step Result action has invalid VALUE", 3)
+        return "zeuz_failed"
+
+
+def testcase_exit(data_set):
+    """ Exits a Test Step wtih passed/failed in the standard format, when the user specifies it in the step data """
+
+    sModuleInfo = inspect.currentframe().f_code.co_name + " : " + MODULE_NAME
+
+    try:
+        action_value = ""
+        for left, mid, right in data_set:
+            if left.lower().strip() == "testcase exit":
+                action_value = right.strip().lower()
+    except:
+        return CommonUtil.Exception_Handler(sys.exc_info())
+
+
+    if "skip" in action_value:
+        CommonUtil.testcase_exit = "Skipped"
+        CommonUtil.ExecLog(sModuleInfo, "The testcase will be force Skipped", 2)
+        return "passed"
+    elif "fail" in action_value:
+        CommonUtil.ExecLog(sModuleInfo, "The testcase will be force Failed", 2)
+        CommonUtil.testcase_exit = "Failed"
+        return "zeuz_failed"
+    elif "pass" in action_value:
+        CommonUtil.ExecLog(sModuleInfo, "The testcase will be force Passed", 2)
+        CommonUtil.testcase_exit = "Passed"
         return "passed"
+    elif "block" in action_value:
+        CommonUtil.ExecLog(sModuleInfo, "The testcase will be force Blocked", 2)
+        CommonUtil.testcase_exit = "Blocked"
+        return "zeuz_failed"
     else:
         CommonUtil.ExecLog(sModuleInfo, "Step Result action has invalid VALUE", 3)
         return "zeuz_failed"
@@ -4608,7 +4638,7 @@ def upload_attachment_to_testcase(data_set):
             CommonUtil.ExecLog(sModuleInfo, "Please insert attachment path ", 3)
             return "zeuz_failed"
         headers = RequestFormatter.add_api_key_to_headers({})
-        res = RequestFormatter.request("post", 
+        res = RequestFormatter.request("post",
             RequestFormatter.form_uri("test_case_file_upload/"),
             files={"file": open(var_path, 'rb')},
             data={"file_upload_tc": var_id},
@@ -4706,7 +4736,7 @@ def upload_attachment_to_step(data_set):
             CommonUtil.ExecLog(sModuleInfo, "Please insert attachment path ", 3)
             return "zeuz_failed"
         headers = RequestFormatter.add_api_key_to_headers({})
-        res = RequestFormatter.request("post", 
+        res = RequestFormatter.request("post",
             RequestFormatter.form_uri("step_file_upload/"),
             files={"file": open(var_path,'rb')},
             data={"file_upload_step": var_id, },
@@ -4800,7 +4830,7 @@ def upload_attachment_to_global(data_set):
             return "zeuz_failed"
 
         headers = RequestFormatter.add_api_key_to_headers({})
-        res = RequestFormatter.request("post", 
+        res = RequestFormatter.request("post",
             RequestFormatter.form_uri("global_file_upload/"),
             files={"file": open(var_path,'rb')},
             verify=False,
@@ -5905,7 +5935,7 @@ def data_store_read(data_set):
         headers = RequestFormatter.add_api_key_to_headers({})
         headers['headers']['content-type'] = 'application/json'
         headers['headers']['X-API-KEY'] = ConfigModule.get_config_value("Authentication", "api-key")
-        res = RequestFormatter.request("get", 
+        res = RequestFormatter.request("get",
             RequestFormatter.form_uri('data_store/data_store/custom_operation/'),
             params=json.dumps(params),
             verify=False,
@@ -5938,19 +5968,19 @@ def data_store_get_data(data_set):
                 columns = right.strip()
             if 'get data' in left.lower().strip():
                 variable_name = right.strip()
-        
+
         if table_name == None or variable_name == None:
             CommonUtil.ExecLog(sModuleInfo, f"Table name and Variable Name both needs to be provided", 3)
             return "zeuz failed"
-        
+
         headers = RequestFormatter.add_api_key_to_headers({})
         headers['headers']['content-type'] = 'application/json'
         headers['headers']['X-API-KEY'] = ConfigModule.get_config_value("Authentication", "api-key")
         params = dict()
         params['table'] = table_name
         if columns:
             params['columns'] = columns
-        res = RequestFormatter.request("get", 
+        res = RequestFormatter.request("get",
             RequestFormatter.form_uri('data_store/get_datastore/'),
             params=params,
             verify=False,
@@ -5977,17 +6007,17 @@ def data_store_get_stats(data_set):
                 table_name = right.strip()
             if 'stats' in left.lower().strip():
                 variable_name = right.strip()
-        
+
         if table_name == None or variable_name == None:
             CommonUtil.ExecLog(sModuleInfo, f"Table name and Variable Name both needs to be provided", 3)
             return "zeuz_failed"
-        
+
         headers = RequestFormatter.add_api_key_to_headers({})
         headers['headers']['content-type'] = 'application/json'
         headers['headers']['X-API-KEY'] = ConfigModule.get_config_value("Authentication", "api-key")
         params = dict()
         params['table'] = table_name
-        res = RequestFormatter.request("get", 
+        res = RequestFormatter.request("get",
             RequestFormatter.form_uri('data_store/get_datastore_stats/'),
             params=params,
             verify=False,
@@ -6191,7 +6221,7 @@ def data_store_insert(data_set):
         headers['headers']['content-type'] = 'application/json'
         headers['headers']['X-API-KEY'] = ConfigModule.get_config_value("Authentication", "api-key")
 
-        res = RequestFormatter.request("post", 
+        res = RequestFormatter.request("post",
             RequestFormatter.form_uri('data_store/data_store/data_store_list/'),
             data=json.dumps(data),
             verify=False,
@@ -6289,25 +6319,25 @@ def xml_to_json(data_set):
                 filepath = Path(CommonUtil.path_parser(filepath))
             if "xml to json" in left:
                 json_var_name = right.strip()
-        
+
         if None in (filepath,json_var_name):
             CommonUtil.ExecLog(sModuleInfo, "Please specify both filename and json variable name", 3)
 
         if filepath != None and filepath.is_file():
             try:
                 with open(filepath) as xml_file:
                     data_dict = xmltodict.parse(xml_file.read())
-                result = sr.Set_Shared_Variables(json_var_name, data_dict)  
+                result = sr.Set_Shared_Variables(json_var_name, data_dict)
                 return result
             except:
                 CommonUtil.ExecLog(sModuleInfo, "Couldn't read and convert the xml file", 3)
                 return "zeuz_failed"
-                
 
-            
+
+
         else:
             CommonUtil.ExecLog(sModuleInfo, "Specified file couldn't be found or downloaded from attachment", 3)
-            return "zeuz_failed" 
+            return "zeuz_failed"
 
     except:
         return CommonUtil.Exception_Handler(sys.exc_info())
@@ -6384,7 +6414,7 @@ def connect_to_S3(data_set):
                 credentials_source = right.strip()
             if "file path" in left:
                 cred_file_path = right.strip()
-        
+
         if credentials_source == "default":
             try:
                 s3_client = boto3.client('s3')
@@ -6438,7 +6468,7 @@ def upload_to_S3(data_set):
             if "s3 resource variable":
                 s3_resource_var = right.strip()
 
-        
+
         if None in (bucket_name, file_name):
             CommonUtil.ExecLog(sModuleInfo, "Bucket, file and key names should be provided", 3)
 
@@ -6470,7 +6500,7 @@ def connect_to_bigquery_client(data_set):
         if left.strip().lower() == 'connect to bigquery client':
             client_var_name = right.strip()
 
-    
+
     os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = cred_path
 
     try:
@@ -6489,7 +6519,7 @@ def execute_bigquery_query(data_set):
         output variable         | input parameter   | output variable name
         execute bigquery query  | common action     | client variable name
     """
-        
+
     from google.cloud import bigquery
 
     sModuleInfo = inspect.currentframe().f_code.co_name + " : " + MODULE_NAME
@@ -6505,7 +6535,7 @@ def execute_bigquery_query(data_set):
             client_var_name = right.strip()
         if left.strip().lower() == 'output variable':
             output_var_name = right.strip()
-    
+
     if None in (query,client_var_name,output_var_name):
         CommonUtil.ExecLog(sModuleInfo, "Incorrect Dataset", 3)
         return "zeuz_failed"
@@ -6527,7 +6557,7 @@ def connect_to_google_service_account(data_set):
           credentials path                      | input parameter   | path to credentails json file
           connect to google service client      | common action     | client variable name
     """
-    
+
     sModuleInfo = inspect.currentframe().f_code.co_name + " : " + MODULE_NAME
     CommonUtil.ExecLog(sModuleInfo, "Actions involving the Google service account may not function correctly without a virtual environment.", 2)
     cred_path = None
@@ -6538,7 +6568,7 @@ def connect_to_google_service_account(data_set):
         if left.strip().lower() == 'connect to google service client':
             client_var_name = right.strip()
 
-    try:    
+    try:
         from google.cloud import storage
         client = storage.Client.from_service_account_json(json_credentials_path=cred_path)
         sr.Set_Shared_Variables(client_var_name, client)
@@ -6552,7 +6582,7 @@ def upload_to_google_storage_bucket(data_set):
     """
     data_set:
         filepath                            | input parameter   | filepath
-        bucket                              | input parameter   | bucket name 
+        bucket                              | input parameter   | bucket name
         upload to google storage bucket     | common action     | client variable name
     """
 
@@ -6569,7 +6599,7 @@ def upload_to_google_storage_bucket(data_set):
             client_var_name = right.strip()
         if left.strip().lower() == 'bucket':
             bucket = right.strip()
-    
+
     if None in (filepath,client_var_name,bucket):
         CommonUtil.ExecLog(sModuleInfo, "Incorrect Dataset", 3)
         return "zeuz_failed"
@@ -6583,7 +6613,7 @@ def upload_to_google_storage_bucket(data_set):
         return "passed"
     except:
         return CommonUtil.Exception_Handler(sys.exc_info())
-    
+
 @logger
 def text_to_speech(data_set):
     """
@@ -6611,7 +6641,7 @@ def text_to_speech(data_set):
                 language = right.strip()
             if "accent" in left:
                 top_level_domain = right.strip()
-                
+
         if None in (dialogue_data, speaker, output_filename):
             CommonUtil.ExecLog(sModuleInfo, "Incorrect dataset", 3)
             return "zeuz_failed"

Framework/Built_In_Automation/Sequential_Actions/sequential_actions.py

@@ -1120,7 +1120,7 @@ def Run_Sequential_Actions(
                 table.add_column("Field", justify="left", max_width=20)
                 table.add_column("Sub-field", justify="left", max_width=15)
                 table.add_column("Value", justify="left", max_width=35)
-                
+
                 data_set_to_print = sr.Hide_Secretive_Text(text_value=data_set)
                 for row in data_set_to_print:
                     table.add_row(*row, style=_color)
@@ -1329,9 +1329,9 @@ def Run_Sequential_Actions(
                 # If middle column = action, call action handler
                 elif "action" in action_name:  # Must be last, since it's a single word that also exists in other action types
                     result = Action_Handler(data_set, row)  # Pass data set, and action_name to action handler
-                    if row[0].lower().strip() == "step exit":
+                    if row[0].lower().strip() in ("step exit", "testcase exit"):
                         global step_exit_fail_called, step_exit_pass_called
-                        CommonUtil.ExecLog(sModuleInfo, "Step Exit called. Stopping Test Step.", 1)
+                        CommonUtil.ExecLog(sModuleInfo, f"{row[0].lower().strip()} Exit called. Stopping Test Step.", 1)
                         if result == "zeuz_failed":
                             step_exit_fail_called = True
                         else:
@@ -2319,7 +2319,7 @@ def Action_Handler(_data_set, action_row, _bypass_bug=True):
                 + python_location \
                 +"\n3.Go to this link and download python https://www.python.org/ftp/python/3.11.4/python-3.11.4-amd64.exe"\
                 +"\n4.During installation, give uncheck 'for all user' and check 'Add Python to Path'. This is very important."\
-                +"\n5.Relaunch zeuz node_cli.py" 
+                +"\n5.Relaunch zeuz node_cli.py"
             CommonUtil.ExecLog(sModuleInfo, error_msg, 3)
             return "zeuz_failed" 
 

Framework/MainDriverApi.py

@@ -605,6 +605,9 @@ def run_all_test_steps_in_a_test_case(
             if StepSeq in CommonUtil.disabled_step or not all_step_info[StepSeq - 1]['step_enable']:
                 CommonUtil.ExecLog(sModuleInfo, "STEP-%s is disabled" % StepSeq, 2)
                 sStepResult = "skipped"
+            elif CommonUtil.testcase_exit:
+                CommonUtil.ExecLog(sModuleInfo, "STEP-%s is skipped" % StepSeq, 2)
+                sStepResult = "skipped"
             else:
                 sStepResult = call_driver_function_of_test_step(
                     sModuleInfo,
@@ -698,7 +701,10 @@ def run_all_test_steps_in_a_test_case(
 
 # from the returned step results, it finds out the test case result
 def calculate_test_case_result(sModuleInfo, TestCaseID, run_id, sTestStepResultList, testcase_info):
-    if "BLOCKED" in sTestStepResultList:
+    if CommonUtil.testcase_exit:
+        CommonUtil.ExecLog(sModuleInfo, f"Test Case {CommonUtil.testcase_exit}", 1)
+        return CommonUtil.testcase_exit
+    elif "BLOCKED" in sTestStepResultList:
         CommonUtil.ExecLog(sModuleInfo, "Test Case Blocked", 3)
         return "Blocked"
     elif "CANCELLED" in sTestStepResultList or "Cancelled" in sTestStepResultList:
@@ -1929,6 +1935,7 @@ def main(device_dict, all_run_id_info):
                     }
                     set_device_info_according_to_user_order(device_order, device_dict, test_case_no, test_case_name, user_info_object, Userid, run_id=run_id)
                     CommonUtil.disabled_step = []
+                    CommonUtil.testcase_exit = ""
 
                     # Download test case and step attachments
                     download_attachments(testcase_info)

Framework/Utilities/CommonUtil.py

@@ -143,6 +143,7 @@
 run_cancel = ""
 run_cancelled = False
 disabled_step = []  # 1 based indexing
+testcase_exit = ""
 max_char = 0
 compare_action_varnames = {"left":"Left", "right":"Right"}    # for labelling left and right variable names of compare action