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

Fix merging current-context in kube-configs

Browse Source
This commit is contained in:
Tomasz Prus 2024-01-31 23:48:15 +01:00
parent df43d155d9
commit 69cfcda963
2 changed files with 66 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
kubernetes/base/config/kube_config.py
View File

@ -727,6 +727,10 @@ class KubeConfigMerger:
self.config_merged = ConfigNode(path, config_merged, path)
for item in ('clusters', 'contexts', 'users'):
self._merge(item, config.get(item, []) or [], path)
if 'current-context' in config:
self.config_merged.value['current-context'] = config['current-context']
self.config_files[path] = config
def _merge(self, item, add_cfg, path):

88
kubernetes/base/config/kube_config_test.py
View File

@ -1653,7 +1653,7 @@ class TestKubernetesClientConfiguration(BaseTestCase):
class TestKubeConfigMerger(BaseTestCase):
TEST_KUBE_CONFIG_PART1 = {
TEST_KUBE_CONFIG_SET1 = [{
"current-context": "no_user",
"contexts": [
{
@ -1672,9 +1672,7 @@ class TestKubeConfigMerger(BaseTestCase):
},
],
"users": []
}
TEST_KUBE_CONFIG_PART2 = {
}, {
"current-context": "",
"contexts": [
{
@ -1712,9 +1710,7 @@ class TestKubeConfigMerger(BaseTestCase):
}
},
]
}
TEST_KUBE_CONFIG_PART3 = {
}, {
"current-context": "no_user",
"contexts": [
{
@ -1761,12 +1757,10 @@ class TestKubeConfigMerger(BaseTestCase):
}
},
]
}
TEST_KUBE_CONFIG_PART4 = {
}, {
"current-context": "no_user",
}
# Config with user having cmd-path
TEST_KUBE_CONFIG_PART5 = {
}, {
# Config with user having cmd-path
"contexts": [
{
"name": "contexttestcmdpath",
@ -1795,8 +1789,7 @@ class TestKubeConfigMerger(BaseTestCase):
}
}
]
}
TEST_KUBE_CONFIG_PART6 = {
}, {
"current-context": "no_user",
"contexts": [
{
@ -1815,22 +1808,49 @@ class TestKubeConfigMerger(BaseTestCase):
},
],
"users": None
}
}]
# 3 parts with different keys/data to merge
TEST_KUBE_CONFIG_SET2 = [{
"clusters": [
{
"name": "default",
"cluster": {
"server": TEST_HOST
}
},
],
}, {
"current-context": "simple_token",
"contexts": [
{
"name": "simple_token",
"context": {
"cluster": "default",
"user": "simple_token"
}
},
],
}, {
"users": [
{
"name": "simple_token",
"user": {
"token": TEST_DATA_BASE64,
"username": TEST_USERNAME,
"password": TEST_PASSWORD,
}
},
]
}]
def _create_multi_config(self):
def _create_multi_config(self, parts):
files = []
for part in (
self.TEST_KUBE_CONFIG_PART1,
self.TEST_KUBE_CONFIG_PART2,
self.TEST_KUBE_CONFIG_PART3,
self.TEST_KUBE_CONFIG_PART4,
self.TEST_KUBE_CONFIG_PART5,
self.TEST_KUBE_CONFIG_PART6):
for part in parts:
files.append(self._create_temp_file(yaml.safe_dump(part)))
return ENV_KUBECONFIG_PATH_SEPARATOR.join(files)
def test_list_kube_config_contexts(self):
kubeconfigs = self._create_multi_config()
kubeconfigs = self._create_multi_config(self.TEST_KUBE_CONFIG_SET1)
expected_contexts = [
{'context': {'cluster': 'default'}, 'name': 'no_user'},
{'context': {'cluster': 'ssl', 'user': 'ssl'}, 'name': 'ssl'},
@ -1849,15 +1869,31 @@ class TestKubeConfigMerger(BaseTestCase):
self.assertEqual(active_context, expected_contexts[0])
def test_new_client_from_config(self):
kubeconfigs = self._create_multi_config()
kubeconfigs = self._create_multi_config(self.TEST_KUBE_CONFIG_SET1)
client = new_client_from_config(
config_file=kubeconfigs, context="simple_token")
self.assertEqual(TEST_HOST, client.configuration.host)
self.assertEqual(BEARER_TOKEN_FORMAT % TEST_DATA_BASE64,
client.configuration.api_key['authorization'])
def test_merge_with_context_in_different_file(self):
kubeconfigs = self._create_multi_config(self.TEST_KUBE_CONFIG_SET2)
client = new_client_from_config(config_file=kubeconfigs)
expected_contexts = [
{'context': {'cluster': 'default', 'user': 'simple_token'},
'name': 'simple_token'}
]
contexts, active_context = list_kube_config_contexts(
config_file=kubeconfigs)
self.assertEqual(contexts, expected_contexts)
self.assertEqual(active_context, expected_contexts[0])
self.assertEqual(TEST_HOST, client.configuration.host)
self.assertEqual(BEARER_TOKEN_FORMAT % TEST_DATA_BASE64,
client.configuration.api_key['authorization'])
def test_save_changes(self):
kubeconfigs = self._create_multi_config()
kubeconfigs = self._create_multi_config(self.TEST_KUBE_CONFIG_SET1)
# load configuration, update token, save config
kconf = KubeConfigMerger(kubeconfigs)