This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-0-test by this push:
new 0a2f8cb6549 [v3-0-test] Use env var connections instead of mocks in
kube system tests (#54261) (#54320)
0a2f8cb6549 is described below
commit 0a2f8cb6549bae14f072b5d19ef70173529ef9b6
Author: Amogh Desai <[email protected]>
AuthorDate: Mon Aug 11 02:55:16 2025 +0530
[v3-0-test] Use env var connections instead of mocks in kube system tests
(#54261) (#54320)
* Use env var connections instead of mocks in kube system tests
* adding right pytest plugings
* adding right pytest plugings
* adding right pytest plugings
(cherry picked from commit d410915)
---
.../tests/kubernetes_tests/conftest.py | 17 ++++
.../test_kubernetes_pod_operator.py | 96 +++++++++++-----------
2 files changed, 65 insertions(+), 48 deletions(-)
diff --git a/kubernetes-tests/tests/kubernetes_tests/conftest.py
b/kubernetes-tests/tests/kubernetes_tests/conftest.py
index 1a78383f4b0..06ad4ff9ca2 100644
--- a/kubernetes-tests/tests/kubernetes_tests/conftest.py
+++ b/kubernetes-tests/tests/kubernetes_tests/conftest.py
@@ -38,3 +38,20 @@ def pod_template() -> Path:
@pytest.fixture
def basic_pod_template() -> Path:
return (DATA_FILES_DIRECTORY / "basic_pod.yaml").resolve(strict=True)
+
+
[email protected]
+def create_connection_without_db(monkeypatch):
+ """
+ Fixture to create connections for tests without using the database.
+
+ This fixture uses monkeypatch to set the appropriate
AIRFLOW_CONN_{conn_id} environment variable.
+ """
+
+ def _create_conn(connection, session=None):
+ """Create connection using environment variable."""
+
+ env_var_name = f"AIRFLOW_CONN_{connection.conn_id.upper()}"
+ monkeypatch.setenv(env_var_name, connection.as_json())
+
+ return _create_conn
diff --git
a/kubernetes-tests/tests/kubernetes_tests/test_kubernetes_pod_operator.py
b/kubernetes-tests/tests/kubernetes_tests/test_kubernetes_pod_operator.py
index c334e858124..2ce9e6ae1ef 100644
--- a/kubernetes-tests/tests/kubernetes_tests/test_kubernetes_pod_operator.py
+++ b/kubernetes-tests/tests/kubernetes_tests/test_kubernetes_pod_operator.py
@@ -101,14 +101,7 @@ def test_label(request):
return label[-63:]
[email protected]
-def mock_get_connection():
- with mock.patch(f"{HOOK_CLASS}.get_connection",
return_value=Connection(conn_id="kubernetes_default")):
- yield
-
-
@pytest.mark.execution_timeout(180)
[email protected]("mock_get_connection")
class TestKubernetesPodOperatorSystem:
@pytest.fixture(autouse=True)
def setup_tests(self, test_label):
@@ -162,12 +155,21 @@ class TestKubernetesPodOperatorSystem:
client = hook.core_v1_client
client.delete_collection_namespaced_pod(namespace="default",
grace_period_seconds=0)
+ @pytest.fixture(autouse=True)
+ def setup_connections(self, create_connection_without_db):
+ """Create kubernetes_default connection"""
+ connection = Connection(
+ conn_id="kubernetes_default",
+ conn_type="kubernetes",
+ )
+ create_connection_without_db(connection)
+
def _get_labels_selector(self) -> str | None:
if not self.labels:
return None
return ",".join([f"{key}={value}" for key, value in
enumerate(self.labels)])
- def test_do_xcom_push_defaults_false(self, kubeconfig_path,
mock_get_connection, tmp_path):
+ def test_do_xcom_push_defaults_false(self, kubeconfig_path, tmp_path):
new_config_path = tmp_path / "kube_config.cfg"
shutil.copy(kubeconfig_path, new_config_path)
k = KubernetesPodOperator(
@@ -183,7 +185,7 @@ class TestKubernetesPodOperatorSystem:
)
assert not k.do_xcom_push
- def test_config_path_move(self, kubeconfig_path, mock_get_connection,
tmp_path):
+ def test_config_path_move(self, kubeconfig_path, tmp_path):
new_config_path = tmp_path / "kube_config.cfg"
shutil.copy(kubeconfig_path, new_config_path)
@@ -205,7 +207,7 @@ class TestKubernetesPodOperatorSystem:
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
assert actual_pod == expected_pod
- def test_working_pod(self, mock_get_connection):
+ def test_working_pod(self):
k = KubernetesPodOperator(
namespace="default",
image="ubuntu:16.04",
@@ -222,7 +224,7 @@ class TestKubernetesPodOperatorSystem:
assert self.expected_pod["spec"] == actual_pod["spec"]
assert self.expected_pod["metadata"]["labels"] ==
actual_pod["metadata"]["labels"]
- def test_skip_cleanup(self, mock_get_connection):
+ def test_skip_cleanup(self):
k = KubernetesPodOperator(
namespace="unknown",
image="ubuntu:16.04",
@@ -237,7 +239,7 @@ class TestKubernetesPodOperatorSystem:
with pytest.raises(ApiException):
k.execute(context)
- def test_delete_operator_pod(self, mock_get_connection):
+ def test_delete_operator_pod(self):
k = KubernetesPodOperator(
namespace="default",
image="ubuntu:16.04",
@@ -255,7 +257,7 @@ class TestKubernetesPodOperatorSystem:
assert self.expected_pod["spec"] == actual_pod["spec"]
assert self.expected_pod["metadata"]["labels"] ==
actual_pod["metadata"]["labels"]
- def test_skip_on_specified_exit_code(self, mock_get_connection):
+ def test_skip_on_specified_exit_code(self):
k = KubernetesPodOperator(
namespace="default",
image="ubuntu:16.04",
@@ -271,7 +273,7 @@ class TestKubernetesPodOperatorSystem:
with pytest.raises(AirflowSkipException):
k.execute(context)
- def test_already_checked_on_success(self, mock_get_connection):
+ def test_already_checked_on_success(self):
"""
When ``on_finish_action="keep_pod"``, pod should have 'already_checked'
label, whether pod is successful or not.
@@ -293,7 +295,7 @@ class TestKubernetesPodOperatorSystem:
actual_pod = self.api_client.sanitize_for_serialization(actual_pod)
assert actual_pod["metadata"]["labels"]["already_checked"] == "True"
- def test_already_checked_on_failure(self, mock_get_connection):
+ def test_already_checked_on_failure(self):
"""
When ``on_finish_action="keep_pod"``, pod should have 'already_checked'
label, whether pod is successful or not.
@@ -318,7 +320,7 @@ class TestKubernetesPodOperatorSystem:
assert status["state"]["terminated"]["reason"] == "Error"
assert actual_pod["metadata"]["labels"]["already_checked"] == "True"
- def test_pod_hostnetwork(self, mock_get_connection):
+ def test_pod_hostnetwork(self):
k = KubernetesPodOperator(
namespace="default",
image="ubuntu:16.04",
@@ -337,7 +339,7 @@ class TestKubernetesPodOperatorSystem:
assert self.expected_pod["spec"] == actual_pod["spec"]
assert self.expected_pod["metadata"]["labels"] ==
actual_pod["metadata"]["labels"]
- def test_pod_dnspolicy(self, mock_get_connection):
+ def test_pod_dnspolicy(self):
dns_policy = "ClusterFirstWithHostNet"
k = KubernetesPodOperator(
namespace="default",
@@ -359,7 +361,7 @@ class TestKubernetesPodOperatorSystem:
assert self.expected_pod["spec"] == actual_pod["spec"]
assert self.expected_pod["metadata"]["labels"] ==
actual_pod["metadata"]["labels"]
- def test_pod_schedulername(self, mock_get_connection):
+ def test_pod_schedulername(self):
scheduler_name = "default-scheduler"
k = KubernetesPodOperator(
namespace="default",
@@ -378,7 +380,7 @@ class TestKubernetesPodOperatorSystem:
self.expected_pod["spec"]["schedulerName"] = scheduler_name
assert self.expected_pod == actual_pod
- def test_pod_node_selector(self, mock_get_connection):
+ def test_pod_node_selector(self):
node_selector = {"beta.kubernetes.io/os": "linux"}
k = KubernetesPodOperator(
namespace="default",
@@ -397,7 +399,7 @@ class TestKubernetesPodOperatorSystem:
self.expected_pod["spec"]["nodeSelector"] = node_selector
assert self.expected_pod == actual_pod
- def test_pod_resources(self, mock_get_connection):
+ def test_pod_resources(self):
resources = k8s.V1ResourceRequirements(
requests={"memory": "64Mi", "cpu": "250m", "ephemeral-storage":
"1Gi"},
limits={"memory": "64Mi", "cpu": 0.25, "nvidia.com/gpu": None,
"ephemeral-storage": "2Gi"},
@@ -467,7 +469,7 @@ class TestKubernetesPodOperatorSystem:
),
],
)
- def test_pod_affinity(self, val, mock_get_connection):
+ def test_pod_affinity(self, val):
expected = {
"nodeAffinity": {
"requiredDuringSchedulingIgnoredDuringExecution": {
@@ -498,7 +500,7 @@ class TestKubernetesPodOperatorSystem:
self.expected_pod["spec"]["affinity"] = expected
assert self.expected_pod == actual_pod
- def test_port(self, mock_get_connection):
+ def test_port(self):
port = k8s.V1ContainerPort(
name="http",
container_port=80,
@@ -521,7 +523,7 @@ class TestKubernetesPodOperatorSystem:
self.expected_pod["spec"]["containers"][0]["ports"] = [{"name":
"http", "containerPort": 80}]
assert self.expected_pod == actual_pod
- def test_volume_mount(self, mock_get_connection):
+ def test_volume_mount(self):
with mock.patch.object(PodManager, "log") as mock_logger:
volume_mount = k8s.V1VolumeMount(
name="test-volume", mount_path="/tmp/test_volume",
sub_path=None, read_only=False
@@ -561,7 +563,7 @@ class TestKubernetesPodOperatorSystem:
assert self.expected_pod == actual_pod
@pytest.mark.parametrize("uid", [0, 1000])
- def test_run_as_user(self, uid, mock_get_connection):
+ def test_run_as_user(self, uid):
security_context = V1PodSecurityContext(run_as_user=uid)
name = str(uuid4())
k = KubernetesPodOperator(
@@ -586,7 +588,7 @@ class TestKubernetesPodOperatorSystem:
assert pod.to_dict()["spec"]["security_context"]["run_as_user"] == uid
@pytest.mark.parametrize("gid", [0, 1000])
- def test_fs_group(self, gid, mock_get_connection):
+ def test_fs_group(self, gid):
security_context = V1PodSecurityContext(fs_group=gid)
name = str(uuid4())
k = KubernetesPodOperator(
@@ -610,7 +612,7 @@ class TestKubernetesPodOperatorSystem:
)
assert pod.to_dict()["spec"]["security_context"]["fs_group"] == gid
- def test_disable_privilege_escalation(self, mock_get_connection):
+ def test_disable_privilege_escalation(self):
container_security_context =
V1SecurityContext(allow_privilege_escalation=False)
k = KubernetesPodOperator(
@@ -632,7 +634,7 @@ class TestKubernetesPodOperatorSystem:
}
assert self.expected_pod == actual_pod
- def test_faulty_image(self, mock_get_connection):
+ def test_faulty_image(self):
bad_image_name = "foobar"
k = KubernetesPodOperator(
namespace="default",
@@ -652,7 +654,7 @@ class TestKubernetesPodOperatorSystem:
self.expected_pod["spec"]["containers"][0]["image"] = bad_image_name
assert self.expected_pod == actual_pod
- def test_faulty_service_account(self, mock_get_connection):
+ def test_faulty_service_account(self):
k = KubernetesPodOperator(
namespace="default",
image="ubuntu:16.04",
@@ -670,7 +672,7 @@ class TestKubernetesPodOperatorSystem:
with pytest.raises(ApiException, match="error looking up service
account default/foobar"):
k.get_or_create_pod(pod, context)
- def test_pod_failure(self, mock_get_connection):
+ def test_pod_failure(self):
"""
Tests that the task fails when a pod reports a failure
"""
@@ -692,7 +694,7 @@ class TestKubernetesPodOperatorSystem:
self.expected_pod["spec"]["containers"][0]["args"] =
bad_internal_command
assert self.expected_pod == actual_pod
- def test_xcom_push(self, test_label, mock_get_connection):
+ def test_xcom_push(self, test_label):
expected = {"test_label": test_label, "buzz": 2}
args = [f"echo '{json.dumps(expected)}' > /airflow/xcom/return.json"]
k = KubernetesPodOperator(
@@ -708,7 +710,7 @@ class TestKubernetesPodOperatorSystem:
context = create_context(k)
assert k.execute(context) == expected
- def test_env_vars(self, mock_get_connection):
+ def test_env_vars(self):
# WHEN
env_vars = [
k8s.V1EnvVar(name="ENV1", value="val1"),
@@ -740,7 +742,7 @@ class TestKubernetesPodOperatorSystem:
]
assert self.expected_pod == actual_pod
- def test_pod_template_file_system(self, mock_get_connection,
basic_pod_template):
+ def test_pod_template_file_system(self, basic_pod_template):
"""Note: this test requires that you have a namespace ``mem-example``
in your cluster."""
k = KubernetesPodOperator(
task_id=str(uuid4()),
@@ -762,9 +764,7 @@ class TestKubernetesPodOperatorSystem:
pytest.param({"env_name": "value"}, id="backcompat"), # todo:
remove?
],
)
- def test_pod_template_file_with_overrides_system(
- self, env_vars, test_label, mock_get_connection, basic_pod_template
- ):
+ def test_pod_template_file_with_overrides_system(self, env_vars,
test_label, basic_pod_template):
k = KubernetesPodOperator(
task_id=str(uuid4()),
labels=self.labels,
@@ -790,7 +790,7 @@ class TestKubernetesPodOperatorSystem:
assert k.pod.spec.containers[0].env == [k8s.V1EnvVar(name="env_name",
value="value")]
assert result == {"hello": "world"}
- def test_pod_template_file_with_full_pod_spec(self, test_label,
mock_get_connection, basic_pod_template):
+ def test_pod_template_file_with_full_pod_spec(self, test_label,
basic_pod_template):
pod_spec = k8s.V1Pod(
metadata=k8s.V1ObjectMeta(
labels={"test_label": test_label, "fizz": "buzz"},
@@ -830,7 +830,7 @@ class TestKubernetesPodOperatorSystem:
assert k.pod.spec.containers[0].env == [k8s.V1EnvVar(name="env_name",
value="value")]
assert result == {"hello": "world"}
- def test_full_pod_spec(self, test_label, mock_get_connection):
+ def test_full_pod_spec(self, test_label):
pod_spec = k8s.V1Pod(
metadata=k8s.V1ObjectMeta(
labels={"test_label": test_label, "fizz": "buzz"},
namespace="default", name="test-pod"
@@ -875,7 +875,7 @@ class TestKubernetesPodOperatorSystem:
assert k.pod.spec.containers[0].env == [k8s.V1EnvVar(name="env_name",
value="value")]
assert result == {"hello": "world"}
- def test_init_container(self, mock_get_connection):
+ def test_init_container(self):
# GIVEN
volume_mounts = [
k8s.V1VolumeMount(mount_path="/etc/foo", name="test-volume",
sub_path=None, read_only=True)
@@ -1070,7 +1070,7 @@ class TestKubernetesPodOperatorSystem:
self.expected_pod["spec"]["priorityClassName"] = priority_class_name
assert self.expected_pod == actual_pod
- def test_pod_name(self, mock_get_connection):
+ def test_pod_name(self):
pod_name_too_long = "a" * 221
k = KubernetesPodOperator(
namespace="default",
@@ -1089,7 +1089,7 @@ class TestKubernetesPodOperatorSystem:
with pytest.raises(AirflowException):
k.execute(context)
- def test_on_kill(self, mock_get_connection):
+ def test_on_kill(self):
hook = KubernetesHook(conn_id=None, in_cluster=False)
client = hook.core_v1_client
name = "test"
@@ -1129,7 +1129,7 @@ class TestKubernetesPodOperatorSystem:
with pytest.raises(ApiException, match=r'pods \\"test.[a-z0-9]+\\" not
found'):
client.read_namespaced_pod(name=name, namespace=namespace)
- def test_reattach_failing_pod_once(self, mock_get_connection):
+ def test_reattach_failing_pod_once(self):
hook = KubernetesHook(conn_id=None, in_cluster=False)
client = hook.core_v1_client
name = "test"
@@ -1196,7 +1196,7 @@ class TestKubernetesPodOperatorSystem:
k.execute(context)
create_mock.assert_called_once()
- def test_changing_base_container_name_with_get_logs(self,
mock_get_connection):
+ def test_changing_base_container_name_with_get_logs(self):
k = KubernetesPodOperator(
namespace="default",
image="ubuntu:16.04",
@@ -1221,7 +1221,7 @@ class TestKubernetesPodOperatorSystem:
self.expected_pod["spec"]["containers"][0]["name"] = "apple-sauce"
assert self.expected_pod["spec"] == actual_pod["spec"]
- def test_changing_base_container_name_no_logs(self, mock_get_connection):
+ def test_changing_base_container_name_no_logs(self):
"""
This test checks BOTH a modified base container name AND the
get_logs=False flow,
and as a result, also checks that the flow works with fast containers
@@ -1251,7 +1251,7 @@ class TestKubernetesPodOperatorSystem:
self.expected_pod["spec"]["containers"][0]["name"] = "apple-sauce"
assert self.expected_pod["spec"] == actual_pod["spec"]
- def test_changing_base_container_name_no_logs_long(self,
mock_get_connection):
+ def test_changing_base_container_name_no_logs_long(self):
"""
Similar to test_changing_base_container_name_no_logs, but ensures that
pods running longer than 1 second work too.
@@ -1282,7 +1282,7 @@ class TestKubernetesPodOperatorSystem:
self.expected_pod["spec"]["containers"][0]["args"] = ["sleep 3"]
assert self.expected_pod["spec"] == actual_pod["spec"]
- def test_changing_base_container_name_failure(self, mock_get_connection):
+ def test_changing_base_container_name_failure(self):
k = KubernetesPodOperator(
namespace="default",
image="ubuntu:16.04",
@@ -1309,7 +1309,7 @@ class TestKubernetesPodOperatorSystem:
assert mock_get_container_termination_message.call_args[0][1] ==
"apple-sauce"
- def test_base_container_name_init_precedence(self, mock_get_connection):
+ def test_base_container_name_init_precedence(self):
assert (
KubernetesPodOperator(base_container_name="apple-sauce",
task_id=str(uuid4())).base_container_name
== "apple-sauce"
@@ -1328,7 +1328,7 @@ class TestKubernetesPodOperatorSystem:
)
assert MyK8SPodOperator(task_id=str(uuid4())).base_container_name ==
"tomato-sauce"
- def test_init_container_logs(self, mock_get_connection):
+ def test_init_container_logs(self):
marker_from_init_container = f"{uuid4()}"
marker_from_main_container = f"{uuid4()}"
callback = MagicMock()
@@ -1359,7 +1359,7 @@ class TestKubernetesPodOperatorSystem:
assert marker_from_init_container in calls_args
assert marker_from_main_container in calls_args
- def test_init_container_logs_filtered(self, mock_get_connection):
+ def test_init_container_logs_filtered(self):
marker_from_init_container_to_log_1 = f"{uuid4()}"
marker_from_init_container_to_log_2 = f"{uuid4()}"
marker_from_init_container_to_ignore = f"{uuid4()}"
