You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by am...@apache.org on 2025/08/10 16:01:38 UTC
(airflow) branch backport-d410915-v3-0-test created (now d88dc83c85b)
This is an automated email from the ASF dual-hosted git repository.
amoghdesai pushed a change to branch backport-d410915-v3-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
at d88dc83c85b [v3-0-test] Use env var connections instead of mocks in kube system tests (#54261)
This branch includes the following new commits:
new d88dc83c85b [v3-0-test] Use env var connections instead of mocks in kube system tests (#54261)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(airflow) 01/01: [v3-0-test] Use env var connections instead of mocks in kube system tests (#54261)
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amoghdesai pushed a commit to branch backport-d410915-v3-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit d88dc83c85b7fa2ee5cd0ba98d99158a9b9a8248
Author: Amogh Desai <am...@gmail.com>
AuthorDate: Sun Aug 10 17:41:27 2025 +0530
[v3-0-test] Use env var connections instead of mocks in kube system tests (#54261)
* 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)
Co-authored-by: Amogh Desai <am...@gmail.com>
---
.../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)
+
+
+@pytest.fixture
+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:]
-@pytest.fixture
-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)
-@pytest.mark.usefixtures("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()}"