Handle cases where container_statuses is None

Author: ForeverWintr

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/pod_manager.py

@@ -19,7 +19,6 @@
 from __future__ import annotations
 
 import enum
-import itertools
 import json
 import math
 import time
@@ -119,9 +118,12 @@ def get_xcom_sidecar_container_resources(self) -> str | None:
 def get_container_status(pod: V1Pod, container_name: str) -> V1ContainerStatus | None:
     """Retrieve container status."""
     if pod and pod.status:
-        container_statuses = itertools.chain(
-            pod.status.container_statuses, pod.status.init_container_statuses
-        )
+        container_statuses = []
+        if pod.status.container_statuses:
+            container_statuses.extend(pod.status.container_statuses)
+        if pod.status.init_container_statuses:
+            container_statuses.extend(pod.status.init_container_statuses)
+
     else:
         container_statuses = None
 

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_pod_manager.py

@@ -38,7 +38,6 @@
     container_is_running,
     container_is_succeeded,
     container_is_terminated,
-    get_container_status,
 )
 from airflow.utils.timezone import utc