mirrors/qemu
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

iotests: add hmp helper with logging

Browse Source 
Minor cleanup for HMP functions; helps with line length and consolidates
HMP helpers through one implementation function.

Although we are adding a universal toggle to turn QMP logging on or off,
many existing callers to hmp functions don't expect that output to be
logged, which causes quite a few changes in the test output.

For now, offer a use_log parameter.

Typing notes:

QMPResponse is just an alias for Dict[str, Any]. It holds no special
meanings and it is not a formal subtype of Dict[str, Any]. It is best
thought of as a lexical synonym.

We may well wish to add stricter subtypes in the future for certain
shapes of data that are not formalized as Python objects, at which point
we can simply retire the alias and allow mypy to more strictly check
usages of the name.

Signed-off-by: John Snow <jsnow@redhat.com>
Message-Id: <20200331000014.11581-11-jsnow@redhat.com>
Reviewed-by: Max Reitz <mreitz@redhat.com>
Reviewed-by: Kevin Wolf <kwolf@redhat.com>
Signed-off-by: Max Reitz <mreitz@redhat.com>
This commit is contained in:
John Snow 2020-03-30 20:00:10 -04:00 committed by Max Reitz
parent b031e9a5a6
commit 239bbcc0ae
1 changed files with 22 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -37,6 +37,10 @@ from qemu import qtest
assert sys.version_info >= (3, 6) assert sys.version_info >= (3, 6)
# Type Aliases
QMPResponse = Dict[str, Any]
faulthandler.enable() faulthandler.enable()
# This will not work if arguments contain spaces but is necessary if we # This will not work if arguments contain spaces but is necessary if we
@ -541,25 +545,30 @@ class VM(qtest.QEMUQtestMachine):
self._args.append(addr) self._args.append(addr)
return self return self
def pause_drive(self, drive, event=None): def hmp(self, command_line: str, use_log: bool = False) -> QMPResponse:
'''Pause drive r/w operations''' cmd = 'human-monitor-command'
kwargs = {'command-line': command_line}
if use_log:
return self.qmp_log(cmd, **kwargs)
else:
return self.qmp(cmd, **kwargs)
def pause_drive(self, drive: str, event: Optional[str] = None) -> None:
"""Pause drive r/w operations"""
if not event: if not event:
self.pause_drive(drive, "read_aio") self.pause_drive(drive, "read_aio")
self.pause_drive(drive, "write_aio") self.pause_drive(drive, "write_aio")
return return
self.qmp('human-monitor-command', self.hmp(f'qemu-io {drive} "break {event} bp_{drive}"')
command_line='qemu-io %s "break %s bp_%s"'
% (drive, event, drive))
def resume_drive(self, drive): def resume_drive(self, drive: str) -> None:
self.qmp('human-monitor-command', """Resume drive r/w operations"""
command_line='qemu-io %s "remove_break bp_%s"' self.hmp(f'qemu-io {drive} "remove_break bp_{drive}"')
% (drive, drive))
def hmp_qemu_io(self, drive, cmd): def hmp_qemu_io(self, drive: str, cmd: str,
'''Write to a given drive using an HMP command''' use_log: bool = False) -> QMPResponse:
return self.qmp('human-monitor-command', """Write to a given drive using an HMP command"""
command_line='qemu-io %s "%s"' % (drive, cmd)) return self.hmp(f'qemu-io {drive} "{cmd}"', use_log=use_log)
def flatten_qmp_object(self, obj, output=None, basestr=''): def flatten_qmp_object(self, obj, output=None, basestr=''):
if output is None: if output is None: