Add missing tests for cncf.kubernetes script and delete utilities

airflow-core/tests/unit/always/test_project_structure.py

@@ -76,8 +76,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_python_kubernetes_script.py",
-            "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_delete_from.py",
             "providers/common/compat/tests/unit/common/compat/standard/test_utils.py",
             "providers/fab/tests/unit/fab/auth_manager/api_fastapi/datamodels/test_login.py",
             "providers/fab/tests/unit/fab/migrations/test_env.py",

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_python_kubernetes_script.py

@@ -0,0 +1,116 @@
+# 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.exceptions import UndefinedError
+
+from airflow.providers.cncf.kubernetes.python_kubernetes_script import (
+    remove_task_decorator,
+    write_python_script,
+)
+
+
+class TestRemoveTaskDecorator:
+    @pytest.mark.parametrize(
+        ("source", "expected"),
+        [
+            pytest.param(
+                "@task.kubernetes\ndef my_task():\n    pass\n",
+                "def my_task():\n    pass\n",
+                id="no-arguments",
+            ),
+            pytest.param(
+                '@task.kubernetes(image="python:3.9")\ndef my_task():\n    pass\n',
+                "def my_task():\n    pass\n",
+                id="with-arguments",
+            ),
+            pytest.param(
+                "@task.kubernetes(retries=get_retries(3))\ndef my_task():\n    pass\n",
+                "def my_task():\n    pass\n",
+                id="nested-parentheses",
+            ),
+        ],
+    )
+    def test_removes_task_decorator(self, source, expected):
+        assert remove_task_decorator(source, "@task.kubernetes") == expected
+
+    def test_removes_setup_and_teardown_alongside(self):
+        source = (
+            "@setup\ndef my_setup():\n    pass\n\n"
+            "@task.kubernetes\ndef my_task():\n    pass\n\n"
+            "@teardown\ndef my_teardown():\n    pass\n"
+        )
+        expected = "def my_setup():\n    pass\n\ndef my_task():\n    pass\n\ndef my_teardown():\n    pass\n"
+        assert remove_task_decorator(source, "@task.kubernetes") == expected
+
+    def test_returns_source_unchanged_when_decorator_missing(self):
+        source = "def my_task():\n    pass\n"
+        assert remove_task_decorator(source, "@task.kubernetes") == source
+
+
+class TestWritePythonScript:
+    @pytest.fixture
+    def jinja_context(self):
+        return {
+            "pickling_library": "pickle",
+            "python_callable_source": "def my_task():\n    return 42\n",
+            "python_callable": "my_task",
+            "op_args": [],
+            "op_kwargs": {},
+        }
+
+    def test_renders_template_to_file(self, tmp_path, jinja_context):
+        target = tmp_path / "script.py"
+        write_python_script(jinja_context, str(target))
+
+        rendered = target.read_text()
+        assert "import pickle" in rendered
+        assert "def my_task():\n    return 42\n" in rendered
+        assert 'res = my_task(*arg_dict["args"], **arg_dict["kwargs"])' in rendered
+
+    def test_native_environment_used_when_requested(self, tmp_path, jinja_context):
+        with (
+            mock.patch(
+                "airflow.providers.cncf.kubernetes.python_kubernetes_script.NativeEnvironment"
+            ) as mock_native_env,
+            mock.patch("airflow.providers.cncf.kubernetes.python_kubernetes_script.Environment") as mock_env,
+        ):
+            write_python_script(
+                jinja_context, str(tmp_path / "script.py"), render_template_as_native_obj=True
+            )
+
+        mock_native_env.assert_called_once()
+        mock_env.assert_not_called()
+
+    def test_standard_environment_used_by_default(self, tmp_path, jinja_context):
+        with (
+            mock.patch(
+                "airflow.providers.cncf.kubernetes.python_kubernetes_script.NativeEnvironment"
+            ) as mock_native_env,
+            mock.patch("airflow.providers.cncf.kubernetes.python_kubernetes_script.Environment") as mock_env,
+        ):
+            write_python_script(jinja_context, str(tmp_path / "script.py"))
+
+        mock_env.assert_called_once()
+        mock_native_env.assert_not_called()
+
+    def test_missing_context_variable_raises(self, tmp_path):
+        with pytest.raises(UndefinedError):
+            write_python_script({}, str(tmp_path / "script.py"))

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_delete_from.py

