Removed deprecated k8s rendering methods from task instance module (a…

…pache#42641)

* Removed deprecated k8s rendering methods from task instance module

* task k8s rendering unit tests update

airflow/models/taskinstance.py

@@ -3405,53 +3405,6 @@ def render_templates(
 
         return original_task
 
-    def render_k8s_pod_yaml(self) -> dict | None:
-        """Render the k8s pod yaml."""
-        try:
-            from airflow.providers.cncf.kubernetes.template_rendering import (
-                render_k8s_pod_yaml as render_k8s_pod_yaml_from_provider,
-            )
-        except ImportError:
-            raise RuntimeError(
-                "You need to have the `cncf.kubernetes` provider installed to use this feature. "
-                "Also rather than calling it directly you should import "
-                "render_k8s_pod_yaml from airflow.providers.cncf.kubernetes.template_rendering "
-                "and call it with TaskInstance as the first argument."
-            )
-        warnings.warn(
-            "You should not call `task_instance.render_k8s_pod_yaml` directly. This method will be removed"
-            "in Airflow 3. Rather than calling it directly you should import "
-            "`render_k8s_pod_yaml` from `airflow.providers.cncf.kubernetes.template_rendering` "
-            "and call it with `TaskInstance` as the first argument.",
-            DeprecationWarning,
-            stacklevel=2,
-        )
-        return render_k8s_pod_yaml_from_provider(self)
-
-    @provide_session
-    def get_rendered_k8s_spec(self, session: Session = NEW_SESSION):
-        """Render the k8s pod yaml."""
-        try:
-            from airflow.providers.cncf.kubernetes.template_rendering import (
-                get_rendered_k8s_spec as get_rendered_k8s_spec_from_provider,
-            )
-        except ImportError:
-            raise RuntimeError(
-                "You need to have the `cncf.kubernetes` provider installed to use this feature. "
-                "Also rather than calling it directly you should import "
-                "`get_rendered_k8s_spec` from `airflow.providers.cncf.kubernetes.template_rendering` "
-                "and call it with `TaskInstance` as the first argument."
-            )
-        warnings.warn(
-            "You should not call `task_instance.render_k8s_pod_yaml` directly. This method will be removed"
-            "in Airflow 3. Rather than calling it directly you should import "
-            "`get_rendered_k8s_spec` from `airflow.providers.cncf.kubernetes.template_rendering` "
-            "and call it with `TaskInstance` as the first argument.",
-            DeprecationWarning,
-            stacklevel=2,
-        )
-        return get_rendered_k8s_spec_from_provider(self, session=session)
-
     def get_email_subject_content(
         self, exception: BaseException, task: BaseOperator | None = None
     ) -> tuple[str, str, str]:

tests/providers/cncf/kubernetes/test_template_rendering.py

@@ -25,6 +25,7 @@
 from airflow.configuration import TEST_DAGS_FOLDER
 from airflow.models.renderedtifields import RenderedTaskInstanceFields, RenderedTaskInstanceFields as RTIF
 from airflow.operators.bash import BashOperator
+from airflow.providers.cncf.kubernetes.template_rendering import get_rendered_k8s_spec, render_k8s_pod_yaml
 from airflow.utils.session import create_session
 from airflow.version import version
 from tests.models import DEFAULT_DATE
@@ -82,7 +83,7 @@ def test_render_k8s_pod_yaml(pod_mutation_hook, create_task_instance):
         },
     }
 
-    assert ti.render_k8s_pod_yaml() == expected_pod_spec
+    assert render_k8s_pod_yaml(ti) == expected_pod_spec
     pod_mutation_hook.assert_called_once_with(mock.ANY)
 
 
@@ -100,7 +101,7 @@ def test_get_rendered_k8s_spec(render_k8s_pod_yaml, rtif_get_k8s_pod_yaml, creat
     session = mock.Mock()
 
     rtif_get_k8s_pod_yaml.return_value = fake_spec
-    assert ti.get_rendered_k8s_spec(session) == fake_spec
+    assert get_rendered_k8s_spec(ti, session=session) == fake_spec
 
     rtif_get_k8s_pod_yaml.assert_called_once_with(ti, session=session)
     render_k8s_pod_yaml.assert_not_called()
@@ -109,7 +110,7 @@ def test_get_rendered_k8s_spec(render_k8s_pod_yaml, rtif_get_k8s_pod_yaml, creat
     rtif_get_k8s_pod_yaml.return_value = None
     render_k8s_pod_yaml.return_value = fake_spec
 
-    assert ti.get_rendered_k8s_spec(session) == fake_spec
+    assert get_rendered_k8s_spec(session) == fake_spec
 
     render_k8s_pod_yaml.assert_called_once()