Handle succeeded pods in readiness wait

aiopslab/service/kubectl.py

@@ -86,6 +86,19 @@ def get_deployment(self, name: str, namespace: str):
  """Fetch the deployment configuration."""
  return self.apps_v1_api.read_namespaced_deployment(name, namespace)
 
+ @staticmethod
+ def _pod_is_ready_or_succeeded(pod):
+ """Return True when a pod is ready or has completed successfully."""
+ status = getattr(pod, "status", None)
+ if getattr(status, "phase", None) == "Succeeded":
+ return True
+
+ container_statuses = getattr(status, "container_statuses", None)
+ return bool(container_statuses) and all(
+ getattr(container_status, "ready", False)
+ for container_status in container_statuses
+ )
+
  def wait_for_ready(self, namespace, sleep=2, max_wait=300):
  """Wait for all pods in a namespace to be in a Ready state before proceeding."""
 
@@ -102,8 +115,7 @@ def wait_for_ready(self, namespace, sleep=2, max_wait=300):
  if pod_list.items:
  ready_pods = [
  pod for pod in pod_list.items
- if pod.status.container_statuses and
- all(cs.ready for cs in pod.status.container_statuses)
+ if self._pod_is_ready_or_succeeded(pod)
  ]
 
  if len(ready_pods) == len(pod_list.items):

tests/service/test_kubectl.py

@@ -0,0 +1,47 @@
+from types import SimpleNamespace
+import unittest
+
+from aiopslab.service.kubectl import KubeCtl
+
+
+def _pod(phase, ready_values=None):
+ container_statuses = None
+ if ready_values is not None:
+ container_statuses = [
+ SimpleNamespace(ready=ready) for ready in ready_values
+ ]
+
+ return SimpleNamespace(
+ status=SimpleNamespace(
+ phase=phase,
+ container_statuses=container_statuses,
+ )
+ )
+
+
+class PodReadinessTest(unittest.TestCase):
+
+ def test_running_pod_with_ready_containers_satisfies_readiness(self):
+ pod = _pod("Running", [True, True])
+
+ self.assertTrue(KubeCtl._pod_is_ready_or_succeeded(pod))
+
+ def test_succeeded_cleanup_pod_satisfies_readiness(self):
+ pod = _pod("Succeeded", [False])
+
+ self.assertTrue(KubeCtl._pod_is_ready_or_succeeded(pod))
+
+ def test_running_pod_with_unready_container_blocks_readiness(self):
+ pod = _pod("Running", [True, False])
+
+ self.assertFalse(KubeCtl._pod_is_ready_or_succeeded(pod))
+
+ def test_pending_pod_without_container_statuses_blocks_readiness(self):
+ pod = _pod("Pending")
+
+ self.assertFalse(KubeCtl._pod_is_ready_or_succeeded(pod))
+
+ def test_failed_pod_with_unready_container_blocks_readiness(self):
+ pod = _pod("Failed", [False])
+
+ self.assertFalse(KubeCtl._pod_is_ready_or_succeeded(pod))