From ddaf6fe21adcbb5e7da105bcf6817cf46f54ddaf Mon Sep 17 00:00:00 2001
From: andyfcx <dreamcastle.andy@gmail.com>
Date: Tue, 9 May 2023 17:04:01 +0800
Subject: [PATCH 1/5] feat(pod): expose active_deadline_seconds to be set

---
 airflow/providers/cncf/kubernetes/operators/pod.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/airflow/providers/cncf/kubernetes/operators/pod.py b/airflow/providers/cncf/kubernetes/operators/pod.py
index 1bfa2ba03479d..752475349a76f 100644
--- a/airflow/providers/cncf/kubernetes/operators/pod.py
+++ b/airflow/providers/cncf/kubernetes/operators/pod.py
@@ -175,6 +175,7 @@ class KubernetesPodOperator(BaseOperator):
         during the next try. If False, always create a new pod for each try.
     :param labels: labels to apply to the Pod. (templated)
     :param startup_timeout_seconds: timeout in seconds to startup the pod.
+    :param active_deadline_seconds: the active_deadline_seconds from V1PodSpec.
     :param get_logs: get the stdout of the base container as logs of the tasks.
     :param image_pull_policy: Specify a policy to cache or always pull an image.
     :param annotations: non-identifying metadata you can attach to the Pod.
