@@ -25,7 +25,13 @@
import subprocess
import tempfile
from types import TracebackType
-from typing import List, Optional, Type
+from typing import (
+ Any,
+ Dict,
+ List,
+ Optional,
+ Type,
+)
from . import console_socket, qmp
from .qmp import QMPMessage, SocketAddrT
@@ -515,17 +521,23 @@ def set_qmp_monitor(self, enabled=True):
self._qmp_set = False
self._qmp = None
- def qmp(self, cmd, conv_keys=True, **args):
- """
- Invoke a QMP command and return the response dict
- """
+ @classmethod
+ def _qmp_args(cls, _conv_keys: bool = True, **args: Any) -> Dict[str, Any]:
qmp_args = dict()
for key, value in args.items():
- if conv_keys:
+ if _conv_keys:
qmp_args[key.replace('_', '-')] = value
else:
qmp_args[key] = value
+ return qmp_args
+ def qmp(self, cmd: str,
+ conv_keys: bool = True,
+ **args: Any) -> QMPMessage:
+ """
+ Invoke a QMP command and return the response dict
+ """
+ qmp_args = self._qmp_args(conv_keys, **args)
return self._qmp.cmd(cmd, args=qmp_args)
def command(self, cmd, conv_keys=True, **args):
@@ -534,12 +546,8 @@ def command(self, cmd, conv_keys=True, **args):
On success return the response dict.
On failure raise an exception.
"""
- reply = self.qmp(cmd, conv_keys, **args)
- if reply is None:
- raise qmp.QMPError("Monitor is closed")
- if "error" in reply:
- raise qmp.QMPResponseError(reply)
- return reply["return"]
+ qmp_args = self._qmp_args(conv_keys, **args)
+ return self._qmp.command(cmd, **qmp_args)
def get_qmp_event(self, wait=False):
"""