kubernetes-client · mbohlool · Dec 2, 2016 · Nov 30, 2016 · Nov 30, 2016 · Nov 30, 2016
diff --git a/examples/example4.py b/examples/example4.py
@@ -0,0 +1,62 @@
+# Copyright 2016 The Kubernetes Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# 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 os
+
+from kubernetes import client, config
+from kubernetes.client import configuration
+
+
+def main():
+    config_file = os.path.join(os.path.expanduser('~'), '.kube', 'config')
+    contexts, active_context = config.list_kube_config_contexts(config_file)
+    if not contexts:
+        print("Cannot find any context in kube-config file.")
+        return
+    contexts = [context['name'] for context in contexts]
+    active_context = active_context['name']
+    for i, context in enumerate(contexts):
+        format_str = "%d. %s"
+        if context == active_context:
+            format_str = "* " + format_str
+        print(format_str % (i, context))
+    context = input("Enter context number: ")
+    context = int(context)
+    if context not in range(len(contexts)):
+        print(
+            "Number out of range. Using default context %s." %
+            active_context)
+        context_name = active_context
+    else:
+        context_name = contexts[context]
+
+    # Configs can be set in Configuration class directly or using helper
+    # utility
+    config.load_kube_config(config_file, context_name)
+
+    print("Active host is %s" % configuration.host)
+
+    v1 = client.CoreV1Api()
+    print("Listing pods with their IPs:")
+    ret = v1.list_pod_for_all_namespaces(watch=False)
+    for item in ret.items:
+        print(
+            "%s\t%s\t%s" %
+            (item.status.pod_ip,
+             item.metadata.namespace,
+             item.metadata.name))
+
+
+if __name__ == '__main__':
+    main()
diff --git a/kubernetes/config/__init__.py b/kubernetes/config/__init__.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from .kube_config import load_kube_config
+from .config_exception import ConfigException
 from .incluster_config import load_incluster_config
-from .incluster_config import ConfigException
+from .kube_config import list_kube_config_contexts
+from .kube_config import load_kube_config
diff --git a/kubernetes/config/config_exception.py b/kubernetes/config/config_exception.py
@@ -0,0 +1,17 @@
+# Copyright 2016 The Kubernetes Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# 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.
+
+
+class ConfigException(Exception):
+    pass
diff --git a/kubernetes/config/incluster_config.py b/kubernetes/config/incluster_config.py
@@ -16,10 +16,12 @@
 
 from kubernetes.client import configuration
 
-_SERVICE_HOST_ENV_NAME = "KUBERNETES_SERVICE_HOST"
-_SERVICE_PORT_ENV_NAME = "KUBERNETES_SERVICE_PORT"
-_SERVICE_TOKEN_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/token"
-_SERVICE_CERT_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
+from .config_exception import ConfigException
+
+SERVICE_HOST_ENV_NAME = "KUBERNETES_SERVICE_HOST"
+SERVICE_PORT_ENV_NAME = "KUBERNETES_SERVICE_PORT"
+SERVICE_TOKEN_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/token"
+SERVICE_CERT_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
 
 
 def _join_host_port(host, port):
@@ -31,16 +33,10 @@ def _join_host_port(host, port):
     return template % (host, port)
 
 
-class ConfigException(Exception):
-    pass
-
-
 class InClusterConfigLoader(object):
 
