fix: change disable_verify_ssl behaviour (#25023)

The problem is that verify_ssl is overwritten by the
configuration from the kube_config or load_incluster_config file.

airflow/kubernetes/kube_client.py

@@ -30,7 +30,10 @@
     has_kubernetes = True
 
     def _disable_verify_ssl() -> None:
-        configuration = Configuration()
+        if hasattr(Configuration, 'get_default_copy'):
+            configuration = Configuration.get_default_copy()
+        else:
+            configuration = Configuration()
         configuration.verify_ssl = False
         Configuration.set_default(configuration)
 
@@ -100,9 +103,6 @@ def get_kube_client(
     if conf.getboolean('kubernetes', 'enable_tcp_keepalive'):
         _enable_tcp_keepalive()
 
-    if not conf.getboolean('kubernetes', 'verify_ssl'):
-        _disable_verify_ssl()
-
     if in_cluster:
         config.load_incluster_config()
     else:
@@ -112,4 +112,7 @@ def get_kube_client(
             config_file = conf.get('kubernetes', 'config_file', fallback=None)
         config.load_kube_config(config_file=config_file, context=cluster_context)
 
+    if not conf.getboolean('kubernetes', 'verify_ssl'):
+        _disable_verify_ssl()
+
     return client.CoreV1Api()

tests/kubernetes/test_client.py

@@ -38,6 +38,19 @@ def test_load_file_config(self, config):
         assert config.load_incluster_config.not_called
         assert config.load_kube_config.called
 
+    @mock.patch('airflow.kubernetes.kube_client.config')
+    @mock.patch('airflow.kubernetes.kube_client.conf')
+    def test_load_config_disable_ssl(self, conf, config):
+        conf.getboolean.return_value = False
+        get_kube_client(in_cluster=False)
+        conf.getboolean.assert_called_with('kubernetes', 'verify_ssl')
+        # Support wide range of kube client libraries
+        if hasattr(Configuration, 'get_default_copy'):
+            configuration = Configuration.get_default_copy()
+        else:
+            configuration = Configuration()
+        self.assertFalse(configuration.verify_ssl)
+
     def test_enable_tcp_keepalive(self):
         socket_options = [
             (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),