OpenE2K
/
qemu-e2k
13
4
Fork 0
Code Issues 4 Pull Requests Projects 2 Releases Activity

Python patches 

This PR finalizes the switch from Luiz's QMP library to mine.
 -----BEGIN PGP SIGNATURE-----
 
 iQIzBAABCAAdFiEE+ber27ys35W+dsvQfe+BBqr8OQ4FAmJhdSEACgkQfe+BBqr8
 OQ43phAAkrqVMU/IJzKKMIYoZtO67gk2u2AG+FNbrQr0FuisnnMSZzvDgnlxQHii
 ingLiIFEUNIfj5QxOiD/glbh/QI6GHY5mh/FYdStc4YALb2MqXYPQhW3UCGxDPlF
 YqJzWk2WbZ20drxCgRzHN/pI5SQY6N+Ev9jyzP/cvCNIFY7xxe0IhApiNjjZt9e2
 ngZ3pX+xjX94YezTQQ1E6lDUSXDUQ4VZWl/VH8nbEeUbOWLfR238/WOqWkv1SHWM
 TtOBeYOLUDjFzplMr4Xbnd9DP/Q3/V8KKT9VHNHcF8eAkOohvxeYJx8AuuohZB4C
 qPQj+gaD0cV63qZNNRyetqtCTG6bd+GDt/s3GhUBxsufz+Y3MTMn/3zHlheiaOwO
 ZIXiEkdgKxPTx5T6Vo0BJoE4/22VhzBRQuTg/i0bWrzgKAyPDOf8uQnm5vvGV8/H
 f7KtXWPoqNVc2wWOh5vJAlsnKFDVW6d+jBbk5jRGofDKvVU31uLLu4eBBHpPgaAs
 9fWd7NgEgqL6ZGYsVSyuwmkhKCLjBtd8K/BGQrpicQUH3J80jagSVnmmmt93KaE3
 HXdZfnE3vxcG45LGdjcu88CHOzUqTEflf6gCGg/ISaP3AlPKPZs2Ck7RPHLK1UeG
 084wYmyuq5C/zXIriBhw75ZGoaJHOdgY31OyMdL1D/Ii+p0h3w0=
 =m2An
 -----END PGP SIGNATURE-----

Merge tag 'python-pull-request' of https://gitlab.com/jsnow/qemu into staging

Python patches

This PR finalizes the switch from Luiz's QMP library to mine.

# -----BEGIN PGP SIGNATURE-----
#
# iQIzBAABCAAdFiEE+ber27ys35W+dsvQfe+BBqr8OQ4FAmJhdSEACgkQfe+BBqr8
# OQ43phAAkrqVMU/IJzKKMIYoZtO67gk2u2AG+FNbrQr0FuisnnMSZzvDgnlxQHii
# ingLiIFEUNIfj5QxOiD/glbh/QI6GHY5mh/FYdStc4YALb2MqXYPQhW3UCGxDPlF
# YqJzWk2WbZ20drxCgRzHN/pI5SQY6N+Ev9jyzP/cvCNIFY7xxe0IhApiNjjZt9e2
# ngZ3pX+xjX94YezTQQ1E6lDUSXDUQ4VZWl/VH8nbEeUbOWLfR238/WOqWkv1SHWM
# TtOBeYOLUDjFzplMr4Xbnd9DP/Q3/V8KKT9VHNHcF8eAkOohvxeYJx8AuuohZB4C
# qPQj+gaD0cV63qZNNRyetqtCTG6bd+GDt/s3GhUBxsufz+Y3MTMn/3zHlheiaOwO
# ZIXiEkdgKxPTx5T6Vo0BJoE4/22VhzBRQuTg/i0bWrzgKAyPDOf8uQnm5vvGV8/H
# f7KtXWPoqNVc2wWOh5vJAlsnKFDVW6d+jBbk5jRGofDKvVU31uLLu4eBBHpPgaAs
# 9fWd7NgEgqL6ZGYsVSyuwmkhKCLjBtd8K/BGQrpicQUH3J80jagSVnmmmt93KaE3
# HXdZfnE3vxcG45LGdjcu88CHOzUqTEflf6gCGg/ISaP3AlPKPZs2Ck7RPHLK1UeG
# 084wYmyuq5C/zXIriBhw75ZGoaJHOdgY31OyMdL1D/Ii+p0h3w0=
# =m2An
# -----END PGP SIGNATURE-----
# gpg: Signature made Thu 21 Apr 2022 08:15:45 AM PDT
# gpg:                using RSA key F9B7ABDBBCACDF95BE76CBD07DEF8106AAFC390E
# gpg: Good signature from "John Snow (John Huston) <jsnow@redhat.com>" [undefined]
# gpg: WARNING: This key is not certified with a trusted signature!
# gpg:          There is no indication that the signature belongs to the owner.
# Primary key fingerprint: FAEB 9711 A12C F475 812F  18F2 88A9 064D 1835 61EB
#      Subkey fingerprint: F9B7 ABDB BCAC DF95 BE76  CBD0 7DEF 8106 AAFC 390E

* tag 'python-pull-request' of https://gitlab.com/jsnow/qemu:
  python/qmp: remove pylint workaround from legacy.py
  python: rename 'aqmp-tui' to 'qmp-tui'
  python: rename qemu.aqmp to qemu.qmp
  python: re-enable pylint duplicate-code warnings
  python: remove the old QMP package
  python/aqmp: copy qmp docstrings to qemu.aqmp.legacy
  python/aqmp: fully separate from qmp.QEMUMonitorProtocol
  python/aqmp: take QMPBadPortError and parse_address from qemu.qmp
  python: temporarily silence pylint duplicate-code warnings
  python/aqmp-tui: relicense as LGPLv2+
  python/qmp-shell: relicense as LGPLv2+
  python/aqmp: relicense as LGPLv2+
  python/aqmp: add explicit GPLv2 license to legacy.py
  iotests: switch to AQMP
  iotests/mirror-top-perms: switch to AQMP
  scripts/bench-block-job: switch to AQMP
  python/machine: permanently switch to AQMP

Signed-off-by: Richard Henderson <richard.henderson@linaro.org>
This commit is contained in:
Richard Henderson 2022-04-21 15:16:52 -07:00
parent 378f973a6c 47430775ed
commit da5006445a
32 changed files with 422 additions and 725 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
python/README.rst
View File

