Added load_config function (#331)

* Added load config function from https://github.com/kubernetes-client/python/blob/master/kubernetes/base/config/__init__.py\#L24

* Added unit tests.

Unfortunately, the `load_incluster_config` function was difficult to test directly as it was as a few constants would have had to be monkeypatched. To make the function more testable, I made those constants configurable via kwargs.

* Fixed bug with contextmanager in test_load_config_helper

* Fix shadowed import detected by flake8

* Fixed import order using isort

---------

Co-authored-by: James Riley McHugh <mchugh_james@bah.com>

Author: james-mchugh

kubernetes_asyncio/config/__init__.py

@@ -11,10 +11,42 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import warnings
+from os.path import exists, expanduser
 
 from .config_exception import ConfigException
 from .incluster_config import load_incluster_config
 from .kube_config import (
-    list_kube_config_contexts, load_kube_config, load_kube_config_from_dict,
-    new_client_from_config, new_client_from_config_dict, refresh_token,
+    KUBE_CONFIG_DEFAULT_LOCATION, list_kube_config_contexts, load_kube_config,
+    load_kube_config_from_dict, new_client_from_config,
+    new_client_from_config_dict,
 )
+
+
+async def load_config(**kwargs):
+    """
+    Wrapper function to load the kube_config.
+    It will initially try to load_kube_config from provided path,
+    then check if the KUBE_CONFIG_DEFAULT_LOCATION exists
+    If neither exists, it will fall back to load_incluster_config
+    and inform the user accordingly.
+
+    :param kwargs: A combination of all possible kwargs that
+    can be passed to either load_kube_config or
+    load_incluster_config functions.
+    """
+    if "config_file" in kwargs.keys():
+        await load_kube_config(**kwargs)
+    elif "kube_config_path" in kwargs.keys():
+        kwargs["config_file"] = kwargs.pop("kube_config_path", None)
+        await load_kube_config(**kwargs)
+    elif exists(expanduser(KUBE_CONFIG_DEFAULT_LOCATION)):
+        await load_kube_config(**kwargs)
+    else:
+        warnings.warn(
+            "kube_config_path not provided and "
+            "default location ({0}) does not exist. "
+            "Using inCluster Config. "
+            "This might not work.".format(KUBE_CONFIG_DEFAULT_LOCATION)
+        )
+        load_incluster_config(**kwargs)

kubernetes_asyncio/config/incluster_config.py

@@ -114,7 +114,7 @@ def _read_token_file(self):
             ) + TOKEN_REFRESH_PERIOD
 
 
-def load_incluster_config(client_configuration=None, try_refresh_token=True):
+def load_incluster_config(client_configuration=None, try_refresh_token=True, **kwargs):
     """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
@@ -123,7 +123,6 @@ def load_incluster_config(client_configuration=None, try_refresh_token=True):
     :param client_configuration: The kubernetes.client.Configuration to
     set configs to.
     """
-    InClusterConfigLoader(
-        token_filename=SERVICE_TOKEN_FILENAME,
-        cert_filename=SERVICE_CERT_FILENAME,
-        try_refresh_token=try_refresh_token).load_and_set(client_configuration)
+    kwargs.setdefault("token_filename", SERVICE_TOKEN_FILENAME)
+    kwargs.setdefault("cert_filename", SERVICE_CERT_FILENAME)
+    InClusterConfigLoader(try_refresh_token=try_refresh_token, **kwargs).load_and_set(client_configuration)

kubernetes_asyncio/config/incluster_config_test.py

@@ -12,18 +12,21 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import contextlib
 import datetime
 import os
 import tempfile
 import unittest
 
+import kubernetes_asyncio.config
 from kubernetes_asyncio.client import Configuration
 
 from .config_exception import ConfigException
 from .incluster_config import (
     SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME, InClusterConfigLoader,
     _join_host_port,
 )
+from .kube_config_test import FakeConfig
 
 _TEST_TOKEN = "temp_token"
 _TEST_NEW_TOKEN = "temp_new_token"
@@ -40,7 +43,17 @@
                       SERVICE_PORT_ENV_NAME: _TEST_PORT}
 
 
-class InClusterConfigTest(unittest.TestCase):
+@contextlib.contextmanager
+def monkeypatch_kube_config_path():
+    old_kube_config_path = kubernetes_asyncio.config.KUBE_CONFIG_DEFAULT_LOCATION
+    kubernetes_asyncio.config.KUBE_CONFIG_DEFAULT_LOCATION = "/path-does-not-exist"
+    try:
+        yield
+    finally:
+        kubernetes_asyncio.config.KUBE_CONFIG_DEFAULT_LOCATION = old_kube_config_path
+
+
+class InClusterConfigTest(unittest.IsolatedAsyncioTestCase):
 
     def setUp(self):
         self._temp_files = []
@@ -198,6 +211,25 @@ def test_client_config(self):
         self.assertEqual(cert_filename, client_config.ssl_ca_cert)
         self.assertEqual("Bearer " + _TEST_TOKEN, client_config.api_key['BearerToken'])
 
+    async def test_load_config_helper(self):
+        token_filename = self._create_file_with_temp_content(_TEST_TOKEN)
+        cert_filename = self._create_file_with_temp_content(_TEST_CERT)
+        expected = FakeConfig(
+            host="https://" + _TEST_HOST_PORT,
+            token="Bearer " + _TEST_TOKEN,
+            ssl_ca_cert=cert_filename,
+        )
+        actual = FakeConfig()
+        with monkeypatch_kube_config_path():
+            await kubernetes_asyncio.config.load_config(
+                client_configuration=actual,
+                try_refresh_token=None,
+                token_filename=token_filename,
+                cert_filename=cert_filename,
+                environ=_TEST_ENVIRON
+            )
+        self.assertEqual(expected, actual)
+
 
 if __name__ == '__main__':
     unittest.main()

kubernetes_asyncio/config/kube_config_test.py

@@ -24,6 +24,7 @@
 import yaml
 from six import PY3
 
+from . import load_config
 from .config_exception import ConfigException
 from .kube_config import (
     ENV_KUBECONFIG_PATH_SEPARATOR, ConfigNode, FileOrData, KubeConfigLoader,
@@ -322,7 +323,7 @@ def __repr__(self):
             if k in self.FILE_KEYS:
                 try:
                     with open(v) as f:
-                        val = "FILE: %s" % str.decode(f.read())
+                        val = "FILE: %s" % str.encode(f.read())
                 except IOError as e:
                     val = "ERROR: %s" % str(e)
             rep += "\t%s: %s\n" % (k, val)
@@ -1073,6 +1074,16 @@ async def load_from_exec_plugin(self):
 
         self.assertEqual(TEST_ANOTHER_DATA_BASE64, mock_config.api_key["BearerToken"])
 
+    async def test_load_config_helper(self):
+        expected = FakeConfig(host=TEST_HOST,
+                              token=BEARER_TOKEN_FORMAT % TEST_DATA_BASE64)
+        config_file = self._create_temp_file(yaml.safe_dump(self.TEST_KUBE_CONFIG))
+        actual = FakeConfig()
+        await load_config(config_file=config_file,
+                          context="simple_token",
+                          client_configuration=actual)
+        self.assertEqual(expected, actual)
+
 
 class TestKubeConfigMerger(BaseTestCase):
     TEST_KUBE_CONFIG_SET1 = [{