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
> ---
> 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
--
Beraldo
