@@ -90,21 +90,30 @@ luks_default_secret_object = 'secret,id=keysec0,data=' + \
luks_default_key_secret_opt = 'key-secret=keysec0'
-def qemu_img_pipe_and_status(*args: str) -> Tuple[str, int]:
+def qemu_tool_pipe_and_status(tool: str, args: Sequence[str],
+ connect_stderr: bool = True) -> Tuple[str, int]:
"""
- Run qemu-img and return both its output and its exit code
+ Run a tool and return both its output and its exit code
"""
- subp = subprocess.Popen(qemu_img_args + list(args),
+ stderr = subprocess.STDOUT if connect_stderr else None
+ subp = subprocess.Popen(args,
stdout=subprocess.PIPE,
- stderr=subprocess.STDOUT,
+ stderr=stderr,
universal_newlines=True)
output = subp.communicate()[0]
if subp.returncode < 0:
- sys.stderr.write('qemu-img received signal %i: %s\n'
- % (-subp.returncode,
+ sys.stderr.write('%s received signal %i: %s\n'
+ % (tool, -subp.returncode,
' '.join(qemu_img_args + list(args))))
return (output, subp.returncode)
+def qemu_img_pipe_and_status(*args: str) -> Tuple[str, int]:
+ """
+ Run qemu-img and return both its output and its exit code
+ """
+ full_args = qemu_img_args + list(args)
+ return qemu_tool_pipe_and_status('qemu-img', full_args)
+
def qemu_img(*args: str) -> int:
'''Run qemu-img and return the exit code'''
return qemu_img_pipe_and_status(*args)[1]
@@ -265,19 +274,13 @@ def qemu_nbd(*args):
'''Run qemu-nbd in daemon mode and return the parent's exit code'''
return subprocess.call(qemu_nbd_args + ['--fork'] + list(args))
-def qemu_nbd_early_pipe(*args):
+def qemu_nbd_early_pipe(*args: str) -> Tuple[int, str]:
'''Run qemu-nbd in daemon mode and return both the parent's exit code
and its output in case of an error'''
- subp = subprocess.Popen(qemu_nbd_args + ['--fork'] + list(args),
- stdout=subprocess.PIPE,
- universal_newlines=True)
- output = subp.communicate()[0]
- if subp.returncode < 0:
- sys.stderr.write('qemu-nbd received signal %i: %s\n' %
- (-subp.returncode,
- ' '.join(qemu_nbd_args + ['--fork'] + list(args))))
-
- return subp.returncode, output if subp.returncode else ''
+ full_args = qemu_nbd_args + ['--fork'] + list(args)
+ output, returncode = qemu_tool_pipe_and_status('qemu-nbd', full_args,
+ connect_stderr=False)
+ return returncode, output if returncode else ''
@contextmanager
def qemu_nbd_popen(*args):
@@ -1143,20 +1146,14 @@ def verify_working_luks():
if not working:
notrun(reason)
-def qemu_pipe(*args):
+def qemu_pipe(*args: str) -> str:
"""
Run qemu with an option to print something and exit (e.g. a help option).
:return: QEMU's stdout output.
"""
- args = [qemu_prog] + qemu_opts + list(args)
- subp = subprocess.Popen(args, stdout=subprocess.PIPE,
- stderr=subprocess.STDOUT,
- universal_newlines=True)
- output = subp.communicate()[0]
- if subp.returncode < 0:
- sys.stderr.write('qemu received signal %i: %s\n' %
- (-subp.returncode, ' '.join(args)))
+ full_args = [qemu_prog] + qemu_opts + list(args)
+ output, _ = qemu_tool_pipe_and_status('qemu', full_args)
return output
def supported_formats(read_only=False):
We have three almost identical functions that call an external process and return its output and return code. Refactor them into small wrappers around a common function. Signed-off-by: Kevin Wolf <kwolf@redhat.com> --- tests/qemu-iotests/iotests.py | 49 ++++++++++++++++------------------- 1 file changed, 23 insertions(+), 26 deletions(-)