@ -59,7 +59,7 @@ Package installation also normally provides executable console scripts,
so that tools like ``qmp-shell`` are always available via $PATH. To
invoke them without installation, you can invoke e.g.:
``> PYTHONPATH=~/src/qemu/python python3 -m qemu.aqmp.qmp_shell``
``> PYTHONPATH=~/src/qemu/python python3 -m qemu.qmp.qmp_shell``
The mappings between console script name and python module path can be
found in ``setup.cfg``.

59
python/qemu/aqmp/__init__.py
View File

@ -1,59 +0,0 @@
"""
QEMU Monitor Protocol (QMP) development library & tooling.
This package provides a fairly low-level class for communicating
asynchronously with QMP protocol servers, as implemented by QEMU, the
QEMU Guest Agent, and the QEMU Storage Daemon.
`QMPClient` provides the main functionality of this package. All errors
raised by this library derive from `QMPError`, see `aqmp.error` for
additional detail. See `aqmp.events` for an in-depth tutorial on
managing QMP events.
"""
# Copyright (C) 2020, 2021 John Snow for Red Hat, Inc.
#
# Authors:
# John Snow <jsnow@redhat.com>
#
# Based on earlier work by Luiz Capitulino <lcapitulino@redhat.com>.
#
# This work is licensed under the terms of the GNU GPL, version 2. See
# the COPYING file in the top-level directory.
import logging
from .error import QMPError
from .events import EventListener
from .message import Message
from .protocol import (
ConnectError,
Runstate,
SocketAddrT,
StateError,
)
from .qmp_client import ExecInterruptedError, ExecuteError, QMPClient
# Suppress logging unless an application engages it.
logging.getLogger('qemu.aqmp').addHandler(logging.NullHandler())
# The order of these fields impact the Sphinx documentation order.
__all__ = (
# Classes, most to least important
'QMPClient',
'Message',
'EventListener',
'Runstate',
# Exceptions, most generic to most explicit
'QMPError',
'StateError',
'ConnectError',
'ExecuteError',
'ExecInterruptedError',
# Type aliases
'SocketAddrT',
)

177
python/qemu/aqmp/legacy.py
View File

@ -1,177 +0,0 @@
"""
Sync QMP Wrapper
This class pretends to be qemu.qmp.QEMUMonitorProtocol.
"""
import asyncio
from typing import (
Any,
Awaitable,
Dict,
List,
Optional,
TypeVar,
Union,
)
import qemu.qmp
from .error import QMPError
from .protocol import Runstate, SocketAddrT
from .qmp_client import QMPClient
# (Temporarily) Re-export QMPBadPortError
QMPBadPortError = qemu.qmp.QMPBadPortError
#: QMPMessage is an entire QMP message of any kind.
QMPMessage = Dict[str, Any]
#: QMPReturnValue is the 'return' value of a command.
QMPReturnValue = object
#: QMPObject is any object in a QMP message.
QMPObject = Dict[str, object]
# QMPMessage can be outgoing commands or incoming events/returns.
# QMPReturnValue is usually a dict/json object, but due to QAPI's
# 'returns-whitelist', it can actually be anything.
#
# {'return': {}} is a QMPMessage,
# {} is the QMPReturnValue.
# pylint: disable=missing-docstring
class QEMUMonitorProtocol(qemu.qmp.QEMUMonitorProtocol):
def __init__(self, address: SocketAddrT,
server: bool = False,
nickname: Optional[str] = None):
# pylint: disable=super-init-not-called
self._aqmp = QMPClient(nickname)
self._aloop = asyncio.get_event_loop()
self._address = address
self._timeout: Optional[float] = None
if server:
self._sync(self._aqmp.start_server(self._address))
_T = TypeVar('_T')
def _sync(
self, future: Awaitable[_T], timeout: Optional[float] = None
) -> _T:
return self._aloop.run_until_complete(
asyncio.wait_for(future, timeout=timeout)
)
def _get_greeting(self) -> Optional[QMPMessage]:
if self._aqmp.greeting is not None:
# pylint: disable=protected-access
return self._aqmp.greeting._asdict()
return None
# __enter__ and __exit__ need no changes
# parse_address needs no changes
def connect(self, negotiate: bool = True) -> Optional[QMPMessage]:
self._aqmp.await_greeting = negotiate
self._aqmp.negotiate = negotiate
self._sync(
self._aqmp.connect(self._address)
)
return self._get_greeting()
def accept(self, timeout: Optional[float] = 15.0) -> QMPMessage:
self._aqmp.await_greeting = True
self._aqmp.negotiate = True
self._sync(self._aqmp.accept(), timeout)
ret = self._get_greeting()
assert ret is not None
return ret
def cmd_obj(self, qmp_cmd: QMPMessage) -> QMPMessage:
return dict(
self._sync(
# pylint: disable=protected-access
# _raw() isn't a public API, because turning off
# automatic ID assignment is discouraged. For
# compatibility with iotests *only*, do it anyway.
self._aqmp._raw(qmp_cmd, assign_id=False),
self._timeout
)
)
# Default impl of cmd() delegates to cmd_obj
def command(self, cmd: str, **kwds: object) -> QMPReturnValue:
return self._sync(
self._aqmp.execute(cmd, kwds),
self._timeout
)
def pull_event(self,
wait: Union[bool, float] = False) -> Optional[QMPMessage]:
if not wait:
# wait is False/0: "do not wait, do not except."
if self._aqmp.events.empty():
return None
# If wait is 'True', wait forever. If wait is False/0, the events
# queue must not be empty; but it still needs some real amount
# of time to complete.
timeout = None
if wait and isinstance(wait, float):
timeout = wait
return dict(
self._sync(
self._aqmp.events.get(),
timeout
)
)
def get_events(self, wait: Union[bool, float] = False) -> List[QMPMessage]:
events = [dict(x) for x in self._aqmp.events.clear()]
if events:
return events
event = self.pull_event(wait)
return [event] if event is not None else []
def clear_events(self) -> None:
self._aqmp.events.clear()
def close(self) -> None:
self._sync(
self._aqmp.disconnect()
)
def settimeout(self, timeout: Optional[float]) -> None:
self._timeout = timeout
def send_fd_scm(self, fd: int) -> None:
self._aqmp.send_fd_scm(fd)
def __del__(self) -> None:
if self._aqmp.runstate == Runstate.IDLE:
return
if not self._aloop.is_running():
self.close()
else:
# Garbage collection ran while the event loop was running.
# Nothing we can do about it now, but if we don't raise our
# own error, the user will be treated to a lot of traceback
# they might not understand.
raise QMPError(
"QEMUMonitorProtocol.close()"
" was not called before object was garbage collected"
)

0
python/qemu/aqmp/py.typed
View File

