fix: kpo async kube_config_path

Author: raphaelauv

providers/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py

@@ -734,9 +734,11 @@ async def _load_config(self):
         """Return Kubernetes API session for use with requests."""
         in_cluster = self._coalesce_param(self.in_cluster, await self._get_field("in_cluster"))
         cluster_context = self._coalesce_param(self.cluster_context, await self._get_field("cluster_context"))
+        kubeconfig_path = self._coalesce_param(self.config_file, self._get_field("kube_config_path"))
         kubeconfig = await self._get_field("kube_config")
-
-        num_selected_configuration = sum(1 for o in [in_cluster, kubeconfig, self.config_dict] if o)
+        num_selected_configuration = sum(
+            1 for o in [in_cluster, kubeconfig, kubeconfig_path, self.config_dict] if o
+        )
 
         if num_selected_configuration > 1:
             raise AirflowException(
@@ -757,6 +759,16 @@ async def _load_config(self):
             await async_config.load_kube_config_from_dict(self.config_dict)
             return async_client.ApiClient()
 
+        if kubeconfig_path is not None:
+            self.log.debug("loading kube_config from: %s", kubeconfig_path)
+            self._is_in_cluster = False
+            await async_config.load_kube_config(
+                config_file=kubeconfig_path,
+                client_configuration=self.client_configuration,
+                context=cluster_context,
+            )
+            return async_client.ApiClient()
+
         if kubeconfig is not None:
             async with aiofiles.tempfile.NamedTemporaryFile() as temp_config:
                 self.log.debug(

providers/tests/cncf/kubernetes/hooks/test_kubernetes.py

@@ -24,8 +24,10 @@
 from unittest import mock
 from unittest.mock import MagicMock, PropertyMock, patch
 
+import anyio
 import kubernetes
 import pytest
+import yaml
 from kubernetes.client import V1Deployment, V1DeploymentStatus
 from kubernetes.client.rest import ApiException
 from kubernetes.config import ConfigException
@@ -860,6 +862,26 @@ async def test_load_config_with_incluster(self, kube_config_merger, kube_config_
         assert not kube_config_loader.called
         assert not kube_config_merger.called
 
+    @pytest.mark.asyncio
+    @mock.patch(INCLUSTER_CONFIG_LOADER)
+    @mock.patch(KUBE_CONFIG_MERGER)
+    async def test_load_config_with_config_file(
+        self, kube_config_merger, incluster_config, kube_config_loader, tmp_path
+    ):
+        file_name = f"{tmp_path}/config"
+        async with await anyio.open_file(file_name, "w+") as f:
+            yaml.dump({"a": "b"}, f)
+        hook = AsyncKubernetesHook(
+            conn_id=None,
+            in_cluster=False,
+            config_file=file_name,
+            cluster_context=None,
+        )
+        await hook._load_config()
+        assert not incluster_config.called
+        assert hook._is_in_cluster is False
+        kube_config_loader.assert_called_once()
+
     @pytest.mark.asyncio
     @mock.patch(INCLUSTER_CONFIG_LOADER)
     @mock.patch(KUBE_CONFIG_MERGER)