add tests for patch logic

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

@@ -163,10 +163,11 @@ def _get_field(self, field_name):
     def _deprecation_warning_core_param(deprecation_warnings):
         settings_list_str = ''.join([f"\n\t{k}={v!r}" for k, v in deprecation_warnings])
         warnings.warn(
-            f"Using core Airflow settings from section [kubernetes] with the following keys: "
-            f"{settings_list_str}"
-            "In a future release, KubernetesPodOperator will no longer consider core "
-            "airflow settings; define an Airflow connection instead."
+            f"\nApplying core Airflow settings from section [kubernetes] with the following keys:"
+            f"{settings_list_str}\n"
+            "In a future release, KubernetesPodOperator will no longer consider core\n"
+            "airflow settings; define an Airflow connection instead.",
+            DeprecationWarning,
         )
 
     def get_conn(self) -> Any:

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

@@ -313,6 +313,59 @@ def test_client_types(self, mock_kube_config_merger, mock_kube_config_loader):
         assert isinstance(hook.api_client, kubernetes.client.ApiClient)
         assert isinstance(hook.get_conn(), kubernetes.client.ApiClient)
 
+    @patch("airflow.providers.cncf.kubernetes.hooks.kubernetes._disable_verify_ssl")
+    def test_patch_core_settings_verify_ssl(self, mock_disable_verify_ssl):
+        hook = KubernetesHook()
+        hook.get_conn()
+        mock_disable_verify_ssl.assert_not_called()
+        mock_disable_verify_ssl.reset_mock()
+        hook._deprecated_core_disable_verify_ssl = True
+        hook.get_conn()
+        mock_disable_verify_ssl.assert_called()
+
+    @patch("airflow.providers.cncf.kubernetes.hooks.kubernetes._enable_tcp_keepalive")
+    def test_patch_core_settings_tcp_keepalive(self, mock_enable_tcp_keepalive):
+        hook = KubernetesHook()
+        hook.get_conn()
+        mock_enable_tcp_keepalive.assert_called()
+        mock_enable_tcp_keepalive.reset_mock()
+        hook._deprecated_core_disable_tcp_keepalive = True
+        hook.get_conn()
+        mock_enable_tcp_keepalive.assert_not_called()
+
+    @patch("kubernetes.config.kube_config.KubeConfigLoader", new=MagicMock())
+    @patch("kubernetes.config.kube_config.KubeConfigMerger", new=MagicMock())
+    @patch("kubernetes.config.incluster_config.InClusterConfigLoader")
+    @patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.KubernetesHook._get_default_client")
+    def test_patch_core_settings_in_cluster(self, mock_get_default_client, mock_in_cluster_loader):
+        hook = KubernetesHook(conn_id=None)
+        hook.get_conn()
+        mock_in_cluster_loader.assert_not_called()
+        mock_in_cluster_loader.reset_mock()
+        hook._deprecated_core_in_cluster = False
+        hook.get_conn()
+        mock_in_cluster_loader.assert_not_called()
+        mock_get_default_client.assert_called()
+
+    @pytest.mark.parametrize(
+        'key, key_val, attr, attr_val',
+        [
+            ('in_cluster', False, '_deprecated_core_in_cluster', False),
+            ('verify_ssl', False, '_deprecated_core_disable_verify_ssl', True),
+            ('cluster_context', 'hi', '_deprecated_core_cluster_context', 'hi'),
+            ('config_file', '/path/to/file.txt', '_deprecated_core_config_file', '/path/to/file.txt'),
+            ('enable_tcp_keepalive', False, '_deprecated_core_disable_tcp_keepalive', True),
+        ],
+    )
+    @patch("kubernetes.config.incluster_config.InClusterConfigLoader", new=MagicMock())
+    @patch("kubernetes.config.kube_config.KubeConfigLoader", new=MagicMock())
+    @patch("kubernetes.config.kube_config.KubeConfigMerger", new=MagicMock())
+    def test_core_settings_warnings(self, key, key_val, attr, attr_val):
+        hook = KubernetesHook(conn_id=None)
+        setattr(hook, attr, attr_val)
+        with pytest.warns(DeprecationWarning, match=rf'.*Airflow settings.*\n.*{key}={key_val!r}.*') as w:
+            hook.get_conn()
+
 
 class TestKubernetesHookIncorrectConfiguration:
     @pytest.mark.parametrize(