From c20a86d550c22a8d4bf7b4f413d14d96eccb43ef Mon Sep 17 00:00:00 2001
From: Csaba Henk <chenk@redhat.com>
Date: Sun, 19 Oct 2014 01:01:01 +0200
Subject: [PATCH] processutils: execute(): fix option incompatibility
Issue: simultaneous usage of 'shell' and 'run_as_root' options led
to error.
Cause: it's actually a concealed TypeError: in 'shell'
mode the command argument is assumed to be a string, elsewhere
a list/tuple.
Fix: the command editing implied by 'run_as_root' is performed
with care taken of above tacit type assumption.
Change-Id: Iaba33e6fda67793ae2874ba278c990271ee5e47f
Closes-Bug: #1382873
---
oslo_concurrency/processutils.py | 7 ++++++-
tests/test_processutils.py | 12 ++++++++++++
2 files changed, 18 insertions(+), 1 deletion(-)
diff --git a/oslo_concurrency/processutils.py b/oslo_concurrency/processutils.py
index 78f8f14..9892aba 100644
--- a/oslo_concurrency/processutils.py
+++ b/oslo_concurrency/processutils.py
@@ -189,7 +189,12 @@ def execute(*cmd, **kwargs):
raise NoRootWrapSpecified(
message=_('Command requested root, but did not '
'specify a root helper.'))
- cmd = shlex.split(root_helper) + list(cmd)
+ if shell:
+ # root helper has to be injected into the command string
+ cmd = [' '.join((root_helper, cmd[0]))] + list(cmd[1:])
+ else:
+ # root helper has to be tokenized into argument list
+ cmd = shlex.split(root_helper) + list(cmd)
cmd = [str(c) for c in cmd]
sanitized_cmd = strutils.mask_password(' '.join(cmd))
diff --git a/tests/test_processutils.py b/tests/test_processutils.py
index ca15385..dfc24cc 100644
--- a/tests/test_processutils.py
+++ b/tests/test_processutils.py
@@ -298,6 +298,18 @@ grep foo
self.assertIn('SUPER_UNIQUE_VAR=The answer is 42', out)
+ def test_as_root(self):
+ out, err = processutils.execute('a', 'b', 'c', run_as_root=True,
+ root_helper='echo')
+
+ self.assertIn('a b c', six.text_type(out))
+
+ def test_as_root_via_shell(self):
+ out, err = processutils.execute('a b c', run_as_root=True,
+ root_helper='echo', shell=True)
+
+ self.assertIn('a b c', six.text_type(out))
+
def test_exception_and_masking(self):
tmpfilename = self.create_tempfiles(
[["test_exceptions_and_masking",