18
python/qemu/machine/machine.py
View File

@ -40,21 +40,16 @@ from typing import (
TypeVar,
)
from qemu.qmp import ( # pylint: disable=import-error
from qemu.qmp import SocketAddrT
from qemu.qmp.legacy import (
QEMUMonitorProtocol,
QMPMessage,
QMPReturnValue,
SocketAddrT,
)
from . import console_socket
if os.environ.get('QEMU_PYTHON_LEGACY_QMP'):
from qemu.qmp import QEMUMonitorProtocol
else:
from qemu.aqmp.legacy import QEMUMonitorProtocol
LOG = logging.getLogger(__name__)
@ -743,8 +738,9 @@ class QEMUMachine:
:param timeout: Optional timeout, in seconds.
See QEMUMonitorProtocol.pull_event.
:raise QMPTimeoutError: If timeout was non-zero and no matching events
were found.
:raise asyncio.TimeoutError:
If timeout was non-zero and no matching events were found.
:return: A QMP event matching the filter criteria.
If timeout was 0 and no event matched, None.
"""
@ -767,7 +763,7 @@ class QEMUMachine:
event = self._qmp.pull_event(wait=timeout)
if event is None:
# NB: None is only returned when timeout is false-ish.
# Timeouts raise QMPTimeoutError instead!
# Timeouts raise asyncio.TimeoutError instead!
break
if _match(event):
return event

2
python/qemu/machine/qtest.py
View File

@ -26,7 +26,7 @@ from typing import (
TextIO,
)
from qemu.qmp import SocketAddrT # pylint: disable=import-error
from qemu.qmp import SocketAddrT
from .machine import QEMUMachine

9
python/qemu/qmp/README.rst
View File

@ -1,9 +0,0 @@
qemu.qmp package
================
This package provides a library used for connecting to and communicating
with QMP servers. It is used extensively by iotests, vm tests,
avocado tests, and other utilities in the ./scripts directory. It is
not a fully-fledged SDK and is subject to change at any time.
See the documentation in ``__init__.py`` for more information.

445
python/qemu/qmp/__init__.py
View File

@ -1,422 +1,59 @@
"""
QEMU Monitor Protocol (QMP) development library & tooling.
This package provides a fairly low-level class for communicating to QMP
protocol servers, as implemented by QEMU, the QEMU Guest Agent, and the
QEMU Storage Daemon. This library is not intended for production use.
This package provides a fairly low-level class for communicating
asynchronously with QMP protocol servers, as implemented by QEMU, the
QEMU Guest Agent, and the QEMU Storage Daemon.
`QEMUMonitorProtocol` is the primary class of interest, and all errors
raised derive from `QMPError`.
`QMPClient` provides the main functionality of this package. All errors
raised by this library derive from `QMPError`, see `qmp.error` for
additional detail. See `qmp.events` for an in-depth tutorial on
managing QMP events.
"""
# Copyright (C) 2009, 2010 Red Hat Inc.
# Copyright (C) 2020-2022 John Snow for Red Hat, Inc.
#
# Authors:
# Luiz Capitulino <lcapitulino@redhat.com>
# John Snow <jsnow@redhat.com>
#
# This work is licensed under the terms of the GNU GPL, version 2. See
# the COPYING file in the top-level directory.
# Based on earlier work by Luiz Capitulino <lcapitulino@redhat.com>.
#
# This work is licensed under the terms of the GNU LGPL, version 2 or
# later. See the COPYING file in the top-level directory.
import errno
import json
import logging
import socket
import struct
from types import TracebackType
from typing import (
Any,
Dict,
List,
Optional,
TextIO,
Tuple,
Type,
TypeVar,
Union,
cast,
from .error import QMPError
from .events import EventListener
from .message import Message
from .protocol import (
ConnectError,
Runstate,
SocketAddrT,
StateError,
)
from .qmp_client import ExecInterruptedError, ExecuteError, QMPClient
#: QMPMessage is an entire QMP message of any kind.
QMPMessage = Dict[str, Any]
# Suppress logging unless an application engages it.
logging.getLogger('qemu.qmp').addHandler(logging.NullHandler())
#: QMPReturnValue is the 'return' value of a command.
QMPReturnValue = object
#: QMPObject is any object in a QMP message.
QMPObject = Dict[str, object]
# The order of these fields impact the Sphinx documentation order.
__all__ = (
# Classes, most to least important
'QMPClient',
'Message',
'EventListener',
'Runstate',
# QMPMessage can be outgoing commands or incoming events/returns.
# QMPReturnValue is usually a dict/json object, but due to QAPI's
# 'returns-whitelist', it can actually be anything.
#
# {'return': {}} is a QMPMessage,
# {} is the QMPReturnValue.
# Exceptions, most generic to most explicit
'QMPError',
'StateError',
'ConnectError',
'ExecuteError',
'ExecInterruptedError',
InternetAddrT = Tuple[str, int]
UnixAddrT = str
SocketAddrT = Union[InternetAddrT, UnixAddrT]
class QMPError(Exception):
"""
QMP base exception
"""
class QMPConnectError(QMPError):
"""
QMP connection exception
"""
class QMPCapabilitiesError(QMPError):
"""
QMP negotiate capabilities exception
"""
class QMPTimeoutError(QMPError):
"""
QMP timeout exception
"""
class QMPProtocolError(QMPError):
"""
QMP protocol error; unexpected response
"""
class QMPResponseError(QMPError):
"""
Represents erroneous QMP monitor reply
"""
def __init__(self, reply: QMPMessage):
try:
desc = reply['error']['desc']
except KeyError:
desc = reply
super().__init__(desc)
self.reply = reply
class QMPBadPortError(QMPError):
"""
Unable to parse socket address: Port was non-numerical.
"""
class QEMUMonitorProtocol:
"""
Provide an API to connect to QEMU via QEMU Monitor Protocol (QMP) and then
allow to handle commands and events.
"""
#: Logger object for debugging messages
logger = logging.getLogger('QMP')
def __init__(self, address: SocketAddrT,
server: bool = False,
nickname: Optional[str] = None):
"""
Create a QEMUMonitorProtocol class.
@param address: QEMU address, can be either a unix socket path (string)
or a tuple in the form ( address, port ) for a TCP
connection
@param server: server mode listens on the socket (bool)
@raise OSError on socket connection errors
@note No connection is established, this is done by the connect() or
accept() methods
"""
self.__events: List[QMPMessage] = []
self.__address = address
self.__sock = self.__get_sock()
self.__sockfile: Optional[TextIO] = None
self._nickname = nickname
if self._nickname:
self.logger = logging.getLogger('QMP').getChild(self._nickname)
if server:
self.__sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.__sock.bind(self.__address)
self.__sock.listen(1)
def __get_sock(self) -> socket.socket:
if isinstance(self.__address, tuple):
family = socket.AF_INET
else:
family = socket.AF_UNIX
return socket.socket(family, socket.SOCK_STREAM)
def __negotiate_capabilities(self) -> QMPMessage:
greeting = self.__json_read()
if greeting is None or "QMP" not in greeting:
raise QMPConnectError
# Greeting seems ok, negotiate capabilities
resp = self.cmd('qmp_capabilities')
if resp and "return" in resp:
return greeting
raise QMPCapabilitiesError
def __json_read(self, only_event: bool = False) -> Optional[QMPMessage]:
assert self.__sockfile is not None
while True:
data = self.__sockfile.readline()
if not data:
return None
# By definition, any JSON received from QMP is a QMPMessage,
# and we are asserting only at static analysis time that it
# has a particular shape.
resp: QMPMessage = json.loads(data)
if 'event' in resp:
self.logger.debug("<<< %s", resp)
self.__events.append(resp)
if not only_event:
continue
return resp
def __get_events(self, wait: Union[bool, float] = False) -> None:
"""
Check for new events in the stream and cache them in __events.
@param wait (bool): block until an event is available.
@param wait (float): If wait is a float, treat it as a timeout value.
@raise QMPTimeoutError: If a timeout float is provided and the timeout
period elapses.
@raise QMPConnectError: If wait is True but no events could be
retrieved or if some other error occurred.
"""
# Current timeout and blocking status
current_timeout = self.__sock.gettimeout()
# Check for new events regardless and pull them into the cache:
self.__sock.settimeout(0) # i.e. setblocking(False)
try:
self.__json_read()
except OSError as err:
# EAGAIN: No data available; not critical
if err.errno != errno.EAGAIN:
raise
finally:
self.__sock.settimeout(current_timeout)
# Wait for new events, if needed.
# if wait is 0.0, this means "no wait" and is also implicitly false.
if not self.__events and wait:
if isinstance(wait, float):
self.__sock.settimeout(wait)
try:
ret = self.__json_read(only_event=True)
except socket.timeout as err:
raise QMPTimeoutError("Timeout waiting for event") from err
except Exception as err:
msg = "Error while reading from socket"
raise QMPConnectError(msg) from err
finally:
self.__sock.settimeout(current_timeout)
if ret is None:
raise QMPConnectError("Error while reading from socket")
T = TypeVar('T')
def __enter__(self: T) -> T:
# Implement context manager enter function.
return self
def __exit__(self,
# pylint: disable=duplicate-code
# see https://github.com/PyCQA/pylint/issues/3619
exc_type: Optional[Type[BaseException]],
exc_val: Optional[BaseException],
exc_tb: Optional[TracebackType]) -> None:
# Implement context manager exit function.
self.close()
@classmethod
def parse_address(cls, address: str) -> SocketAddrT:
"""
Parse a string into a QMP address.
Figure out if the argument is in the port:host form.
If it's not, it's probably a file path.
"""
components = address.split(':')
if len(components) == 2:
try:
port = int(components[1])
except ValueError:
msg = f"Bad port: '{components[1]}' in '{address}'."
raise QMPBadPortError(msg) from None
return (components[0], port)
# Treat as filepath.
return address
def connect(self, negotiate: bool = True) -> Optional[QMPMessage]:
"""
Connect to the QMP Monitor and perform capabilities negotiation.
@return QMP greeting dict, or None if negotiate is false
@raise OSError on socket connection errors
@raise QMPConnectError if the greeting is not received
@raise QMPCapabilitiesError if fails to negotiate capabilities
"""
self.__sock.connect(self.__address)
self.__sockfile = self.__sock.makefile(mode='r')
if negotiate:
return self.__negotiate_capabilities()
return None
def accept(self, timeout: Optional[float] = 15.0) -> QMPMessage:
"""
Await connection from QMP Monitor and perform capabilities negotiation.
@param timeout: timeout in seconds (nonnegative float number, or
None). The value passed will set the behavior of the
underneath QMP socket as described in [1].
Default value is set to 15.0.
@return QMP greeting dict
@raise OSError on socket connection errors
@raise QMPConnectError if the greeting is not received
@raise QMPCapabilitiesError if fails to negotiate capabilities
[1]
https://docs.python.org/3/library/socket.html#socket.socket.settimeout
"""
self.__sock.settimeout(timeout)
self.__sock, _ = self.__sock.accept()
self.__sockfile = self.__sock.makefile(mode='r')
return self.__negotiate_capabilities()
def cmd_obj(self, qmp_cmd: QMPMessage) -> QMPMessage:
"""
Send a QMP command to the QMP Monitor.
@param qmp_cmd: QMP command to be sent as a Python dict
@return QMP response as a Python dict
"""
self.logger.debug(">>> %s", qmp_cmd)
self.__sock.sendall(json.dumps(qmp_cmd).encode('utf-8'))
resp = self.__json_read()
if resp is None:
raise QMPConnectError("Unexpected empty reply from server")
self.logger.debug("<<< %s", resp)
return resp
def cmd(self, name: str,
args: Optional[Dict[str, object]] = None,
cmd_id: Optional[object] = None) -> QMPMessage:
"""
Build a QMP command and send it to the QMP Monitor.
@param name: command name (string)
@param args: command arguments (dict)
@param cmd_id: command id (dict, list, string or int)
"""
qmp_cmd: QMPMessage = {'execute': name}
if args:
qmp_cmd['arguments'] = args
if cmd_id:
qmp_cmd['id'] = cmd_id
return self.cmd_obj(qmp_cmd)
def command(self, cmd: str, **kwds: object) -> QMPReturnValue:
"""
Build and send a QMP command to the monitor, report errors if any
"""
ret = self.cmd(cmd, kwds)
if 'error' in ret:
raise QMPResponseError(ret)
if 'return' not in ret:
raise QMPProtocolError(
"'return' key not found in QMP response '{}'".format(str(ret))
)
return cast(QMPReturnValue, ret['return'])
def pull_event(self,
wait: Union[bool, float] = False) -> Optional[QMPMessage]:
"""
Pulls a single event.
@param wait (bool): block until an event is available.
@param wait (float): If wait is a float, treat it as a timeout value.
@raise QMPTimeoutError: If a timeout float is provided and the timeout
period elapses.
@raise QMPConnectError: If wait is True but no events could be
retrieved or if some other error occurred.
@return The first available QMP event, or None.
"""
self.__get_events(wait)
if self.__events:
return self.__events.pop(0)
return None
def get_events(self, wait: bool = False) -> List[QMPMessage]:
"""
Get a list of available QMP events and clear all pending events.
@param wait (bool): block until an event is available.
@param wait (float): If wait is a float, treat it as a timeout value.
@raise QMPTimeoutError: If a timeout float is provided and the timeout
period elapses.
@raise QMPConnectError: If wait is True but no events could be
retrieved or if some other error occurred.
@return The list of available QMP events.
"""
self.__get_events(wait)
events = self.__events
self.__events = []
return events
def clear_events(self) -> None:
"""
Clear current list of pending events.
"""
self.__events = []
def close(self) -> None:
"""
Close the socket and socket file.
"""
if self.__sock:
self.__sock.close()
if self.__sockfile:
self.__sockfile.close()
def settimeout(self, timeout: Optional[float]) -> None:
"""
Set the socket timeout.
@param timeout (float): timeout in seconds (non-zero), or None.
@note This is a wrap around socket.settimeout
@raise ValueError: if timeout was set to 0.
"""
if timeout == 0:
msg = "timeout cannot be 0; this engages non-blocking mode."
msg += " Use 'None' instead to disable timeouts."
raise ValueError(msg)
self.__sock.settimeout(timeout)
def send_fd_scm(self, fd: int) -> None:
"""
Send a file descriptor to the remote via SCM_RIGHTS.
"""
if self.__sock.family != socket.AF_UNIX:
raise RuntimeError("Can't use SCM_RIGHTS on non-AF_UNIX socket.")
self.__sock.sendmsg(
[b' '],
[(socket.SOL_SOCKET, socket.SCM_RIGHTS, struct.pack('@i', fd))]
)
# Type aliases
'SocketAddrT',
)

0
python/qemu/aqmp/error.py → python/qemu/qmp/error.py
View File

2
python/qemu/aqmp/events.py → python/qemu/qmp/events.py
View File

@ -1,5 +1,5 @@
"""
AQMP Events and EventListeners
QMP Events and EventListeners
Asynchronous QMP uses `EventListener` objects to listen for events. An
`EventListener` is a FIFO event queue that can be pre-filtered to listen

315
python/qemu/qmp/legacy.py Normal file
View File

@ -0,0 +1,315 @@
"""
(Legacy) Sync QMP Wrapper
This module provides the `QEMUMonitorProtocol` class, which is a
synchronous wrapper around `QMPClient`.
Its design closely resembles that of the original QEMUMonitorProtocol
class, originally written by Luiz Capitulino. It is provided here for
compatibility with scripts inside the QEMU source tree that expect the
old interface.
"""
#
# Copyright (C) 2009-2022 Red Hat Inc.
#
# Authors:
# Luiz Capitulino <lcapitulino@redhat.com>
# John Snow <jsnow@redhat.com>
#
# This work is licensed under the terms of the GNU GPL, version 2. See
# the COPYING file in the top-level directory.
#
import asyncio
from types import TracebackType
from typing import (
Any,
Awaitable,
Dict,
List,
Optional,
Type,
TypeVar,
Union,
)
from .error import QMPError
from .protocol import Runstate, SocketAddrT
from .qmp_client import QMPClient
#: QMPMessage is an entire QMP message of any kind.
QMPMessage = Dict[str, Any]
#: QMPReturnValue is the 'return' value of a command.
QMPReturnValue = object
#: QMPObject is any object in a QMP message.
QMPObject = Dict[str, object]
# QMPMessage can be outgoing commands or incoming events/returns.
# QMPReturnValue is usually a dict/json object, but due to QAPI's
# 'returns-whitelist', it can actually be anything.
#
# {'return': {}} is a QMPMessage,
# {} is the QMPReturnValue.
class QMPBadPortError(QMPError):
"""
Unable to parse socket address: Port was non-numerical.
"""
class QEMUMonitorProtocol:
"""
Provide an API to connect to QEMU via QEMU Monitor Protocol (QMP)
and then allow to handle commands and events.
:param address: QEMU address, can be either a unix socket path (string)
or a tuple in the form ( address, port ) for a TCP
connection
:param server: Act as the socket server. (See 'accept')
:param nickname: Optional nickname used for logging.
"""
def __init__(self, address: SocketAddrT,
server: bool = False,
nickname: Optional[str] = None):
self._qmp = QMPClient(nickname)
self._aloop = asyncio.get_event_loop()
self._address = address
self._timeout: Optional[float] = None
if server:
self._sync(self._qmp.start_server(self._address))
_T = TypeVar('_T')
def _sync(
self, future: Awaitable[_T], timeout: Optional[float] = None
) -> _T:
return self._aloop.run_until_complete(
asyncio.wait_for(future, timeout=timeout)
)
def _get_greeting(self) -> Optional[QMPMessage]:
if self._qmp.greeting is not None:
# pylint: disable=protected-access
return self._qmp.greeting._asdict()
return None
def __enter__(self: _T) -> _T:
# Implement context manager enter function.
return self
def __exit__(self,
exc_type: Optional[Type[BaseException]],
exc_val: Optional[BaseException],
exc_tb: Optional[TracebackType]) -> None:
# Implement context manager exit function.
self.close()
@classmethod
def parse_address(cls, address: str) -> SocketAddrT:
"""
Parse a string into a QMP address.
Figure out if the argument is in the port:host form.
If it's not, it's probably a file path.
"""
components = address.split(':')
if len(components) == 2:
try:
port = int(components[1])
except ValueError:
msg = f"Bad port: '{components[1]}' in '{address}'."
raise QMPBadPortError(msg) from None
return (components[0], port)
# Treat as filepath.
return address
def connect(self, negotiate: bool = True) -> Optional[QMPMessage]:
"""
Connect to the QMP Monitor and perform capabilities negotiation.
:return: QMP greeting dict, or None if negotiate is false
:raise ConnectError: on connection errors
"""
self._qmp.await_greeting = negotiate
self._qmp.negotiate = negotiate
self._sync(
self._qmp.connect(self._address)
)
return self._get_greeting()
def accept(self, timeout: Optional[float] = 15.0) -> QMPMessage:
"""
Await connection from QMP Monitor and perform capabilities negotiation.
:param timeout:
timeout in seconds (nonnegative float number, or None).
If None, there is no timeout, and this may block forever.
:return: QMP greeting dict
:raise ConnectError: on connection errors
"""
self._qmp.await_greeting = True
self._qmp.negotiate = True
self._sync(self._qmp.accept(), timeout)
ret = self._get_greeting()
assert ret is not None
return ret
def cmd_obj(self, qmp_cmd: QMPMessage) -> QMPMessage:
"""
Send a QMP command to the QMP Monitor.
:param qmp_cmd: QMP command to be sent as a Python dict
:return: QMP response as a Python dict
"""
return dict(
self._sync(
# pylint: disable=protected-access
# _raw() isn't a public API, because turning off
# automatic ID assignment is discouraged. For
# compatibility with iotests *only*, do it anyway.
self._qmp._raw(qmp_cmd, assign_id=False),
self._timeout
)
)
def cmd(self, name: str,
args: Optional[Dict[str, object]] = None,
cmd_id: Optional[object] = None) -> QMPMessage:
"""
Build a QMP command and send it to the QMP Monitor.
:param name: command name (string)
:param args: command arguments (dict)
:param cmd_id: command id (dict, list, string or int)
"""
qmp_cmd: QMPMessage = {'execute': name}
if args:
qmp_cmd['arguments'] = args
if cmd_id:
qmp_cmd['id'] = cmd_id
return self.cmd_obj(qmp_cmd)
def command(self, cmd: str, **kwds: object) -> QMPReturnValue:
"""
Build and send a QMP command to the monitor, report errors if any
"""
return self._sync(
self._qmp.execute(cmd, kwds),
self._timeout
)
def pull_event(self,
wait: Union[bool, float] = False) -> Optional[QMPMessage]:
"""
Pulls a single event.
:param wait:
If False or 0, do not wait. Return None if no events ready.
If True, wait forever until the next event.
Otherwise, wait for the specified number of seconds.
:raise asyncio.TimeoutError:
When a timeout is requested and the timeout period elapses.
:return: The first available QMP event, or None.
"""
if not wait:
# wait is False/0: "do not wait, do not except."
if self._qmp.events.empty():
return None
# If wait is 'True', wait forever. If wait is False/0, the events
# queue must not be empty; but it still needs some real amount
# of time to complete.
timeout = None
if wait and isinstance(wait, float):
timeout = wait
return dict(
self._sync(
self._qmp.events.get(),
timeout
)
)
def get_events(self, wait: Union[bool, float] = False) -> List[QMPMessage]:
"""
Get a list of QMP events and clear all pending events.
:param wait:
If False or 0, do not wait. Return None if no events ready.
If True, wait until we have at least one event.
Otherwise, wait for up to the specified number of seconds for at
least one event.
:raise asyncio.TimeoutError:
When a timeout is requested and the timeout period elapses.
:return: A list of QMP events.
"""
events = [dict(x) for x in self._qmp.events.clear()]
if events:
return events
event = self.pull_event(wait)
return [event] if event is not None else []
def clear_events(self) -> None:
"""Clear current list of pending events."""
self._qmp.events.clear()
def close(self) -> None:
"""Close the connection."""
self._sync(
self._qmp.disconnect()
)
def settimeout(self, timeout: Optional[float]) -> None:
"""
Set the timeout for QMP RPC execution.
This timeout affects the `cmd`, `cmd_obj`, and `command` methods.
The `accept`, `pull_event` and `get_event` methods have their
own configurable timeouts.
:param timeout:
timeout in seconds, or None.
None will wait indefinitely.
"""
self._timeout = timeout
def send_fd_scm(self, fd: int) -> None:
"""
Send a file descriptor to the remote via SCM_RIGHTS.
"""
self._qmp.send_fd_scm(fd)
def __del__(self) -> None:
if self._qmp.runstate == Runstate.IDLE:
return
if not self._aloop.is_running():
self.close()
else:
# Garbage collection ran while the event loop was running.
# Nothing we can do about it now, but if we don't raise our
# own error, the user will be treated to a lot of traceback
# they might not understand.
raise QMPError(
"QEMUMonitorProtocol.close()"
" was not called before object was garbage collected"
)

0
python/qemu/aqmp/message.py → python/qemu/qmp/message.py
View File

0
python/qemu/aqmp/models.py → python/qemu/qmp/models.py
View File

4
python/qemu/aqmp/protocol.py → python/qemu/qmp/protocol.py
View File

@ -196,9 +196,9 @@ class AsyncProtocol(Generic[T]):
:param name:
Name used for logging messages, if any. By default, messages
will log to 'qemu.aqmp.protocol', but each individual connection
will log to 'qemu.qmp.protocol', but each individual connection
can be given its own logger by giving it a name; messages will
then log to 'qemu.aqmp.protocol.${name}'.
then log to 'qemu.qmp.protocol.${name}'.
"""
# pylint: disable=too-many-instance-attributes

16
python/qemu/aqmp/qmp_client.py → python/qemu/qmp/qmp_client.py
View File

@ -192,7 +192,7 @@ class QMPClient(AsyncProtocol[Message], Events):
await self.qmp.runstate_changed.wait()
await self.disconnect()
See `aqmp.events` for more detail on event handling patterns.
See `qmp.events` for more detail on event handling patterns.
"""
#: Logger object used for debugging messages.
logger = logging.getLogger(__name__)
@ -416,7 +416,7 @@ class QMPClient(AsyncProtocol[Message], Events):
@upper_half
def _get_exec_id(self) -> str:
exec_id = f"__aqmp#{self._execute_id:05d}"
exec_id = f"__qmp#{self._execute_id:05d}"
self._execute_id += 1
return exec_id
@ -476,7 +476,7 @@ class QMPClient(AsyncProtocol[Message], Events):
An execution ID will be assigned if assign_id is `True`. It can be
disabled, but this requires that an ID is manually assigned
instead. For manually assigned IDs, you must not use the string
'__aqmp#' anywhere in the ID.
'__qmp#' anywhere in the ID.
:param msg: The QMP `Message` to execute.
:param assign_id: If True, assign a new execution ID.
@ -490,7 +490,7 @@ class QMPClient(AsyncProtocol[Message], Events):
msg['id'] = self._get_exec_id()
elif 'id' in msg:
assert isinstance(msg['id'], str)
assert '__aqmp#' not in msg['id']
assert '__qmp#' not in msg['id']
exec_id = await self._issue(msg)
return await self._reply(exec_id)
@ -512,7 +512,7 @@ class QMPClient(AsyncProtocol[Message], Events):
Assign an arbitrary execution ID to this message. If
`False`, the existing id must either be absent (and no other
such pending execution may omit an ID) or a string. If it is
a string, it must not start with '__aqmp#' and no other such
a string, it must not start with '__qmp#' and no other such
pending execution may currently be using that ID.
:return: Execution reply from the server.
@ -524,7 +524,7 @@ class QMPClient(AsyncProtocol[Message], Events):
When assign_id is `False`, an ID is given, and it is not a string.
:raise ValueError:
When assign_id is `False`, but the ID is not usable;
Either because it starts with '__aqmp#' or it is already in-use.
Either because it starts with '__qmp#' or it is already in-use.
"""
# 1. convert generic Mapping or bytes to a QMP Message
# 2. copy Message objects so that we assign an ID only to the copy.
@ -534,9 +534,9 @@ class QMPClient(AsyncProtocol[Message], Events):
if not assign_id and 'id' in msg:
if not isinstance(exec_id, str):
raise TypeError(f"ID ('{exec_id}') must be a string.")
if exec_id.startswith('__aqmp#'):
if exec_id.startswith('__qmp#'):
raise ValueError(
f"ID ('{exec_id}') must not start with '__aqmp#'."
f"ID ('{exec_id}') must not start with '__qmp#'."
)
if not assign_id and exec_id in self._pending:

11
python/qemu/aqmp/qmp_shell.py → python/qemu/qmp/qmp_shell.py
View File

@ -1,11 +1,12 @@
#
# Copyright (C) 2009, 2010 Red Hat Inc.
# Copyright (C) 2009-2022 Red Hat Inc.
#
# Authors:
# Luiz Capitulino <lcapitulino@redhat.com>
# John Snow <jsnow@redhat.com>
#
# This work is licensed under the terms of the GNU GPL, version 2. See
# the COPYING file in the top-level directory.
# This work is licensed under the terms of the GNU LGPL, version 2 or
# later. See the COPYING file in the top-level directory.
#
"""
@ -97,8 +98,8 @@ from typing import (
Sequence,
)
from qemu.aqmp import ConnectError, QMPError, SocketAddrT
from qemu.aqmp.legacy import (
from qemu.qmp import ConnectError, QMPError, SocketAddrT
from qemu.qmp.legacy import (
QEMUMonitorProtocol,
QMPBadPortError,
QMPMessage,

17
python/qemu/aqmp/aqmp_tui.py → python/qemu/qmp/qmp_tui.py
View File

@ -3,16 +3,16 @@
# Authors:
# Niteesh Babu G S <niteesh.gs@gmail.com>
#
# This work is licensed under the terms of the GNU GPL, version 2 or
# This work is licensed under the terms of the GNU LGPL, version 2 or
# later. See the COPYING file in the top-level directory.
"""
AQMP TUI
QMP TUI
AQMP TUI is an asynchronous interface built on top the of the AQMP library.
QMP TUI is an asynchronous interface built on top the of the QMP library.
It is the successor of QMP-shell and is bought-in as a replacement for it.
Example Usage: aqmp-tui <SOCKET | TCP IP:PORT>
Full Usage: aqmp-tui --help
Example Usage: qmp-tui <SOCKET | TCP IP:PORT>
Full Usage: qmp-tui --help
"""
import argparse
@ -35,9 +35,8 @@ from pygments import token as Token
import urwid
import urwid_readline
from qemu.qmp import QEMUMonitorProtocol, QMPBadPortError
from .error import ProtocolError
from .legacy import QEMUMonitorProtocol, QMPBadPortError
from .message import DeserializationError, Message, UnexpectedTypeError
from .protocol import ConnectError, Runstate
from .qmp_client import ExecInterruptedError, QMPClient
@ -130,7 +129,7 @@ def has_handler_type(logger: logging.Logger,
class App(QMPClient):
"""
Implements the AQMP TUI.
Implements the QMP TUI.
Initializes the widgets and starts the urwid event loop.
@ -613,7 +612,7 @@ def main() -> None:
Driver of the whole script, parses arguments, initialize the TUI and
the logger.
"""
parser = argparse.ArgumentParser(description='AQMP TUI')
parser = argparse.ArgumentParser(description='QMP TUI')
parser.add_argument('qmp_server', help='Address of the QMP server. '
'Format <UNIX socket path | TCP addr:port>')
parser.add_argument('--num-retries', type=int, default=10,

0
python/qemu/aqmp/util.py → python/qemu/qmp/util.py
View File

4
python/qemu/utils/qemu_ga_client.py
View File

@ -50,8 +50,8 @@ from typing import (
Sequence,
)
from qemu.aqmp import ConnectError, SocketAddrT
from qemu.aqmp.legacy import QEMUMonitorProtocol
from qemu.qmp import ConnectError, SocketAddrT
from qemu.qmp.legacy import QEMUMonitorProtocol
# This script has not seen many patches or careful attention in quite

2
python/qemu/utils/qom.py
View File

@ -32,7 +32,7 @@ QOM commands:
import argparse
from qemu.aqmp import ExecuteError
from qemu.qmp import ExecuteError
from .qom_common import QOMCommand

4
python/qemu/utils/qom_common.py
View File

@ -27,8 +27,8 @@ from typing import (
TypeVar,
)
from qemu.aqmp import QMPError
from qemu.aqmp.legacy import QEMUMonitorProtocol
from qemu.qmp import QMPError
from qemu.qmp.legacy import QEMUMonitorProtocol
class ObjectPropertyInfo:

2
python/qemu/utils/qom_fuse.py
View File

@ -48,7 +48,7 @@ from typing import (
import fuse
from fuse import FUSE, FuseOSError, Operations
from qemu.aqmp import ExecuteError
from qemu.qmp import ExecuteError
from .qom_common import QOMCommand

11
python/setup.cfg
View File

@ -27,7 +27,6 @@ packages =
qemu.qmp
qemu.machine
qemu.utils
qemu.aqmp
[options.package_data]
* = py.typed
@ -52,7 +51,7 @@ devel =
fuse =
fusepy >= 2.0.4
# AQMP TUI dependencies
# QMP TUI dependencies
tui =
urwid >= 2.1.2
urwid-readline >= 0.13
@ -67,9 +66,9 @@ console_scripts =
qom-tree = qemu.utils.qom:QOMTree.entry_point
qom-fuse = qemu.utils.qom_fuse:QOMFuse.entry_point [fuse]
qemu-ga-client = qemu.utils.qemu_ga_client:main
qmp-shell = qemu.aqmp.qmp_shell:main
qmp-shell-wrap = qemu.aqmp.qmp_shell:main_wrap
aqmp-tui = qemu.aqmp.aqmp_tui:main [tui]
qmp-shell = qemu.qmp.qmp_shell:main
qmp-shell-wrap = qemu.qmp.qmp_shell:main_wrap
qmp-tui = qemu.qmp.qmp_tui:main [tui]
[flake8]
extend-ignore = E722 # Prefer pylint's bare-except checks to flake8's
@ -85,7 +84,7 @@ namespace_packages = True
# fusepy has no type stubs:
allow_subclassing_any = True
[mypy-qemu.aqmp.aqmp_tui]
[mypy-qemu.qmp.qmp_tui]
# urwid and urwid_readline have no type stubs:
allow_subclassing_any = True

14
python/tests/protocol.py
View File

@ -6,9 +6,9 @@ from tempfile import TemporaryDirectory
import avocado
from qemu.aqmp import ConnectError, Runstate
from qemu.aqmp.protocol import AsyncProtocol, StateError
from qemu.aqmp.util import asyncio_run, create_task
from qemu.qmp import ConnectError, Runstate
from qemu.qmp.protocol import AsyncProtocol, StateError
from qemu.qmp.util import asyncio_run, create_task
class NullProtocol(AsyncProtocol[None]):
@ -183,7 +183,7 @@ class Smoke(avocado.Test):
def testLogger(self):
self.assertEqual(
self.proto.logger.name,
'qemu.aqmp.protocol'
'qemu.qmp.protocol'
)
def testName(self):
@ -196,7 +196,7 @@ class Smoke(avocado.Test):
self.assertEqual(
self.proto.logger.name,
'qemu.aqmp.protocol.Steve'
'qemu.qmp.protocol.Steve'
)
self.assertEqual(
@ -431,7 +431,7 @@ class Accept(Connect):
await self.proto.start_server_and_accept('/dev/null')
async def _hanging_connection(self):
with TemporaryDirectory(suffix='.aqmp') as tmpdir:
with TemporaryDirectory(suffix='.qmp') as tmpdir:
sock = os.path.join(tmpdir, type(self.proto).__name__ + ".sock")
await self.proto.start_server_and_accept(sock)
@ -587,7 +587,7 @@ class SimpleSession(TestBase):
@TestBase.async_test
async def testSmoke(self):
with TemporaryDirectory(suffix='.aqmp') as tmpdir:
with TemporaryDirectory(suffix='.qmp') as tmpdir:
sock = os.path.join(tmpdir, type(self.proto).__name__ + ".sock")
server_task = create_task(self.server.start_server_and_accept(sock))

2
scripts/cpu-x86-uarch-abi.py
View File

@ -6,7 +6,7 @@
# compatibility levels for each CPU model.
#
from qemu.aqmp.legacy import QEMUMonitorProtocol
from qemu.qmp.legacy import QEMUMonitorProtocol
import sys
if len(sys.argv) != 2:

4
scripts/device-crash-test
View File

@ -36,7 +36,7 @@ from itertools import chain
sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'python'))
from qemu.machine import QEMUMachine
from qemu.aqmp import ConnectError
from qemu.qmp import ConnectError
logger = logging.getLogger('device-crash-test')
dbg = logger.debug
@ -517,7 +517,7 @@ def main():
# Async QMP, when in use, is chatty about connection failures.
# This script knowingly generates a ton of connection errors.
# Silence this logger.
logging.getLogger('qemu.aqmp.qmp_client').setLevel(logging.CRITICAL)
logging.getLogger('qemu.qmp.qmp_client').setLevel(logging.CRITICAL)
fatal_failures = []
wl_stats = {}

2
scripts/qmp/qmp-shell
View File

@ -4,7 +4,7 @@ import os
import sys
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'python'))
from qemu.aqmp import qmp_shell
from qemu.qmp import qmp_shell
if __name__ == '__main__':

2
scripts/qmp/qmp-shell-wrap
View File

@ -4,7 +4,7 @@ import os
import sys
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'python'))
from qemu.aqmp import qmp_shell
from qemu.qmp import qmp_shell
if __name__ == '__main__':

4
scripts/render_block_graph.py
View File

@ -25,8 +25,8 @@ import json
from graphviz import Digraph
sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'python'))
from qemu.aqmp import QMPError
from qemu.aqmp.legacy import QEMUMonitorProtocol
from qemu.qmp import QMPError
from qemu.qmp.legacy import QEMUMonitorProtocol
def perm(arr):