@@ -267,6 +268,7 @@ def __init__(
         labels: dict | None = None,
         reattach_on_restart: bool = True,
         startup_timeout_seconds: int = 120,
+        active_deadline_seconds: int | Nont = None,
         get_logs: bool = True,
         image_pull_policy: str | None = None,
         annotations: dict | None = None,
@@ -321,6 +323,7 @@ def __init__(
         self.arguments = arguments or []
         self.labels = labels or {}
         self.startup_timeout_seconds = startup_timeout_seconds
+        self.active_deadline_seconds = active_deadline_seconds
         self.env_vars = convert_env_vars(env_vars) if env_vars else []
         if pod_runtime_info_envs:
             self.env_vars.extend([convert_pod_runtime_info_env(p) for p in pod_runtime_info_envs])
@@ -798,6 +801,7 @@ def build_pod_request_obj(self, context: Context | None = None) -> k8s.V1Pod:
                 affinity=self.affinity,
                 tolerations=self.tolerations,
                 init_containers=self.init_containers,
+                active_deadline_seconds=self.active_deadline_seconds,
                 containers=[
                     k8s.V1Container(
                         image=self.image,

From c64f783614cd9ced43dd41e43794dba90fb6d066 Mon Sep 17 00:00:00 2001
From: andyfcx <dreamcastle.andy@gmail.com>
Date: Tue, 9 May 2023 17:04:01 +0800
Subject: [PATCH 2/5] feat(pod): expose active_deadline_seconds to be set

---
 airflow/providers/cncf/kubernetes/operators/pod.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/airflow/providers/cncf/kubernetes/operators/pod.py b/airflow/providers/cncf/kubernetes/operators/pod.py
index 1bfa2ba03479d..c5b26bfc9a631 100644
--- a/airflow/providers/cncf/kubernetes/operators/pod.py
+++ b/airflow/providers/cncf/kubernetes/operators/pod.py
@@ -175,6 +175,7 @@ class KubernetesPodOperator(BaseOperator):
         during the next try. If False, always create a new pod for each try.
     :param labels: labels to apply to the Pod. (templated)
     :param startup_timeout_seconds: timeout in seconds to startup the pod.
+    :param active_deadline_seconds: the active_deadline_seconds from V1PodSpec.
     :param get_logs: get the stdout of the base container as logs of the tasks.
     :param image_pull_policy: Specify a policy to cache or always pull an image.
     :param annotations: non-identifying metadata you can attach to the Pod.
@@ -267,6 +268,7 @@ def __init__(
         labels: dict | None = None,
         reattach_on_restart: bool = True,
         startup_timeout_seconds: int = 120,
+        active_deadline_seconds: int | None = None,
         get_logs: bool = True,
         image_pull_policy: str | None = None,
         annotations: dict | None = None,
@@ -321,6 +323,7 @@ def __init__(
         self.arguments = arguments or []
         self.labels = labels or {}
         self.startup_timeout_seconds = startup_timeout_seconds
+        self.active_deadline_seconds = active_deadline_seconds
         self.env_vars = convert_env_vars(env_vars) if env_vars else []
         if pod_runtime_info_envs:
             self.env_vars.extend([convert_pod_runtime_info_env(p) for p in pod_runtime_info_envs])
@@ -798,6 +801,7 @@ def build_pod_request_obj(self, context: Context | None = None) -> k8s.V1Pod:
                 affinity=self.affinity,
                 tolerations=self.tolerations,
                 init_containers=self.init_containers,
+                active_deadline_seconds=self.active_deadline_seconds,
                 containers=[
                     k8s.V1Container(
                         image=self.image,

From 3fa6ef147387215ca5ed86c943af4b68b02d3913 Mon Sep 17 00:00:00 2001
From: andyfcx <dreamcastle.andy@gmail.com>
Date: Tue, 9 May 2023 23:25:11 +0800
Subject: [PATCH 3/5] wip(test): k8sPodOperator with k8s api client

---
 tests/kubernetes/test_active_deadline.py | 62 ++++++++++++++++++++++++
 1 file changed, 62 insertions(+)
 create mode 100644 tests/kubernetes/test_active_deadline.py

diff --git a/tests/kubernetes/test_active_deadline.py b/tests/kubernetes/test_active_deadline.py
new file mode 100644
index 0000000000000..99b57f6f478e8
--- /dev/null
+++ b/tests/kubernetes/test_active_deadline.py
@@ -0,0 +1,62 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import time
+import pytest
+from kubernetes import client, config
+from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
+
+
+# Create a test class for the KubernetesPodOperator
+class TestKubernetesPodOperator:
+    @pytest.mark.parametrize("active_deadline_seconds", [10])
+    def test_kubernetes_pod_operator_active_deadline_seconds(self, active_deadline_seconds):
+        task_id = "test_task"
+        image = "busybox"
+        cmds = ["sh", "-c", "echo 'hello world' && sleep 20"]
+        namespace = "default"
+
+        operator = KubernetesPodOperator(
+            task_id=task_id,
+            active_deadline_seconds=active_deadline_seconds,
+            image=image,
+            cmds=cmds,
+            namespace=namespace
+        )
+
+        operator.execute(context={})
+
+        pod = operator.find_pod()
+        pod_name = pod.metadata.name
+        # namespace = pod.metadata.namespace
+
+        k8s_client = client.CoreV1Api()
+        config.load_kube_config()
+
+        time.sleep(active_deadline_seconds)
+
+        pod_status = k8s_client.read_namespaced_pod_status(name=pod_name, namespace=namespace)
+
+        # TODO: Check the state is correct
+
+        phase = pod_status.status.phase
+        conditions = pod_status.status.conditions
+
+        assert pod.status.phase == "Succeeded"
+

From a6105054a74a3d865a53920867dbc799dafa7f58 Mon Sep 17 00:00:00 2001
From: andyfcx <dreamcastle.andy@gmail.com>
Date: Thu, 11 May 2023 00:02:22 +0800
Subject: [PATCH 4/5] wip(test): assert phase Fail

---
 tests/kubernetes/test_active_deadline.py | 7 +------
 1 file changed, 1 insertion(+), 6 deletions(-)

diff --git a/tests/kubernetes/test_active_deadline.py b/tests/kubernetes/test_active_deadline.py
index 99b57f6f478e8..7d3fb745a06f3 100644
--- a/tests/kubernetes/test_active_deadline.py
+++ b/tests/kubernetes/test_active_deadline.py
@@ -44,7 +44,6 @@ def test_kubernetes_pod_operator_active_deadline_seconds(self, active_deadline_s
 
         pod = operator.find_pod()
         pod_name = pod.metadata.name
-        # namespace = pod.metadata.namespace
 
         k8s_client = client.CoreV1Api()
         config.load_kube_config()
@@ -52,11 +51,7 @@ def test_kubernetes_pod_operator_active_deadline_seconds(self, active_deadline_s
         time.sleep(active_deadline_seconds)
 
         pod_status = k8s_client.read_namespaced_pod_status(name=pod_name, namespace=namespace)
-
-        # TODO: Check the state is correct
-
         phase = pod_status.status.phase
-        conditions = pod_status.status.conditions
 
-        assert pod.status.phase == "Succeeded"
+        assert phase == "Failed"
 

From d4b72de199f5b60b79d66000cc550f1fce709121 Mon Sep 17 00:00:00 2001
From: andyfcx <dreamcastle.andy@gmail.com>
Date: Tue, 6 Jun 2023 15:42:02 +0800
Subject: [PATCH 5/5] test: TestKubernetesPodOperator

---
 .../test_kubernetes_pod_operator.py           | 39 ++++++++++++++++++-
 1 file changed, 37 insertions(+), 2 deletions(-)

diff --git a/kubernetes_tests/test_kubernetes_pod_operator.py b/kubernetes_tests/test_kubernetes_pod_operator.py
index 4afd4507fda58..8de8540748dd5 100644
--- a/kubernetes_tests/test_kubernetes_pod_operator.py
+++ b/kubernetes_tests/test_kubernetes_pod_operator.py
@@ -21,6 +21,7 @@
 import os
 import shutil
 import sys
+import time
 from copy import copy
 from tempfile import NamedTemporaryFile
 from unittest import mock
@@ -43,6 +44,7 @@
 from airflow.utils.context import Context
 from airflow.utils.types import DagRunType
 from airflow.version import version as airflow_version
+from kubernetes_tests.test_base import BaseK8STest
 
 HOOK_CLASS = "airflow.providers.cncf.kubernetes.operators.pod.KubernetesHook"
 POD_MANAGER_CLASS = "airflow.providers.cncf.kubernetes.utils.pod_manager.PodManager"
@@ -896,8 +898,6 @@ def test_pod_template_file(
         # todo: This isn't really a system test
         await_xcom_sidecar_container_start_mock.return_value = None
         hook_mock.return_value.is_in_cluster = False
-        hook_mock.return_value.get_xcom_sidecar_container_image.return_value = None
-        hook_mock.return_value.get_xcom_sidecar_container_resources.return_value = None
         hook_mock.return_value.get_connection.return_value = Connection(conn_id="kubernetes_default")
         extract_xcom_mock.return_value = "{}"
         path = sys.path[0] + "/tests/kubernetes/pod.yaml"
@@ -1327,3 +1327,38 @@ def __getattr__(self, name):
         task.render_template_fields(context=context)
     assert "password" in caplog.text
     assert "secretpassword" not in caplog.text
+
+
+class TestKubernetesPodOperator(BaseK8STest):
+    @pytest.mark.parametrize("active_deadline_seconds", [10])
+    def test_kubernetes_pod_operator_active_deadline_seconds(self, active_deadline_seconds):
+        task_id = "test_task"
+        image = "busybox"
+        cmds = ["sh", "-c", "echo 'hello world' && sleep 20"]
+        namespace = "default"
+
+        operator = KubernetesPodOperator(
+            task_id=task_id,
+            active_deadline_seconds=active_deadline_seconds,
+            image=image,
+            cmds=cmds,
+            namespace=namespace
+        )
+
+        operator.execute(context={})
+
+        pod = operator.find_pod()
+        pod_name = pod.metadata.name
+        # namespace = pod.metadata.namespace
+
+        k8s_client = client.CoreV1Api()
+        config.load_kube_config()
+
+        time.sleep(active_deadline_seconds)
+
+        pod_status = k8s_client.read_namespaced_pod_status(name=pod_name, namespace=namespace)
+
+        phase = pod_status.status.phase
+        conditions = pod_status.status.conditions
+
+        assert phase == "Failed"
