Re: [PATCH v2 21/25] python/aqmp: fully separate from qmp.QEMUMonitorProtocol

[Beraldo Leal <bleal@redhat.com>]

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 <jsnow@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 <bleal@redhat.com>

--
Beraldo