On Wed, Dec 15, 2021 at 02:39:35PM -0500, John Snow wrote: > After this patch, qemu.aqmp.legacy.QEMUMonitorProtocol no longer > inherits from qemu.qmp.QEMUMonitorProtocol. To do this, several > inherited methods need to be explicitly re-defined. > > Signed-off-by: John Snow <js...@redhat.com> > --- > python/qemu/aqmp/legacy.py | 38 ++++++++++++++++++++++++++++++++------ > 1 file changed, 32 insertions(+), 6 deletions(-) > > diff --git a/python/qemu/aqmp/legacy.py b/python/qemu/aqmp/legacy.py > index 76b09671cc..8f38e7d912 100644 > --- a/python/qemu/aqmp/legacy.py > +++ b/python/qemu/aqmp/legacy.py > @@ -5,18 +5,18 @@ > """ > > import asyncio > +from types import TracebackType > from typing import ( > Any, > Awaitable, > Dict, > List, > Optional, > + Type, > TypeVar, > Union, > ) > > -import qemu.qmp > - > from .error import QMPError > from .protocol import Runstate, SocketAddrT > from .qmp_client import QMPClient > @@ -48,9 +48,9 @@ class QMPBadPortError(QMPError): > """ > > > -class QEMUMonitorProtocol(qemu.qmp.QEMUMonitorProtocol): > +class QEMUMonitorProtocol: > def __init__(self, address: SocketAddrT, > - server: bool = False, > + server: bool = False, # pylint: disable=unused-argument > nickname: Optional[str] = None): > > # pylint: disable=super-init-not-called > @@ -74,7 +74,18 @@ def _get_greeting(self) -> Optional[QMPMessage]: > return self._aqmp.greeting._asdict() > return None > > - # __enter__ and __exit__ need no changes > + 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: > @@ -131,7 +142,22 @@ def cmd_obj(self, qmp_cmd: QMPMessage) -> QMPMessage: > ) > ) > > - # Default impl of cmd() delegates to cmd_obj > + 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: > return self._sync(
Reviewed-by: Beraldo Leal <bl...@redhat.com> -- Beraldo