@@ -0,0 +1,225 @@
+# 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.delete_from import (
+    DEFAULT_DELETION_BODY,
+    FailToDeleteError,
+    _delete_from_yaml_single_item,
+    delete_from_dict,
+    delete_from_yaml,
+)
+
+
+@pytest.fixture
+def mock_client():
+    with mock.patch("airflow.providers.cncf.kubernetes.utils.delete_from.client") as m:
+        m.rest.ApiException = ApiException
+        yield m
+
+
+def _mock_api(mock_client, api_class_name, method_names):
+    """Point client.<api_class_name> at a mock exposing exactly method_names."""
+    api_class_mock = mock.MagicMock()
+    api_mock = mock.MagicMock(spec=method_names)
+    api_class_mock.return_value = api_mock
+    setattr(mock_client, api_class_name, api_class_mock)
+    return api_mock
+
+
+def _doc(api_version="apps/v1", kind="Deployment", name="my-dep", namespace=None):
+    metadata = {"name": name}
+    if namespace is not None:
+        metadata["namespace"] = namespace
+    return {"apiVersion": api_version, "kind": kind, "metadata": metadata}
+
+
+class TestDeleteFromYamlSingleItem:
+    @pytest.mark.parametrize(
+        ("api_version", "kind", "api_class", "method"),
+        [
+            pytest.param("v1", "Pod", "CoreV1Api", "delete_namespaced_pod", id="core-group"),
+            pytest.param(
+                "apps/v1", "Deployment", "AppsV1Api", "delete_namespaced_deployment", id="named-group"
+            ),
+            pytest.param(
+                "apiextensions.k8s.io/v1",
+                "CustomResourceDefinition",
+                "ApiextensionsV1Api",
+                "delete_custom_resource_definition",
+                id="k8s-io-suffix-stripped",
+            ),
+            pytest.param(
+                "networking.k8s.io/v1",
+                "Ingress",
+                "NetworkingV1Api",
+                "delete_namespaced_ingress",
+                id="dns-subdomain-group",
+            ),
+        ],
+    )
+    def test_api_class_derivation_and_dispatch(self, mock_client, api_version, kind, api_class, method):
+        api_mock = _mock_api(mock_client, api_class, [method])
+
+        resp = _delete_from_yaml_single_item(
+            k8s_client=mock.sentinel.k8s_client,
+            yml_document=_doc(api_version, kind),
+            namespace="default",
+        )
+
+        getattr(mock_client, api_class).assert_called_once_with(mock.sentinel.k8s_client)
+        getattr(api_mock, method).assert_called_once()
+        assert resp is getattr(api_mock, method).return_value
+
+    def test_document_namespace_takes_precedence(self, mock_client):
+        api_mock = _mock_api(mock_client, "AppsV1Api", ["delete_namespaced_deployment"])
+
+        _delete_from_yaml_single_item(
+            k8s_client=mock.sentinel.k8s_client,
+            yml_document=_doc(namespace="doc-namespace"),
+            namespace="arg-namespace",
+        )
+
+        api_mock.delete_namespaced_deployment.assert_called_once_with(
+            name="my-dep", namespace="doc-namespace", body=DEFAULT_DELETION_BODY
+        )
+
+    def test_argument_namespace_used_when_document_has_none(self, mock_client):
+        api_mock = _mock_api(mock_client, "AppsV1Api", ["delete_namespaced_deployment"])
+
+        _delete_from_yaml_single_item(
+            k8s_client=mock.sentinel.k8s_client,
+            yml_document=_doc(),
+            namespace="arg-namespace",
+        )
+
+        api_mock.delete_namespaced_deployment.assert_called_once_with(
+            name="my-dep", namespace="arg-namespace", body=DEFAULT_DELETION_BODY
+        )
+
+    def test_explicit_body_passed_through(self, mock_client):
+        api_mock = _mock_api(mock_client, "AppsV1Api", ["delete_namespaced_deployment"])
+        body = mock.sentinel.body
+
+        _delete_from_yaml_single_item(
+            k8s_client=mock.sentinel.k8s_client,
+            yml_document=_doc(),
+            namespace="default",
+            body=body,
+        )
+
+        api_mock.delete_namespaced_deployment.assert_called_once_with(
+            name="my-dep", namespace="default", body=body
+        )
+
+    def test_verbose_prints_deletion_status(self, mock_client, capsys):
+        api_mock = _mock_api(mock_client, "CoreV1Api", ["delete_namespaced_pod"])
+        api_mock.delete_namespaced_pod.return_value.status = "Success"
+
+        _delete_from_yaml_single_item(
+            k8s_client=mock.sentinel.k8s_client,
+            yml_document=_doc("v1", "Pod"),
+            namespace="default",
+            verbose=True,
+        )
+
+        assert "pod deleted. status='Success'" in capsys.readouterr().out
+
+
+class TestDeleteFromDict:
+    def test_list_kind_fans_out_over_items(self, mock_client):
+        api_mock = _mock_api(mock_client, "AppsV1Api", ["delete_namespaced_deployment"])
+        data = {
+            "apiVersion": "apps/v1",
+            "kind": "DeploymentList",
+            "items": [{"metadata": {"name": "dep-1"}}, {"metadata": {"name": "dep-2"}}],
+        }
+
+        delete_from_dict(k8s_client=mock.sentinel.k8s_client, data=data, body=None, namespace="default")
+
+        assert api_mock.delete_namespaced_deployment.call_count == 2
+        for item in data["items"]:
+            assert item["apiVersion"] == "apps/v1"
+            assert item["kind"] == "Deployment"
+
+    def test_api_exceptions_collected_into_fail_to_delete_error(self, mock_client):
+        api_mock = _mock_api(mock_client, "AppsV1Api", ["delete_namespaced_deployment"])
+        error = ApiException(status=404, reason="Not Found")
+        api_mock.delete_namespaced_deployment.side_effect = error
+        data = {
+            "apiVersion": "apps/v1",
+            "kind": "DeploymentList",
+            "items": [{"metadata": {"name": "dep-1"}}, {"metadata": {"name": "dep-2"}}],
+        }
+
+        with pytest.raises(FailToDeleteError) as exc_info:
+            delete_from_dict(k8s_client=mock.sentinel.k8s_client, data=data, body=None, namespace="default")
+
+        assert exc_info.value.api_exceptions == [error, error]
+
+    def test_single_document_api_exception_raises_fail_to_delete_error(self, mock_client):
+        api_mock = _mock_api(mock_client, "AppsV1Api", ["delete_namespaced_deployment"])
+        error = ApiException(status=409, reason="Conflict")
+        api_mock.delete_namespaced_deployment.side_effect = error
+
+        with pytest.raises(FailToDeleteError) as exc_info:
+            delete_from_dict(
+                k8s_client=mock.sentinel.k8s_client,
+                data=_doc(),
+                body=None,
+                namespace="default",
+            )
+
+        assert exc_info.value.api_exceptions == [error]
+
+
+class TestDeleteFromYaml:
+    def test_skips_none_documents(self, mock_client):
+        with mock.patch(
+            "airflow.providers.cncf.kubernetes.utils.delete_from.delete_from_dict"
+        ) as mock_delete_from_dict:
+            delete_from_yaml(
+                k8s_client=mock.sentinel.k8s_client,
+                yaml_objects=[None, {"kind": "Pod"}],
+                namespace="custom",
+            )
+
+        mock_delete_from_dict.assert_called_once_with(
+            k8s_client=mock.sentinel.k8s_client,
+            data={"kind": "Pod"},
+            body=None,
+            namespace="custom",
+            verbose=False,
+        )
+
+
+class TestFailToDeleteError:
+    def test_str_formats_all_exceptions(self):
+        errors = [
+            mock.MagicMock(reason="Not Found", body="no such deployment"),
+            mock.MagicMock(reason="Conflict", body="already deleting"),
+        ]
+
+        assert str(FailToDeleteError(errors)) == (
+            "Error from server (Not Found):no such deployment\n"
+            "Error from server (Conflict):already deleting\n"
+        )