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

python/aqmp: copy qmp docstrings to qemu.aqmp.legacy

Browse Source 
Copy the docstrings out of qemu.qmp, adjusting them as necessary to
more accurately reflect the current state of this class.

(Licensing: This is copying and modifying GPLv2-only licensed docstrings
into a GPLv2-only file.)

Signed-off-by: John Snow <jsnow@redhat.com>
Reviewed-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
Reviewed-by: Beraldo Leal <bleal@redhat.com>
Message-id: 20220330172812.3427355-5-jsnow@redhat.com
Signed-off-by: John Snow <jsnow@redhat.com>
This commit is contained in:
John Snow 2022-03-30 13:28:07 -04:00
parent 0c78ebf722
commit b0654f4f98
1 changed files with 90 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,7 +1,13 @@
"""
Sync QMP Wrapper
(Legacy) Sync QMP Wrapper
This class pretends to be qemu.qmp.QEMUMonitorProtocol.
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.
"""
#
@ -50,9 +56,6 @@ QMPObject = Dict[str, object]
# {} is the QMPReturnValue.
# pylint: disable=missing-docstring
class QMPBadPortError(QMPError):
"""
Unable to parse socket address: Port was non-numerical.
@ -60,6 +63,17 @@ class QMPBadPortError(QMPError):
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):
@ -121,6 +135,12 @@ class QEMUMonitorProtocol:
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._aqmp.await_greeting = negotiate
self._aqmp.negotiate = negotiate
@ -130,6 +150,16 @@ class QEMUMonitorProtocol:
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._aqmp.await_greeting = True
self._aqmp.negotiate = True
@ -140,6 +170,12 @@ class QEMUMonitorProtocol:
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
@ -158,9 +194,9 @@ class QEMUMonitorProtocol:
"""
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)
: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:
@ -170,6 +206,9 @@ class QEMUMonitorProtocol:
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._aqmp.execute(cmd, kwds),
self._timeout
@ -177,6 +216,19 @@ class QEMUMonitorProtocol:
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._aqmp.events.empty():
@ -197,6 +249,20 @@ class QEMUMonitorProtocol:
)
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._aqmp.events.clear()]
if events:
return events
@ -205,17 +271,33 @@ class QEMUMonitorProtocol:
return [event] if event is not None else []
def clear_events(self) -> None:
"""Clear current list of pending events."""
self._aqmp.events.clear()
def close(self) -> None:
"""Close the connection."""
self._sync(
self._aqmp.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._aqmp.send_fd_scm(fd)
def __del__(self) -> None: