Proxmox-Port/qemu
1
0
Fork 0
mirror of https://github.com/qemu/qemu.git synced 2025-08-09 01:50:43 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

iotests: Factor out qemu_tool_pipe_and_status()

Browse Source 
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>
Message-Id: <20200924152717.287415-29-kwolf@redhat.com>
Acked-by: Stefan Hajnoczi <stefanha@redhat.com>
Reviewed-by: Max Reitz <mreitz@redhat.com>
Signed-off-by: Kevin Wolf <kwolf@redhat.com>
This commit is contained in:
Kevin Wolf 2020-09-24 17:27:14 +02:00
parent 443127e81b
commit 91efbae938
1 changed files with 27 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
tests/qemu-iotests/iotests.py
View File

@ -88,20 +88,29 @@
luks_default_key_secret_opt = 'key-secret=keysec0' luks_default_key_secret_opt = 'key-secret=keysec0'
def qemu_tool_pipe_and_status(tool: str, args: Sequence[str],
connect_stderr: bool = True) -> Tuple[str, int]:
"""
Run a tool and return both its output and its exit code
"""
stderr = subprocess.STDOUT if connect_stderr else None
subp = subprocess.Popen(args,
stdout=subprocess.PIPE,
stderr=stderr,
universal_newlines=True)
output = subp.communicate()[0]
if subp.returncode < 0:
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]: def qemu_img_pipe_and_status(*args: str) -> Tuple[str, int]:
""" """
Run qemu-img and return both its output and its exit code Run qemu-img and return both its output and its exit code
""" """
subp = subprocess.Popen(qemu_img_args + list(args), full_args = qemu_img_args + list(args)
stdout=subprocess.PIPE, return qemu_tool_pipe_and_status('qemu-img', full_args)
stderr=subprocess.STDOUT,
universal_newlines=True)
output = subp.communicate()[0]
if subp.returncode < 0:
sys.stderr.write('qemu-img received signal %i: %s\n'
% (-subp.returncode,
' '.join(qemu_img_args + list(args))))
return (output, subp.returncode)
def qemu_img(*args: str) -> int: def qemu_img(*args: str) -> int:
'''Run qemu-img and return the exit code''' '''Run qemu-img and return the exit code'''
@ -263,19 +272,13 @@ def qemu_nbd(*args):
'''Run qemu-nbd in daemon mode and return the parent's exit code''' '''Run qemu-nbd in daemon mode and return the parent's exit code'''
return subprocess.call(qemu_nbd_args + ['--fork'] + list(args)) 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 '''Run qemu-nbd in daemon mode and return both the parent's exit code
and its output in case of an error''' and its output in case of an error'''
subp = subprocess.Popen(qemu_nbd_args + ['--fork'] + list(args), full_args = qemu_nbd_args + ['--fork'] + list(args)
stdout=subprocess.PIPE, output, returncode = qemu_tool_pipe_and_status('qemu-nbd', full_args,
universal_newlines=True) connect_stderr=False)
output = subp.communicate()[0] return returncode, output if returncode else ''
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 ''
@contextmanager @contextmanager
def qemu_nbd_popen(*args): def qemu_nbd_popen(*args):
@ -1141,20 +1144,14 @@ def verify_working_luks():
if not working: if not working:
notrun(reason) 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). Run qemu with an option to print something and exit (e.g. a help option).
:return: QEMU's stdout output. :return: QEMU's stdout output.
""" """
args = [qemu_prog] + qemu_opts + list(args) full_args = [qemu_prog] + qemu_opts + list(args)
subp = subprocess.Popen(args, stdout=subprocess.PIPE, output, _ = qemu_tool_pipe_and_status('qemu', full_args)
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)))
return output return output
def supported_formats(read_only=False): def supported_formats(read_only=False):