Python 3: fix Ansible Vault password helper

In Python 3, the return type of subprocess.check_output is bytes by
default. It needs to be decoded to a string.

Passing universal_newlines=True to the subprocess functions results in
file objects opened in text mode, decoded using the system locale. A
unit test has also been added.

Change-Id: Idb4e66aa1f2c973a956907d7986d1c05c7ce2e67
Story: 2004959
Task: 39483
This commit is contained in:
Mark Goddard 2020-05-20 15:26:35 +01:00
parent 8fb3020827
commit 49c3e1dab4
4 changed files with 19 additions and 5 deletions

View File

@ -144,7 +144,7 @@ class TestCase(unittest.TestCase):
"KAYOBE_VAULT_PASSWORD": "test-pass"} "KAYOBE_VAULT_PASSWORD": "test-pass"}
expected_calls = [ expected_calls = [
mock.call(["which", "kayobe-vault-password-helper"], mock.call(["which", "kayobe-vault-password-helper"],
check_output=True), check_output=True, universal_newlines=True),
mock.call(expected_cmd, check_output=False, quiet=False, mock.call(expected_cmd, check_output=False, quiet=False,
env=expected_env) env=expected_env)
] ]
@ -191,7 +191,8 @@ class TestCase(unittest.TestCase):
ansible.add_args(parser) ansible.add_args(parser)
vault.add_args(parser) vault.add_args(parser)
mock_run.assert_called_once_with( mock_run.assert_called_once_with(
["which", "kayobe-vault-password-helper"], check_output=True) ["which", "kayobe-vault-password-helper"], check_output=True,
universal_newlines=True)
mock_run.reset_mock() mock_run.reset_mock()
parsed_args = parser.parse_args([]) parsed_args = parser.parse_args([])
ansible.run_playbooks(parsed_args, ["playbook1.yml"]) ansible.run_playbooks(parsed_args, ["playbook1.yml"])

View File

@ -113,7 +113,7 @@ class TestCase(unittest.TestCase):
expected_env = {"KAYOBE_VAULT_PASSWORD": "test-pass"} expected_env = {"KAYOBE_VAULT_PASSWORD": "test-pass"}
expected_calls = [ expected_calls = [
mock.call(["which", "kayobe-vault-password-helper"], mock.call(["which", "kayobe-vault-password-helper"],
check_output=True), check_output=True, universal_newlines=True),
mock.call(expected_cmd, shell=True, quiet=False, env=expected_env) mock.call(expected_cmd, shell=True, quiet=False, env=expected_env)
] ]
self.assertEqual(expected_calls, mock_run.mock_calls) self.assertEqual(expected_calls, mock_run.mock_calls)
@ -155,7 +155,8 @@ class TestCase(unittest.TestCase):
kolla_ansible.add_args(parser) kolla_ansible.add_args(parser)
vault.add_args(parser) vault.add_args(parser)
mock_run.assert_called_once_with( mock_run.assert_called_once_with(
["which", "kayobe-vault-password-helper"], check_output=True) ["which", "kayobe-vault-password-helper"], check_output=True,
universal_newlines=True)
mock_run.reset_mock() mock_run.reset_mock()
parsed_args = parser.parse_args([]) parsed_args = parser.parse_args([])
kolla_ansible.run(parsed_args, "command", "overcloud") kolla_ansible.run(parsed_args, "command", "overcloud")

View File

@ -23,6 +23,15 @@ from kayobe import vault
class TestCase(unittest.TestCase): class TestCase(unittest.TestCase):
@mock.patch.object(vault.utils, "run_command", autospec=True)
def test__get_vault_password_helper(self, mock_run):
mock_run.return_value = "fake-password\n"
result = vault._get_vault_password_helper()
mock_run.assert_called_once_with(
["which", "kayobe-vault-password-helper"], check_output=True,
universal_newlines=True)
self.assertEqual('fake-password', result)
def test_validate_args_ok(self): def test_validate_args_ok(self):
parser = argparse.ArgumentParser() parser = argparse.ArgumentParser()
vault.add_args(parser) vault.add_args(parser)

View File

@ -30,7 +30,10 @@ def _get_vault_password_helper():
"""Return the path to the kayobe-vault-password-helper executable.""" """Return the path to the kayobe-vault-password-helper executable."""
cmd = ["which", "kayobe-vault-password-helper"] cmd = ["which", "kayobe-vault-password-helper"]
try: try:
output = utils.run_command(cmd, check_output=True) # NOTE(mgoddard): universal_newlines ensures stdout is opened in text
# mode, and we get a string rather than bytes.
output = utils.run_command(cmd, check_output=True,
universal_newlines=True)
except subprocess.CalledProcessError: except subprocess.CalledProcessError:
return None return None
return output.strip() return output.strip()