prefect-kubernetes: add missing config loaders

src/integrations/prefect-kubernetes/prefect_kubernetes/credentials.py

@@ -181,6 +181,14 @@ async def get_client(
                 context=context,
                 client_configuration=client_configuration,
             )
+        else:
+            try:
+                config.load_incluster_config(client_configuration=client_configuration)
+            except config.ConfigException:
+                # If in-cluster config fails, load the local kubeconfig
+                config.load_kube_config(
+                    client_configuration=client_configuration,
+                )
         async with ApiClient(configuration=client_configuration) as api_client:
             try:
                 yield await self.get_resource_specific_client(

src/integrations/prefect-kubernetes/tests/test_credentials.py

@@ -2,6 +2,7 @@
 import tempfile
 from pathlib import Path
 from typing import Dict
+from unittest.mock import AsyncMock, MagicMock
 
 import pydantic
 import pytest
@@ -13,9 +14,13 @@
     CoreV1Api,
     CustomObjectsApi,
 )
+from kubernetes_asyncio.config import ConfigException
 from kubernetes_asyncio.config.kube_config import list_kube_config_contexts
 from OpenSSL import crypto
-from prefect_kubernetes.credentials import KubernetesClusterConfig
+from prefect_kubernetes.credentials import (
+    KubernetesClusterConfig,
+    KubernetesCredentials,
+)
 
 
 @pytest.fixture
@@ -107,6 +112,19 @@ def config_file(tmp_path, config_context) -> Path:
     return config_file
 
 
+@pytest.fixture
+def mock_cluster_config(monkeypatch):
+    mock = MagicMock()
+    # We cannot mock this or the `except` clause will complain
+    mock.ConfigException.return_value = ConfigException
+    mock.load_kube_config = AsyncMock()
+    monkeypatch.setattr("prefect_kubernetes.credentials.config", mock)
+    monkeypatch.setattr(
+        "prefect_kubernetes.credentials.config.ConfigException", ConfigException
+    )
+    return mock
+
+
 class TestCredentials:
     @pytest.mark.parametrize(
         "resource_type,client_type",
@@ -130,6 +148,21 @@ async def test_client_bad_resource_type(self, kubernetes_credentials):
             async with kubernetes_credentials.get_client("shoo-ba-daba-doo"):
                 pass
 
+    async def test_incluster_config(self, mock_cluster_config):
+        kubernetes_credentials = KubernetesCredentials()
+        mock_cluster_config.load_incluster_config.return_value = None
+        async with kubernetes_credentials.get_client("batch"):
+            assert mock_cluster_config.load_incluster_config.called
+            assert not mock_cluster_config.load_kube_config.called
+
+    async def test_load_kube_config(self, mock_cluster_config):
+        kubernetes_credentials = KubernetesCredentials()
+        mock_cluster_config.load_incluster_config.side_effect = ConfigException()
+        mock_cluster_config.load_kube_config.return_value = None
+        async with kubernetes_credentials.get_client("batch"):
+            assert mock_cluster_config.load_incluster_config.called
+            assert mock_cluster_config.load_kube_config.called
+
 
 class TestKubernetesClusterConfig:
     async def test_instantiation_from_file(self, config_file, config_context):