-    def __init__(self, host_env_name, port_env_name, token_filename,
+    def __init__(self, token_filename,
                  cert_filename, environ=os.environ):
-        self._host_env_name = host_env_name
-        self._port_env_name = port_env_name
         self._token_filename = token_filename
         self._cert_filename = cert_filename
         self._environ = environ
@@ -50,24 +46,34 @@ def load_and_set(self):
         self._set_config()
 
     def _load_config(self):
-        if (self._host_env_name not in self._environ or
-                self._port_env_name not in self._environ):
+        if (SERVICE_HOST_ENV_NAME not in self._environ or
+                SERVICE_PORT_ENV_NAME not in self._environ):
             raise ConfigException("Service host/port is not set.")
 
+        if (not self._environ[SERVICE_HOST_ENV_NAME] or
+                not self._environ[SERVICE_PORT_ENV_NAME]):
+            raise ConfigException("Service host/port is set but empty.")
+
         self.host = (
-            "https://" + _join_host_port(self._environ[self._host_env_name],
-                                         self._environ[self._port_env_name]))
+            "https://" + _join_host_port(self._environ[SERVICE_HOST_ENV_NAME],
+                                         self._environ[SERVICE_PORT_ENV_NAME]))
 
         if not os.path.isfile(self._token_filename):
             raise ConfigException("Service token file does not exists.")
 
         with open(self._token_filename) as f:
             self.token = f.read()
+            if not self.token:
+                raise ConfigException("Token file exists but empty.")
 
         if not os.path.isfile(self._cert_filename):
             raise ConfigException(
                 "Service certification file does not exists.")
 
+        with open(self._cert_filename) as f:
+            if not f.read():
+                raise ConfigException("Cert file exists but empty.")
+    
         self.ssl_ca_cert = self._cert_filename
 
     def _set_config(self):
