kubernetes-client/python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

commiting changes to branch

Browse Source
This commit is contained in:
Dylan Shepard 2020-08-19 12:36:32 -07:00
parent 54d188f89e
commit f65f06b1ed
No known key found for this signature in database
GPG Key ID: F8150643D5643467
2 changed files with 36 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
config/kube_config.py
View File

@ -19,6 +19,7 @@ import datetime
import json
import logging
import os
import io
import platform
import subprocess
import tempfile
@ -667,19 +668,31 @@ class KubeConfigMerger:
self.paths = []
self.config_files = {}
self.config_merged = None
for path in paths.split(ENV_KUBECONFIG_PATH_SEPARATOR):
if path:
path = os.path.expanduser(path)
if os.path.exists(path):
self.paths.append(path)
self.load_config(path)
self.config_saved = copy.deepcopy(self.config_files)
if hasattr(paths, 'read'):
self.load_config_from_fileish(paths)
else:
for path in paths.split(ENV_KUBECONFIG_PATH_SEPARATOR):
if path:
path = os.path.expanduser(path)
if os.path.exists(path):
self.paths.append(path)
self.load_config(path)
self.config_saved = copy.deepcopy(self.config_files)
@property
def config(self):
return self.config_merged
def load_config_from_fileish(self, string):
if hasattr(string, 'getvalue'):
config = yaml.safe_load(string.getvalue())
else:
config = yaml.safe_load(string.read())
if self.config_merged is None:
self.config_merged = copy.deepcopy(config)
# doesn't need to do any further merging
def load_config(self, path):
with open(path) as f:
config = yaml.safe_load(f)

15
config/kube_config_test.py
View File

@ -16,6 +16,7 @@ import base64
import datetime
import json
import os
import io
import shutil
import tempfile
import unittest
@ -1257,6 +1258,14 @@ class TestKubeConfigLoader(BaseTestCase):
client_configuration=actual)
self.assertEqual(expected, actual)
def test_load_kube_config_from_stringio(self):
expected = FakeConfig(host=TEST_HOST,
token=BEARER_TOKEN_FORMAT % TEST_DATA_BASE64)
kubeconfig = self._create_stringio_config()
actual = FakeConfig()
load_kube_config(config_file=kubeconfig, context="simple_token", client_configuration=actual)
self.assertEqual(expected, actual)
def test_load_kube_config_from_dict(self):
expected = FakeConfig(host=TEST_HOST,
token=BEARER_TOKEN_FORMAT % TEST_DATA_BASE64)
@ -1633,6 +1642,11 @@ class TestKubeConfigMerger(BaseTestCase):
files.append(self._create_temp_file(yaml.safe_dump(part)))
return ENV_KUBECONFIG_PATH_SEPARATOR.join(files)
def _create_stringio_config(self):
obj = io.StringIO()
obj.write(self.TEST_KUBE_CONFIG_PART1)
return obj
def test_list_kube_config_contexts(self):
kubeconfigs = self._create_multi_config()
expected_contexts = [
@ -1660,6 +1674,7 @@ class TestKubeConfigMerger(BaseTestCase):
self.assertEqual(BEARER_TOKEN_FORMAT % TEST_DATA_BASE64,
client.configuration.api_key['authorization'])
def test_save_changes(self):
kubeconfigs = self._create_multi_config()