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

Run kubeconfig exec commands in the correct directory. This fixes configs that rely on relative paths.

Browse Source
This commit is contained in:
April Schleck 2021-12-23 14:46:23 -08:00
parent f0fa950bb2
commit 1c5bf586f0
3 changed files with 20 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
config/exec_provider.py
View File

@ -31,7 +31,7 @@ class ExecProvider(object):
* caching
"""
def __init__(self, exec_config):
def __init__(self, exec_config, cwd):
"""
exec_config must be of type ConfigNode because we depend on
safe_get(self, key) to correctly handle optional exec provider
@ -53,6 +53,7 @@ class ExecProvider(object):
value = item['value']
additional_vars[name] = value
self.env.update(additional_vars)
self.cwd = cwd
def run(self, previous_response=None):
kubernetes_exec_info = {
@ -69,6 +70,7 @@ class ExecProvider(object):
self.args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
cwd=self.cwd,
env=self.env,
universal_newlines=True)
(stdout, stderr) = process.communicate()

21
config/exec_provider_test.py
View File

@ -47,7 +47,7 @@ class ExecProviderTest(unittest.TestCase):
ConfigNode('test3', {'apiVersion': ''})]
for exec_config in exec_configs:
with self.assertRaises(ConfigException) as context:
ExecProvider(exec_config)
ExecProvider(exec_config, None)
self.assertIn('exec: malformed request. missing key',
context.exception.args[0])
@ -57,7 +57,7 @@ class ExecProviderTest(unittest.TestCase):
instance.wait.return_value = 1
instance.communicate.return_value = ('', '')
with self.assertRaises(ConfigException) as context:
ep = ExecProvider(self.input_ok)
ep = ExecProvider(self.input_ok, None)
ep.run()
self.assertIn('exec: process returned %d' %
instance.wait.return_value, context.exception.args[0])
@ -68,7 +68,7 @@ class ExecProviderTest(unittest.TestCase):
instance.wait.return_value = 0
instance.communicate.return_value = ('', '')
with self.assertRaises(ConfigException) as context:
ep = ExecProvider(self.input_ok)
ep = ExecProvider(self.input_ok, None)
ep.run()
self.assertIn('exec: failed to decode process output',
context.exception.args[0])
@ -102,7 +102,7 @@ class ExecProviderTest(unittest.TestCase):
for output in outputs:
instance.communicate.return_value = (output, '')
with self.assertRaises(ConfigException) as context:
ep = ExecProvider(self.input_ok)
ep = ExecProvider(self.input_ok, None)
ep.run()
self.assertIn('exec: malformed response. missing key',
context.exception.args[0])
@ -123,7 +123,7 @@ class ExecProviderTest(unittest.TestCase):
""" % wrong_api_version
instance.communicate.return_value = (output, '')
with self.assertRaises(ConfigException) as context:
ep = ExecProvider(self.input_ok)
ep = ExecProvider(self.input_ok, None)
ep.run()
self.assertIn(
'exec: plugin api version %s does not match' %
@ -135,11 +135,20 @@ class ExecProviderTest(unittest.TestCase):
instance = mock.return_value
instance.wait.return_value = 0
instance.communicate.return_value = (self.output_ok, '')
ep = ExecProvider(self.input_ok)
ep = ExecProvider(self.input_ok, None)
result = ep.run()
self.assertTrue(isinstance(result, dict))
self.assertTrue('token' in result)
@mock.patch('subprocess.Popen')
def test_run_in_dir(self, mock):
instance = mock.return_value
instance.wait.return_value = 0
instance.communicate.return_value = (self.output_ok, '')
ep = ExecProvider(self.input_ok, '/some/directory')
ep.run()
self.assertEqual(mock.call_args.kwargs['cwd'], '/some/directory')
if __name__ == '__main__':
unittest.main()

4
config/kube_config.py
View File

@ -483,7 +483,8 @@ class KubeConfigLoader(object):
if 'exec' not in self._user:
return
try:
status = ExecProvider(self._user['exec']).run()
base_path = self._get_base_path(self._cluster.path)
status = ExecProvider(self._user['exec'], base_path).run()
if 'token' in status:
self.token = "Bearer %s" % status['token']
elif 'clientCertificateData' in status:
@ -493,7 +494,6 @@ class KubeConfigLoader(object):
logging.error('exec: missing clientKeyData field in '
'plugin output')
return None
base_path = self._get_base_path(self._cluster.path)
self.cert_file = FileOrData(
status, None,
data_key_name='clientCertificateData',