add plan task and handle 304 response

Author: eliezer-zup

runtime-manager-action/script.py

@@ -136,13 +136,18 @@ def runtime_manager_run_self_hosted_deploy(request_data: dict, manifest: dict):
         url=url, body=request_data, headers=HEADERS, timeout=TIMEOUT
     )
 
-    if response.ok:
-        # Parse the response and extract relevant data
-        response_data = response.json()
-        print(f"> Deploy successfully started:\n{json.dumps(response_data, indent=4)}")
 
-        # Save the response to the output log
-        save_output(response_data)
+    if response.status_code == 201:
+        try:
+            # Parse the response and extract relevant data
+            response_data = response.json()
+            print(f"> Deploy successfully started:\n{json.dumps(response_data, indent=4)}")
+
+            # Save the response to the output log
+            save_output(response_data)
+        except json.JSONDecodeError:
+            print("> Error: Failed to parse JSON response.")
+            exit(1)
     else:
         print(
             f"> Error: Failed to start self-hosted deploy run. Status: {response.status_code}"

script.py

@@ -44,7 +44,6 @@ def rollback_deploy_run(run_action: RunAction):
     run_action("runtime-rollback-action")
     run_tasks("rollback-output.log", run_action)
 
-
 def run_tasks(file_tasks: str, run_action: RunAction):
     with open(file_tasks, "r") as file:
         data = json.loads(file.read().replace("'", '"'))
@@ -53,6 +52,7 @@ def run_tasks(file_tasks: str, run_action: RunAction):
         IAC_SELF_HOSTED=lambda **i: run_action("runtime-iac-action", **i),
         DEPLOY_SELF_HOSTED=lambda **i: run_action("runtime-deploy-action", **i),
         DESTROY_SELF_HOSTED=lambda **i: run_action("runtime-destroy-action", **i),
+        PLAN=lambda **i: run_action("runtime-unified-action", **i),
         UNIFIED_IAC=lambda **i: run_action("runtime-unified-action", **i),
         UNIFIED_DEPLOY=lambda **i: run_action("runtime-unified-action", **i),
         UNIFIED_DESTROY=lambda **i: run_action("runtime-unified-action", **i),
@@ -63,8 +63,10 @@ def run_tasks(file_tasks: str, run_action: RunAction):
         runner = task_runners.get(task_type)
         if "UNIFIED" in task_type:
             runner and runner(run_id=data.get("runId"))
+        elif "PLAN" == task_type:
+            runner and runner(run_id=data.get("runId"))
         else:
-            runner and runner(run_task_id=t["runTaskId"])
+            runner and runner(run_task_id=t["runTaskId"]) 
 
 
 def run(metadata):