kubeadm: detect empty token lists

Fix #58116
This commit is contained in:
Alberto Planas 2020-08-05 13:03:56 +02:00 committed by Daniel Wozniak
parent 65642b1c4c
commit 6b8f31576c
3 changed files with 27 additions and 17 deletions

1
changelog/58116.fixed Normal file
View file

@ -0,0 +1 @@
Fix kubeadm token_list when the list of tokens is empty

View file

@ -328,20 +328,21 @@ def token_list(kubeconfig=None, rootfs=None):
lines = _cmd(cmd).splitlines()
# Find the header and parse it. We do not need to validate the
# content, as the regex will take care of future changes.
header = lines.pop(0)
header = [i.lower() for i in re.findall(r"(\w+(?:\s\w+)*)", header)]
tokens = []
for line in lines:
# TODO(aplanas): descriptions with multiple spaces can break
# the parser.
values = re.findall(r"(\S+(?:\s\S+)*)", line)
if len(header) != len(values):
log.error("Error parsing line: {}".format(line))
continue
tokens.append({key: value for key, value in zip(header, values)})
if lines:
# Find the header and parse it. We do not need to validate
# the content, as the regex will take care of future changes.
header = lines.pop(0)
header = [i.lower() for i in re.findall(r"(\w+(?:\s\w+)*)", header)]
for line in lines:
# TODO(aplanas): descriptions with multiple spaces can
# break the parser.
values = re.findall(r"(\S+(?:\s\S+)*)", line)
if len(header) != len(values):
log.error("Error parsing line: {}".format(line))
continue
tokens.append({key: value for key, value in zip(header, values)})
return tokens

View file

@ -1,7 +1,3 @@
# -*- coding: utf-8 -*-
from __future__ import absolute_import, print_function, unicode_literals
import pytest
import salt.modules.kubeadm as kubeadm
from salt.exceptions import CommandExecutionError
@ -227,6 +223,18 @@ class KubeAdmTestCase(TestCase, LoaderModuleMockMixin):
with pytest.raises(CommandExecutionError):
assert kubeadm.token_generate()
def test_token_empty(self):
"""
Test kuebadm.token_list when no outout
"""
result = {"retcode": 0, "stdout": ""}
salt_mock = {
"cmd.run_all": MagicMock(return_value=result),
}
with patch.dict(kubeadm.__salt__, salt_mock):
assert kubeadm.token_list() == []
salt_mock["cmd.run_all"].assert_called_with(["kubeadm", "token", "list"])
def test_token_list(self):
"""
Test kuebadm.token_list without parameters