From c40604be6cfb8c043789ebd3d1ab97fcaad79dff Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 10 May 2023 10:28:55 -0700
Subject: [PATCH 1/7] KubernetesHook should work even if "kubernetes_default"
 is missing

It's common when using KPO / K8s hook to want to simply use the RBAC of the cluster.  This was the default behavior prior to #28848.  After that change, users are forced to add a k8s connection or their dags will break.

To fix this, in the special case where conn_id=="kubernetes_default", if the conn is missing, we ignore the failure.
---
 .../cncf/kubernetes/hooks/kubernetes.py       | 13 +++++++-
 ...t_kubernetes_pod.py => test_kubernetes.py} | 33 +++++++++++++++++++
 2 files changed, 45 insertions(+), 1 deletion(-)
 rename tests/providers/cncf/kubernetes/hooks/{test_kubernetes_pod.py => test_kubernetes.py} (95%)

diff --git a/airflow/providers/cncf/kubernetes/hooks/kubernetes.py b/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
index 045a7aafb286a..d5360272db4ed 100644
--- a/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
+++ b/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
@@ -30,9 +30,10 @@
 from urllib3.exceptions import HTTPError
 
 from airflow.compat.functools import cached_property
-from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.exceptions import AirflowException, AirflowNotFoundException, AirflowProviderDeprecationWarning
 from airflow.hooks.base import BaseHook
 from airflow.kubernetes.kube_client import _disable_verify_ssl, _enable_tcp_keepalive
+from airflow.models import Connection
 from airflow.utils import yaml
 
 LOADING_KUBE_CONFIG_FILE_RESOURCE = "Loading Kubernetes configuration file kube_config from {}..."
@@ -141,6 +142,16 @@ def _coalesce_param(*params):
             if param is not None:
                 return param
 
+    @classmethod
+    def get_connection(cls, conn_id: str) -> Connection:
+        try:
+            return super().get_connection(conn_id)
+        except AirflowNotFoundException as e:
+            if conn_id == cls.default_conn_name:
+                return Connection(conn_id=cls.default_conn_name)
+            else:
+                raise e
+
     @cached_property
     def conn_extras(self):
         if self.conn_id:
diff --git a/tests/providers/cncf/kubernetes/hooks/test_kubernetes_pod.py b/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
similarity index 95%
rename from tests/providers/cncf/kubernetes/hooks/test_kubernetes_pod.py
rename to tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
index 9ccca08abcad5..a878dc98e083f 100644
--- a/tests/providers/cncf/kubernetes/hooks/test_kubernetes_pod.py
+++ b/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
@@ -27,8 +27,11 @@
 import kubernetes
 import pytest
 from kubernetes.config import ConfigException
+from sqlalchemy.orm import make_transient
 
 from airflow import AirflowException
+from airflow.exceptions import AirflowNotFoundException
+from airflow.hooks.base import BaseHook
 from airflow.models import Connection
 from airflow.providers.cncf.kubernetes.hooks.kubernetes import AsyncKubernetesHook, KubernetesHook
 from airflow.utils import db
@@ -54,6 +57,22 @@ class DeprecationRemovalRequired(AirflowException):
     ...
 
 
+DEFAULT_CONN_ID = "kubernetes_default"
+
+
+@pytest.fixture()
+def remove_default_conn(session):
+    before_conn = session.query(Connection).filter(Connection.conn_id == DEFAULT_CONN_ID).one_or_none()
+    if before_conn:
+        session.delete(before_conn)
+        session.commit()
+    yield
+    if before_conn:
+        make_transient(before_conn)
+        session.add(before_conn)
+        session.commit()
+
+
 class TestKubernetesHook:
     @classmethod
     def setup_class(cls) -> None:
@@ -391,6 +410,20 @@ def test_prefixed_names_still_work(self, mock_get_client):
             mock_get_client.assert_called_with(cluster_context="test")
             assert kubernetes_hook.get_namespace() == "test"
 
+    def test_missing_default_connection_is_ok(self, remove_default_conn):
+        # prove to ourselves that the default conn no exist
+        with pytest.raises(AirflowNotFoundException):
+            BaseHook.get_connection(DEFAULT_CONN_ID)
+
+        # verify K8sHook still works
+        hook = KubernetesHook()
+        assert hook.conn_extras == {}
+
+        # meanwhile, asking for non-default should still fail if no exist
+        hook = KubernetesHook("some_conn")
+        with pytest.raises(AirflowNotFoundException, match="The conn_id `some_conn` isn't defined"):
+            hook.conn_extras
+
     @patch("kubernetes.config.kube_config.KubeConfigLoader")
     @patch("kubernetes.config.kube_config.KubeConfigMerger")
     @patch(f"{HOOK_MODULE}.client.CustomObjectsApi")

From 297a299cd887c6f22a2a1a2ad63a2312c408f70c Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 10 May 2023 10:34:10 -0700
Subject: [PATCH 2/7] add release notes

---
 airflow/providers/cncf/kubernetes/CHANGELOG.rst | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/airflow/providers/cncf/kubernetes/CHANGELOG.rst b/airflow/providers/cncf/kubernetes/CHANGELOG.rst