@@ -81,7 +87,5 @@ def load_incluster_config():
     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."""
-    InClusterConfigLoader(host_env_name=_SERVICE_HOST_ENV_NAME,
-                          port_env_name=_SERVICE_PORT_ENV_NAME,
-                          token_filename=_SERVICE_TOKEN_FILENAME,
-                          cert_filename=_SERVICE_CERT_FILENAME).load_and_set()
+    InClusterConfigLoader(token_filename=SERVICE_TOKEN_FILENAME,
+                          cert_filename=SERVICE_CERT_FILENAME).load_and_set()
diff --git a/kubernetes/config/incluster_config_test.py b/kubernetes/config/incluster_config_test.py
@@ -16,19 +16,22 @@
 import tempfile
 import unittest
 
-from kubernetes.client import configuration
-
-from .incluster_config import (_SERVICE_HOST_ENV_NAME, _SERVICE_PORT_ENV_NAME,
-                               ConfigException, InClusterConfigLoader)
+from .config_exception import ConfigException
+from .incluster_config import (SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME,
+                               InClusterConfigLoader, _join_host_port)
 
 _TEST_TOKEN = "temp_token"
+_TEST_CERT = "temp_cert"
 _TEST_HOST = "127.0.0.1"
-_TEST_IPV6_HOST = "::1"
 _TEST_PORT = "80"
-_TEST_ENVIRON = {_SERVICE_HOST_ENV_NAME: _TEST_HOST,
-                 _SERVICE_PORT_ENV_NAME: _TEST_PORT}
-_TEST_IPV6_ENVIRON = {_SERVICE_HOST_ENV_NAME: _TEST_IPV6_HOST,
-                      _SERVICE_PORT_ENV_NAME: _TEST_PORT}
+_TEST_HOST_PORT = "127.0.0.1:80"
+_TEST_IPV6_HOST = "::1"
+_TEST_IPV6_HOST_PORT = "[::1]:80"
+
+_TEST_ENVIRON = {SERVICE_HOST_ENV_NAME: _TEST_HOST,
+                 SERVICE_PORT_ENV_NAME: _TEST_PORT}
+_TEST_IPV6_ENVIRON = {SERVICE_HOST_ENV_NAME: _TEST_IPV6_HOST,
+                      SERVICE_PORT_ENV_NAME: _TEST_PORT}
 
 
 class InClusterConfigTest(unittest.TestCase):
@@ -49,38 +52,29 @@ def _create_file_with_temp_content(self, content=""):
 
     def get_test_loader(
             self,
-            host_env_name=_SERVICE_HOST_ENV_NAME,
-            port_env_name=_SERVICE_PORT_ENV_NAME,
             token_filename=None,
             cert_filename=None,
             environ=_TEST_ENVIRON):
         if not token_filename:
             token_filename = self._create_file_with_temp_content(_TEST_TOKEN)
         if not cert_filename:
-            cert_filename = self._create_file_with_temp_content()
+            cert_filename = self._create_file_with_temp_content(_TEST_CERT)
         return InClusterConfigLoader(
-            host_env_name=host_env_name,
-            port_env_name=port_env_name,
             token_filename=token_filename,
             cert_filename=cert_filename,
             environ=environ)
 
+    def test_join_host_port(self):
+        self.assertEqual(_TEST_HOST_PORT,
+                         _join_host_port(_TEST_HOST, _TEST_PORT))
+        self.assertEqual(_TEST_IPV6_HOST_PORT,
+                         _join_host_port(_TEST_IPV6_HOST, _TEST_PORT))
+
     def test_load_config(self):
-        cert_filename = self._create_file_with_temp_content()
+        cert_filename = self._create_file_with_temp_content(_TEST_CERT)
         loader = self.get_test_loader(cert_filename=cert_filename)
         loader._load_config()
-        self.assertEqual("https://%s:%s" % (_TEST_HOST, _TEST_PORT),
-                         loader.host)
-        self.assertEqual(cert_filename, loader.ssl_ca_cert)
-        self.assertEqual(_TEST_TOKEN, loader.token)
-
-    def test_load_config_with_bracketed_hostname(self):
-        cert_filename = self._create_file_with_temp_content()
-        loader = self.get_test_loader(cert_filename=cert_filename,
-                                      environ=_TEST_IPV6_ENVIRON)
-        loader._load_config()
-        self.assertEqual("https://[%s]:%s" % (_TEST_IPV6_HOST, _TEST_PORT),
-                         loader.host)
+        self.assertEqual("https://" + _TEST_HOST_PORT, loader.host)
         self.assertEqual(cert_filename, loader.ssl_ca_cert)
         self.assertEqual(_TEST_TOKEN, loader.token)
 
@@ -93,21 +87,45 @@ def _should_fail_load(self, config_loader, reason):
             pass
 
     def test_no_port(self):
-        loader = self.get_test_loader(port_env_name="not_exists_port")
+        loader = self.get_test_loader(
+            environ={SERVICE_HOST_ENV_NAME: _TEST_HOST})
         self._should_fail_load(loader, "no port specified")
 
+    def test_empty_port(self):
+        loader = self.get_test_loader(
+            environ={SERVICE_HOST_ENV_NAME: _TEST_HOST,
+                     SERVICE_PORT_ENV_NAME: ""})
+        self._should_fail_load(loader, "empty port specified")
+
     def test_no_host(self):
-        loader = self.get_test_loader(host_env_name="not_exists_host")
+        loader = self.get_test_loader(
+            environ={SERVICE_PORT_ENV_NAME: _TEST_PORT})
         self._should_fail_load(loader, "no host specified")
 
+    def test_empty_host(self):
+        loader = self.get_test_loader(
+            environ={SERVICE_HOST_ENV_NAME: "",
+                     SERVICE_PORT_ENV_NAME: _TEST_PORT})
+        self._should_fail_load(loader, "empty host specified")
+
     def test_no_cert_file(self):
         loader = self.get_test_loader(cert_filename="not_exists_file_1123")
         self._should_fail_load(loader, "cert file does not exists")
 
+    def test_empty_cert_file(self):
+        loader = self.get_test_loader(
+            cert_filename=self._create_file_with_temp_content())
+        self._should_fail_load(loader, "empty cert file provided")
+
     def test_no_token_file(self):
         loader = self.get_test_loader(token_filename="not_exists_file_1123")
         self._should_fail_load(loader, "token file does not exists")
 
+    def test_empty_token_file(self):
+        loader = self.get_test_loader(
+            token_filename=self._create_file_with_temp_content())
+        self._should_fail_load(loader, "empty token file provided")
+
 
 if __name__ == '__main__':
     unittest.main()