5
scripts/simplebench/bench_block_job.py
View File

@ -27,8 +27,7 @@ import json
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'python'))
from qemu.machine import QEMUMachine
from qemu.qmp import QMPConnectError
from qemu.aqmp import ConnectError
from qemu.qmp import ConnectError
def bench_block_job(cmd, cmd_args, qemu_args):
@ -50,7 +49,7 @@ def bench_block_job(cmd, cmd_args, qemu_args):
vm.launch()
except OSError as e:
return {'error': 'popen failed: ' + str(e)}
except (QMPConnectError, ConnectError, socket.timeout):
except (ConnectError, socket.timeout):
return {'error': 'qemu failed: ' + str(vm.get_log())}
try:

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

@ -37,9 +37,8 @@ import unittest
from contextlib import contextmanager
from qemu.aqmp.legacy import QEMUMonitorProtocol
from qemu.machine import qtest
from qemu.qmp import QMPMessage
from qemu.qmp.legacy import QMPMessage, QEMUMonitorProtocol
from qemu.utils import VerboseProcessError
# Use this logger for logging messages directly from the iotests module

11
tests/qemu-iotests/tests/mirror-top-perms
View File

@ -22,7 +22,6 @@
import os
from qemu.machine import machine
from qemu.qmp import QMPConnectError
import iotests
from iotests import change_log_level, qemu_img
@ -98,15 +97,13 @@ class TestMirrorTopPerms(iotests.QMPTestCase):
self.vm_b.add_blockdev(f'file,node-name=drive0,filename={source}')
self.vm_b.add_device('virtio-blk,drive=drive0,share-rw=on')
try:
# Silence AQMP errors temporarily.
# TODO: Remove this and just allow the errors to be logged when
# AQMP fully replaces QMP.
with change_log_level('qemu.aqmp'):
# Silence QMP logging errors temporarily.
with change_log_level('qemu.qmp'):
self.vm_b.launch()
print('ERROR: VM B launched successfully, '
'this should not have happened')
except (QMPConnectError, machine.VMLaunchFailure):
assert 'Is another process using the image' in self.vm_b.get_log()
except machine.VMLaunchFailure as exc:
assert 'Is another process using the image' in exc.output
result = self.vm.qmp('block-job-cancel',
device='mirror')