From bc4c914e17bcd42cedc6a371be97a9e5700ab796 Mon Sep 17 00:00:00 2001 From: YusefSyed <211442445+YusefSyed@users.noreply.github.com> Date: Wed, 2 Sep 2026 19:58:46 -0400 Subject: [PATCH 1/2] Add Kubernetes provider utility test coverage The decorator source rewriting and YAML deletion helpers can silently corrupt user code or hide aggregate API failures without focused regression coverage. --- .../unit/always/test_project_structure.py | 3 +- .../test_python_kubernetes_script.py | 117 +++++++++++++ .../cncf/kubernetes/utils/test_delete_from.py | 156 ++++++++++++++++++ 3 files changed, 274 insertions(+), 2 deletions(-) create mode 100644 providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_python_kubernetes_script.py create mode 100644 providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_delete_from.py diff --git a/airflow-core/tests/unit/always/test_project_structure.py b/airflow-core/tests/unit/always/test_project_structure.py index cbade3332f677..6820a1dc07b11 100644 --- a/airflow-core/tests/unit/always/test_project_structure.py +++ b/airflow-core/tests/unit/always/test_project_structure.py @@ -73,8 +73,7 @@ def test_providers_modules_should_have_tests(self): "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor_utils.py", "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_exceptions.py", "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_kube_config.py", - "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_python_kubernetes_script.py", - "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_delete_from.py", + "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_secret.py", "providers/common/compat/tests/unit/common/compat/standard/test_utils.py", "providers/fab/tests/unit/fab/migrations/test_env.py", "providers/fab/tests/unit/fab/www/extensions/test_init_appbuilder.py", diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_python_kubernetes_script.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_python_kubernetes_script.py new file mode 100644 index 0000000000000..b13e3f9d96781 --- /dev/null +++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_python_kubernetes_script.py @@ -0,0 +1,117 @@ +# 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 + +from unittest import mock + +import pytest +from jinja2 import Environment, StrictUndefined, UndefinedError +from jinja2.nativetypes import NativeEnvironment + +from airflow.providers.cncf.kubernetes.python_kubernetes_script import ( + remove_task_decorator, + write_python_script, +) + + +@pytest.mark.parametrize( + ("python_source", "expected"), + [ + ( + "@task.kubernetes\ndef callable():\n return 1\n", + "def callable():\n return 1\n", + ), + ( + '@task.kubernetes(image=build_image(name="worker"), retries=2)\ndef callable():\n return 1\n', + "def callable():\n return 1\n", + ), + ( + "@setup\n@teardown\n@task.kubernetes()\ndef callable():\n return 1\n", + "def callable():\n return 1\n", + ), + ], +) +def test_remove_task_decorator_removes_task_and_lifecycle_decorators(python_source, expected): + assert remove_task_decorator(python_source, "@task.kubernetes") == expected + + +def test_remove_task_decorator_returns_source_unchanged_without_task_decorator(): + python_source = "@other_decorator\ndef callable():\n return 1\n" + + assert remove_task_decorator(python_source, "@task.kubernetes") == python_source + + +def test_write_python_script_renders_template(tmp_path): + filename = tmp_path / "script.py" + + write_python_script( + { + "op_args": [], + "op_kwargs": {}, + "pickling_library": "pickle", + "python_callable": "callable", + "python_callable_source": "def callable():\n return 1", + }, + str(filename), + ) + + rendered_script = filename.read_text() + assert "import pickle" in rendered_script + assert 'arg_dict = {"args": [], "kwargs": {}}' in rendered_script + assert "def callable():\n return 1" in rendered_script + assert 'res = callable(*arg_dict["args"], **arg_dict["kwargs"])' in rendered_script + + +@mock.patch( + "airflow.providers.cncf.kubernetes.python_kubernetes_script.NativeEnvironment", + wraps=NativeEnvironment, +) +def test_write_python_script_uses_native_environment_for_native_rendering(mock_native_environment, tmp_path): + filename = tmp_path / "script.py" + + write_python_script( + { + "op_args": [], + "op_kwargs": {}, + "pickling_library": "pickle", + "python_callable": "callable", + "python_callable_source": "def callable():\n return 1", + }, + str(filename), + render_template_as_native_obj=True, + ) + + mock_native_environment.assert_called_once_with( + loader=mock.ANY, + undefined=StrictUndefined, + ) + assert "def callable():" in filename.read_text() + + +@mock.patch( + "airflow.providers.cncf.kubernetes.python_kubernetes_script.Environment", + wraps=Environment, +) +def test_write_python_script_raises_for_missing_template_context(mock_environment, tmp_path): + with pytest.raises(UndefinedError): + write_python_script({}, str(tmp_path / "script.py")) + + mock_environment.assert_called_once_with( + loader=mock.ANY, + undefined=StrictUndefined, + autoescape=mock.ANY, + ) diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_delete_from.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_delete_from.py new file mode 100644 index 0000000000000..ed27794a889ae --- /dev/null +++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_delete_from.py @@ -0,0 +1,156 @@ +# 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 + +from unittest import mock + +import pytest +from kubernetes.client.rest import ApiException + +from airflow.providers.cncf.kubernetes.utils import delete_from + + +def create_api_constructor(monkeypatch, api_class_name): + api_constructor = mock.create_autospec(getattr(delete_from.client, api_class_name)) + monkeypatch.setattr(delete_from.client, api_class_name, api_constructor) + return api_constructor + + +@pytest.mark.parametrize( + ("api_version", "kind", "api_class_name", "method_name"), + [ + ("v1", "ConfigMap", "CoreV1Api", "delete_namespaced_config_map"), + ("apps/v1", "Deployment", "AppsV1Api", "delete_namespaced_deployment"), + ( + "apiextensions.k8s.io/v1", + "CustomResourceDefinition", + "ApiextensionsV1Api", + "delete_custom_resource_definition", + ), + ( + "flowcontrol.apiserver.k8s.io/v1", + "FlowSchema", + "FlowcontrolApiserverV1Api", + "delete_flow_schema", + ), + ], +) +def test_delete_from_yaml_derives_api_class_and_kind( + monkeypatch, api_version, kind, api_class_name, method_name +): + api_constructor = create_api_constructor(monkeypatch, api_class_name) + api_client = mock.sentinel.api_client + api = api_constructor.return_value + getattr(api, method_name).return_value = mock.Mock(status="Success") + + delete_from._delete_from_yaml_single_item( + k8s_client=api_client, + yml_document={"apiVersion": api_version, "kind": kind, "metadata": {"name": "resource"}}, + ) + + api_constructor.assert_called_once_with(api_client) + if method_name.startswith("delete_namespaced"): + getattr(api, method_name).assert_called_once_with( + name="resource", namespace="default", body=delete_from.DEFAULT_DELETION_BODY + ) + else: + getattr(api, method_name).assert_called_once_with( + name="resource", body=delete_from.DEFAULT_DELETION_BODY + ) + + +def test_delete_from_yaml_document_namespace_takes_precedence(monkeypatch): + api_constructor = create_api_constructor(monkeypatch, "CoreV1Api") + api = api_constructor.return_value + api.delete_namespaced_config_map.return_value = mock.Mock(status="Success") + + delete_from._delete_from_yaml_single_item( + k8s_client=mock.sentinel.api_client, + yml_document={ + "apiVersion": "v1", + "kind": "ConfigMap", + "metadata": {"name": "resource", "namespace": "document-namespace"}, + }, + namespace="argument-namespace", + ) + + api.delete_namespaced_config_map.assert_called_once_with( + name="resource", namespace="document-namespace", body=delete_from.DEFAULT_DELETION_BODY + ) + + +def test_delete_from_dict_expands_lists_and_inherits_api_version(monkeypatch): + api_constructor = create_api_constructor(monkeypatch, "CoreV1Api") + api = api_constructor.return_value + api.delete_namespaced_config_map.return_value = mock.Mock(status="Success") + document = { + "apiVersion": "v1", + "kind": "ConfigMapList", + "items": [ + {"metadata": {"name": "first"}}, + {"metadata": {"name": "second", "namespace": "item-namespace"}}, + ], + } + + delete_from.delete_from_dict( + k8s_client=mock.sentinel.api_client, + data=document, + body=None, + namespace="argument-namespace", + ) + + assert document["items"] == [ + {"apiVersion": "v1", "kind": "ConfigMap", "metadata": {"name": "first"}}, + { + "apiVersion": "v1", + "kind": "ConfigMap", + "metadata": {"name": "second", "namespace": "item-namespace"}, + }, + ] + api.delete_namespaced_config_map.assert_has_calls( + [ + mock.call(name="first", namespace="argument-namespace", body=delete_from.DEFAULT_DELETION_BODY), + mock.call(name="second", namespace="item-namespace", body=delete_from.DEFAULT_DELETION_BODY), + ] + ) + + +def test_delete_from_dict_collects_api_exceptions(monkeypatch): + api_constructor = create_api_constructor(monkeypatch, "CoreV1Api") + api = api_constructor.return_value + first_exception = ApiException(reason="first") + first_exception.body = "first body" + second_exception = ApiException(reason="second") + second_exception.body = "second body" + api.delete_namespaced_config_map.side_effect = [first_exception, second_exception] + + with pytest.raises(delete_from.FailToDeleteError) as error: + delete_from.delete_from_dict( + k8s_client=mock.sentinel.api_client, + data={ + "apiVersion": "v1", + "kind": "ConfigMapList", + "items": [{"metadata": {"name": "first"}}, {"metadata": {"name": "second"}}], + }, + body=None, + namespace="default", + ) + + assert error.value.api_exceptions == [first_exception, second_exception] + assert ( + str(error.value) == "Error from server (first):first body\nError from server (second):second body\n" + ) From 7d0a6f383142036ea7e0e503ef1dae9251da4b18 Mon Sep 17 00:00:00 2001 From: YusefSyed <211442445+YusefSyed@users.noreply.github.com> Date: Tue, 22 Sep 2026 19:07:12 -0400 Subject: [PATCH 2/2] Cover Kubernetes utility edge cases Keep generated task source and YAML deletion behavior protected against regressions identified in review. --- .../unit/always/test_project_structure.py | 1 - .../test_python_kubernetes_script.py | 5 + .../cncf/kubernetes/utils/test_delete_from.py | 138 ++++++++++++------ 3 files changed, 101 insertions(+), 43 deletions(-) diff --git a/airflow-core/tests/unit/always/test_project_structure.py b/airflow-core/tests/unit/always/test_project_structure.py index 6820a1dc07b11..315372ab9d38e 100644 --- a/airflow-core/tests/unit/always/test_project_structure.py +++ b/airflow-core/tests/unit/always/test_project_structure.py @@ -73,7 +73,6 @@ def test_providers_modules_should_have_tests(self): "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor_utils.py", "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_exceptions.py", "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_kube_config.py", - "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_secret.py", "providers/common/compat/tests/unit/common/compat/standard/test_utils.py", "providers/fab/tests/unit/fab/migrations/test_env.py", "providers/fab/tests/unit/fab/www/extensions/test_init_appbuilder.py", diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_python_kubernetes_script.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_python_kubernetes_script.py index b13e3f9d96781..559850df4228e 100644 --- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_python_kubernetes_script.py +++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_python_kubernetes_script.py @@ -43,6 +43,11 @@ "@setup\n@teardown\n@task.kubernetes()\ndef callable():\n return 1\n", "def callable():\n return 1\n", ), + ( + "@before\n@setup\n@teardown(on_failure_fail_dagrun=True)\n" + '@task.kubernetes(image=build_image(name="worker"))\n@after\ndef callable():\n return 1\n', + "@before\n@after\ndef callable():\n return 1\n", + ), ], ) def test_remove_task_decorator_removes_task_and_lifecycle_decorators(python_source, expected): diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_delete_from.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_delete_from.py index ed27794a889ae..6318bf988751e 100644 --- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_delete_from.py +++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_delete_from.py @@ -24,59 +24,76 @@ from airflow.providers.cncf.kubernetes.utils import delete_from -def create_api_constructor(monkeypatch, api_class_name): - api_constructor = mock.create_autospec(getattr(delete_from.client, api_class_name)) - monkeypatch.setattr(delete_from.client, api_class_name, api_constructor) - return api_constructor - - @pytest.mark.parametrize( - ("api_version", "kind", "api_class_name", "method_name"), + ("api_version", "kind", "method_name", "expected_kwargs"), [ - ("v1", "ConfigMap", "CoreV1Api", "delete_namespaced_config_map"), - ("apps/v1", "Deployment", "AppsV1Api", "delete_namespaced_deployment"), ( - "apiextensions.k8s.io/v1", - "CustomResourceDefinition", - "ApiextensionsV1Api", - "delete_custom_resource_definition", + "v1", + "ConfigMap", + "delete_namespaced_config_map", + {"name": "resource", "namespace": "default", "body": delete_from.DEFAULT_DELETION_BODY}, ), ( - "flowcontrol.apiserver.k8s.io/v1", - "FlowSchema", - "FlowcontrolApiserverV1Api", - "delete_flow_schema", + "apps/v1", + "Deployment", + "delete_namespaced_deployment", + {"name": "resource", "namespace": "default", "body": delete_from.DEFAULT_DELETION_BODY}, ), ], ) -def test_delete_from_yaml_derives_api_class_and_kind( - monkeypatch, api_version, kind, api_class_name, method_name +@mock.patch("airflow.providers.cncf.kubernetes.utils.delete_from.client.AppsV1Api", autospec=True) +@mock.patch("airflow.providers.cncf.kubernetes.utils.delete_from.client.CoreV1Api", autospec=True) +def test_delete_from_yaml_derives_namespaced_api_class_and_kind( + mock_core_v1_api, mock_apps_v1_api, api_version, kind, method_name, expected_kwargs ): - api_constructor = create_api_constructor(monkeypatch, api_class_name) - api_client = mock.sentinel.api_client + api_constructor = mock_core_v1_api if api_version == "v1" else mock_apps_v1_api api = api_constructor.return_value - getattr(api, method_name).return_value = mock.Mock(status="Success") + getattr(api, method_name).return_value = mock.Mock(spec=["status"], status="Success") delete_from._delete_from_yaml_single_item( - k8s_client=api_client, + k8s_client=mock.sentinel.api_client, yml_document={"apiVersion": api_version, "kind": kind, "metadata": {"name": "resource"}}, ) - api_constructor.assert_called_once_with(api_client) - if method_name.startswith("delete_namespaced"): - getattr(api, method_name).assert_called_once_with( - name="resource", namespace="default", body=delete_from.DEFAULT_DELETION_BODY - ) - else: - getattr(api, method_name).assert_called_once_with( - name="resource", body=delete_from.DEFAULT_DELETION_BODY - ) + api_constructor.assert_called_once_with(mock.sentinel.api_client) + getattr(api, method_name).assert_called_once_with(**expected_kwargs) -def test_delete_from_yaml_document_namespace_takes_precedence(monkeypatch): - api_constructor = create_api_constructor(monkeypatch, "CoreV1Api") +@pytest.mark.parametrize( + ("api_version", "kind", "method_name"), + [ + ("apiextensions.k8s.io/v1", "CustomResourceDefinition", "delete_custom_resource_definition"), + ("flowcontrol.apiserver.k8s.io/v1", "FlowSchema", "delete_flow_schema"), + ], +) +@mock.patch( + "airflow.providers.cncf.kubernetes.utils.delete_from.client.FlowcontrolApiserverV1Api", autospec=True +) +@mock.patch("airflow.providers.cncf.kubernetes.utils.delete_from.client.ApiextensionsV1Api", autospec=True) +def test_delete_from_yaml_derives_cluster_scoped_api_class_and_kind( + mock_apiextensions_v1_api, mock_flowcontrol_apiserver_v1_api, api_version, kind, method_name +): + api_constructor = ( + mock_apiextensions_v1_api + if api_version == "apiextensions.k8s.io/v1" + else mock_flowcontrol_apiserver_v1_api + ) api = api_constructor.return_value - api.delete_namespaced_config_map.return_value = mock.Mock(status="Success") + getattr(api, method_name).return_value = mock.Mock(spec=["status"], status="Success") + + delete_from._delete_from_yaml_single_item( + k8s_client=mock.sentinel.api_client, + yml_document={"apiVersion": api_version, "kind": kind, "metadata": {"name": "resource"}}, + ) + + api_constructor.assert_called_once_with(mock.sentinel.api_client) + getattr(api, method_name).assert_called_once_with(name="resource", body=delete_from.DEFAULT_DELETION_BODY) + + +@mock.patch("airflow.providers.cncf.kubernetes.utils.delete_from.client.CoreV1Api", autospec=True) +def test_delete_from_yaml_document_namespace_takes_precedence(mock_core_v1_api): + api = mock_core_v1_api.return_value + api.delete_namespaced_config_map.return_value = mock.Mock(spec=["status"], status="Success") delete_from._delete_from_yaml_single_item( k8s_client=mock.sentinel.api_client, @@ -93,10 +110,10 @@ def test_delete_from_yaml_document_namespace_takes_precedence(monkeypatch): ) -def test_delete_from_dict_expands_lists_and_inherits_api_version(monkeypatch): - api_constructor = create_api_constructor(monkeypatch, "CoreV1Api") - api = api_constructor.return_value - api.delete_namespaced_config_map.return_value = mock.Mock(status="Success") +@mock.patch("airflow.providers.cncf.kubernetes.utils.delete_from.client.CoreV1Api", autospec=True) +def test_delete_from_dict_expands_lists_and_inherits_api_version(mock_core_v1_api): + api = mock_core_v1_api.return_value + api.delete_namespaced_config_map.return_value = mock.Mock(spec=["status"], status="Success") document = { "apiVersion": "v1", "kind": "ConfigMapList", @@ -129,9 +146,9 @@ def test_delete_from_dict_expands_lists_and_inherits_api_version(monkeypatch): ) -def test_delete_from_dict_collects_api_exceptions(monkeypatch): - api_constructor = create_api_constructor(monkeypatch, "CoreV1Api") - api = api_constructor.return_value +@mock.patch("airflow.providers.cncf.kubernetes.utils.delete_from.client.CoreV1Api", autospec=True) +def test_delete_from_dict_collects_api_exceptions(mock_core_v1_api): + api = mock_core_v1_api.return_value first_exception = ApiException(reason="first") first_exception.body = "first body" second_exception = ApiException(reason="second") @@ -154,3 +171,40 @@ def test_delete_from_dict_collects_api_exceptions(monkeypatch): assert ( str(error.value) == "Error from server (first):first body\nError from server (second):second body\n" ) + + +@mock.patch("airflow.providers.cncf.kubernetes.utils.delete_from.delete_from_dict", autospec=True) +def test_delete_from_yaml_skips_empty_documents_and_forwards_verbose(mock_delete_from_dict): + document = {"apiVersion": "v1", "kind": "ConfigMap", "metadata": {"name": "resource"}} + + delete_from.delete_from_yaml( + k8s_client=mock.sentinel.api_client, + yaml_objects=[None, document], + verbose=True, + namespace="configured-namespace", + body=mock.sentinel.body, + grace_period_seconds=0, + ) + + mock_delete_from_dict.assert_called_once_with( + k8s_client=mock.sentinel.api_client, + data=document, + body=mock.sentinel.body, + namespace="configured-namespace", + verbose=True, + grace_period_seconds=0, + ) + + +@mock.patch("airflow.providers.cncf.kubernetes.utils.delete_from.client.CoreV1Api", autospec=True) +def test_delete_from_yaml_prints_status_when_verbose(mock_core_v1_api, capsys): + api = mock_core_v1_api.return_value + api.delete_namespaced_config_map.return_value = mock.Mock(spec=["status"], status="Success") + + delete_from._delete_from_yaml_single_item( + k8s_client=mock.sentinel.api_client, + yml_document={"apiVersion": "v1", "kind": "ConfigMap", "metadata": {"name": "resource"}}, + verbose=True, + ) + + assert capsys.readouterr().out == "config_map deleted. status='Success'\n"