fixed issue #14 : Improve handling of jobs being deleted

src/pygeoapi_k8s_manager/finalizer.py

@@ -166,6 +166,9 @@ def handle_job_ended_event(
         )
         LOGGER.debug(f"Found '{len(pods.items)}' pod{'s' if len(pods.items) > 1 else ''} for job '{job.metadata.name}'")
         if len(pods.items) < 1:
+            if job.metadata.deletion_timestamp:
+                LOGGER.debug(f"Job '{job.metadata.name}' is being deleted. No pods found.")
+                return
             LOGGER.error(f"Could not get pod for job '{job.metadata.name}'")
             return
         pod = pods.items[0]

tests/test_finalizer_deleted_job.py

@@ -0,0 +1,35 @@
+
+import datetime
+from unittest.mock import MagicMock, patch
+
+import pytest
+from kubernetes.client import V1PodList
+
+from pygeoapi_k8s_manager.finalizer import KubernetesFinalizerController
+
+@pytest.fixture()
+def finalizer():
+    return KubernetesFinalizerController("")
+
+def test_handle_job_ended_event_handles_deleted_job(finalizer):
+    finalizer.is_upload_logs_to_s3 = False
+    k8s_core_api = MagicMock()
+    k8s_core_api.list_namespaced_pod.return_value = V1PodList(items=[])
+
+    test_job = MagicMock()
+    test_job.metadata.name = "test-job"
+    test_job.metadata.deletion_timestamp = datetime.datetime.now()
+
+    with (
+        patch(
+            "pygeoapi_k8s_manager.finalizer.KubernetesFinalizerController.upload_logs_to_s3"
+        ) as mocked_upload_logs_to_s3,
+        patch("pygeoapi_k8s_manager.util.get_logs_for_pod") as mocked_get_logs_for_pod,
+    ):
+        finalizer.handle_job_ended_event(
+            k8s_core_api=k8s_core_api,
+            job=test_job,
+        )
+
+    mocked_get_logs_for_pod.assert_not_called()
+    mocked_upload_logs_to_s3.assert_not_called()