add pod exit staus helper and use in test_base

Author: willcl-ark

src/warnet/k8s.py

@@ -36,6 +36,19 @@ def get_mission(mission: str) -> list[V1PodList]:
     return crew
 
 
+def get_pod_exit_status(pod_name):
+    try:
+        sclient = get_static_client()
+        pod = sclient.read_namespaced_pod(name=pod_name, namespace=get_default_namespace())
+        for container_status in pod.status.container_statuses:
+            if container_status.state.terminated:
+                return container_status.state.terminated.exit_code
+        return None
+    except client.ApiException as e:
+        print(f"Exception when calling CoreV1Api->read_namespaced_pod: {e}")
+        return None
+
+
 def get_edges() -> any:
     sclient = get_static_client()
     configmap = sclient.read_namespaced_config_map(name="edges", namespace="warnet")

test/test_base.py

@@ -12,6 +12,7 @@
 
 from warnet import SRC_DIR
 from warnet.control import get_active_scenarios
+from warnet.k8s import get_pod_exit_status
 from warnet.network import _connected as network_connected
 from warnet.network import _status as network_status
 
@@ -132,7 +133,12 @@ def check_scenarios():
             scns = get_active_scenarios()
             if len(scns) == 0:
                 return True
-            return all(s["status"] == "succeeded" for s in scns)
+            for s in scns:
+                exit_status = get_pod_exit_status(s)
+                self.log.debug(f"Scenario {s} exited with code {exit_status}")
+                if exit_status != 0:
+                    return False
+            return True
 
         self.wait_for_predicate(check_scenarios)