index c129f99e1bb0b..b2576d4dc9752 100644
--- a/airflow/providers/cncf/kubernetes/CHANGELOG.rst
+++ b/airflow/providers/cncf/kubernetes/CHANGELOG.rst
@@ -39,6 +39,14 @@ Features
    * ``Deprecate 'skip_exit_code' in 'DockerOperator' and 'KubernetesPodOperator' (#30733)``
   * ``Remove skip_exit_code from KubernetesPodOperator (#30788)``
 
+6.1.0
+.....
+
+Features
+~~~~~~~~
+
+If ``kubernetes_default`` connection is not defined, then KubernetesHook / KubernetesPodOperator will behave as though given ``conn_id=None``.
+
 6.0.0
 .....
 

From 43cb5a234388d8ba08d2140b4fdd509c14da4d7e Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 10 May 2023 10:43:04 -0700
Subject: [PATCH 3/7] docstring

---
 airflow/providers/cncf/kubernetes/hooks/kubernetes.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/airflow/providers/cncf/kubernetes/hooks/kubernetes.py b/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
index d5360272db4ed..1cfa751ee5daf 100644
--- a/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
+++ b/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
@@ -144,6 +144,12 @@ def _coalesce_param(*params):
 
     @classmethod
     def get_connection(cls, conn_id: str) -> Connection:
+        """
+        Return requested connection.
+
+        If missing and conn_id is "kubernetes_default", will return empty connection so that hook will
+        default to cluster-derived creds.
+        """
         try:
             return super().get_connection(conn_id)
         except AirflowNotFoundException as e:

From 6ed38c9f5cb40364d2a2029422d0ec714643617a Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 10 May 2023 11:21:37 -0700
Subject: [PATCH 4/7] typo

---
 airflow/providers/cncf/kubernetes/hooks/kubernetes.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/airflow/providers/cncf/kubernetes/hooks/kubernetes.py b/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
index 1cfa751ee5daf..82f38b301b7ce 100644
--- a/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
+++ b/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
@@ -148,7 +148,7 @@ def get_connection(cls, conn_id: str) -> Connection:
         Return requested connection.
 
         If missing and conn_id is "kubernetes_default", will return empty connection so that hook will
-        default to cluster-derived creds.
+        default to cluster-derived credentials.
         """
         try:
             return super().get_connection(conn_id)

From 03b2ee182baea35aece9273f9fcbc31e129afe7d Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 10 May 2023 13:19:33 -0700
Subject: [PATCH 5/7] Update
 tests/providers/cncf/kubernetes/hooks/test_kubernetes.py

Co-authored-by: Jed Cunningham <66968678+jedcunningham@users.noreply.github.com>
---
 tests/providers/cncf/kubernetes/hooks/test_kubernetes.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py b/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
index a878dc98e083f..95914c574e98f 100644
--- a/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
+++ b/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
@@ -419,7 +419,7 @@ def test_missing_default_connection_is_ok(self, remove_default_conn):
         hook = KubernetesHook()
         assert hook.conn_extras == {}
 
-        # meanwhile, asking for non-default should still fail if no exist
+        # meanwhile, asking for non-default should still fail if it doesn't exist
         hook = KubernetesHook("some_conn")
         with pytest.raises(AirflowNotFoundException, match="The conn_id `some_conn` isn't defined"):
             hook.conn_extras

From bb4ebd2cc48d67618c5008cd0e0ac52d36ae86f5 Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 10 May 2023 13:19:41 -0700
Subject: [PATCH 6/7] Update
 tests/providers/cncf/kubernetes/hooks/test_kubernetes.py

Co-authored-by: Jed Cunningham <66968678+jedcunningham@users.noreply.github.com>
---
 tests/providers/cncf/kubernetes/hooks/test_kubernetes.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py b/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
index 95914c574e98f..63740af2e4b74 100644
--- a/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
+++ b/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
@@ -411,7 +411,7 @@ def test_prefixed_names_still_work(self, mock_get_client):
             assert kubernetes_hook.get_namespace() == "test"
 
     def test_missing_default_connection_is_ok(self, remove_default_conn):
-        # prove to ourselves that the default conn no exist
+        # prove to ourselves that the default conn doesn't exist
         with pytest.raises(AirflowNotFoundException):
             BaseHook.get_connection(DEFAULT_CONN_ID)
 

From 165faab59f5e12fadea019ce32574b8eb7571b43 Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 10 May 2023 13:19:54 -0700
Subject: [PATCH 7/7] Update
 airflow/providers/cncf/kubernetes/hooks/kubernetes.py

Co-authored-by: Jed Cunningham <66968678+jedcunningham@users.noreply.github.com>
---
 airflow/providers/cncf/kubernetes/hooks/kubernetes.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/airflow/providers/cncf/kubernetes/hooks/kubernetes.py b/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
index 82f38b301b7ce..fb2cd11bd0dc0 100644
--- a/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
+++ b/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
@@ -152,11 +152,11 @@ def get_connection(cls, conn_id: str) -> Connection:
         """
         try:
             return super().get_connection(conn_id)
-        except AirflowNotFoundException as e:
+        except AirflowNotFoundException:
             if conn_id == cls.default_conn_name:
                 return Connection(conn_id=cls.default_conn_name)
             else:
-                raise e
+                raise
 
     @cached_property
     def conn_extras(self):
