Added optional explicit config parameter to InClusterConfigLoader

kubernetes_asyncio/config/incluster_config.py

@@ -41,9 +41,14 @@ def __init__(self, token_filename,
         self._cert_filename = cert_filename
         self._environ = environ
 
-    def load_and_set(self):
+    def load_and_set(self, client_configuration=None):
         self._load_config()
-        self._set_config()
+        if client_configuration:
+            self._set_config(client_configuration)
+        else:
+            configuration = Configuration()
+            self._set_config(configuration)
+            Configuration.set_default(configuration)
 
     def _load_config(self):
         if (SERVICE_HOST_ENV_NAME not in self._environ or
@@ -76,18 +81,20 @@ def _load_config(self):
 
         self.ssl_ca_cert = self._cert_filename
 
-    def _set_config(self):
-        configuration = Configuration()
+    def _set_config(self, configuration):
         configuration.host = self.host
         configuration.ssl_ca_cert = self.ssl_ca_cert
         configuration.api_key['BearerToken'] = "Bearer " + self.token
-        Configuration.set_default(configuration)
 
 
-def load_incluster_config():
+def load_incluster_config(client_configuration=None):
     """Use the service account kubernetes gives to pods to connect to kubernetes
     cluster. It's intended for clients that expect to be running inside a pod
     running on kubernetes. It will raise an exception if called from a process
-    not running in a kubernetes environment."""
+    not running in a kubernetes environment.
+
+    :param client_configuration: The kubernetes.client.Configuration to
+    set configs to.
+    """
     InClusterConfigLoader(token_filename=SERVICE_TOKEN_FILENAME,
-                          cert_filename=SERVICE_CERT_FILENAME).load_and_set()
+                          cert_filename=SERVICE_CERT_FILENAME).load_and_set(client_configuration)

kubernetes_asyncio/config/incluster_config_test.py

@@ -16,6 +16,8 @@
 import tempfile
 import unittest
 
+from kubernetes_asyncio.client import Configuration
+
 from .config_exception import ConfigException
 from .incluster_config import (
     SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME, InClusterConfigLoader,
@@ -128,6 +130,16 @@ def test_empty_token_file(self):
             token_filename=self._create_file_with_temp_content())
         self._should_fail_load(loader, "empty token file provided")
 
+    def test_client_config(self):
+        cert_filename = self._create_file_with_temp_content(_TEST_CERT)
+        loader = self.get_test_loader(cert_filename=cert_filename)
+        loader._load_config()
+        client_config = Configuration()
+        loader._set_config(client_config)
+        self.assertEqual("https://" + _TEST_HOST_PORT, client_config.host)
+        self.assertEqual(cert_filename, client_config.ssl_ca_cert)
+        self.assertEqual("Bearer " + _TEST_TOKEN, client_config.api_key['BearerToken'])
+
 
 if __name__ == '__main__':
     unittest.main()