@@ -97,6 +97,12 @@ def __init__(self, reply: QMPMessage):
self.reply = reply
+class QMPBadPortError(QMPError):
+ """
+ Unable to parse socket address: Port was non-numerical.
+ """
+
+
class QEMUMonitorProtocol:
"""
Provide an API to connect to QEMU via QEMU Monitor Protocol (QMP) and then
@@ -224,6 +230,26 @@ def __exit__(self,
# Implement context manager exit function.
self.close()
+ @classmethod
+ def parse_address(cls, address: str) -> SocketAddrT:
+ """
+ Parse a string into a QMP address.
+
+ Figure out if the argument is in the port:host form.
+ If it's not, it's probably a file path.
+ """
+ components = address.split(':')
+ if len(components) == 2:
+ try:
+ port = int(components[1])
+ except ValueError:
+ msg = f"Bad port: '{components[1]}' in '{address}'."
+ raise QMPBadPortError(msg) from None
+ return (components[0], port)
+
+ # Treat as filepath.
+ return address
+
def connect(self, negotiate: bool = True) -> Optional[QMPMessage]:
"""
Connect to the QMP Monitor and perform capabilities negotiation.
@@ -92,10 +92,6 @@ class QMPShellError(Exception):
pass
-class QMPShellBadPort(QMPShellError):
- pass
-
-
class FuzzyJSON(ast.NodeTransformer):
"""
This extension of ast.NodeTransformer filters literal "true/false/null"
@@ -118,7 +114,7 @@ class FuzzyJSON(ast.NodeTransformer):
# _execute_cmd()). Let's design a better one.
class QMPShell(qmp.QEMUMonitorProtocol):
def __init__(self, address, pretty=False):
- super().__init__(self.__get_address(address))
+ super().__init__(self.parse_address(address))
self._greeting = None
self._completer = None
self._pretty = pretty
@@ -128,22 +124,6 @@ class QMPShell(qmp.QEMUMonitorProtocol):
'.qmp-shell_history')
self._verbose = False
- @classmethod
- def __get_address(cls, arg):
- """
- Figure out if the argument is in the port:host form, if it's not it's
- probably a file path.
- """
- addr = arg.split(':')
- if len(addr) == 2:
- try:
- port = int(addr[1])
- except ValueError as err:
- raise QMPShellBadPort from err
- return addr[0], port
- # socket path
- return arg
-
def _fill_completion(self):
cmds = self.cmd('query-commands')
if 'error' in cmds:
@@ -337,9 +317,6 @@ class QMPShell(qmp.QEMUMonitorProtocol):
return self._execute_cmd(cmdline)
- def set_verbosity(self, verbose):
- self._verbose = verbose
-
class HMPShell(QMPShell):
def __init__(self, address):
@@ -447,7 +424,7 @@ def main():
qemu = HMPShell(args.qmp_server)
else:
qemu = QMPShell(args.qmp_server, args.pretty)
- except QMPShellBadPort:
+ except qmp.QMPBadPortError:
parser.error(f"Bad port number: {args.qmp_server}")
return # pycharm doesn't know error() is noreturn
This takes the place of qmp-shell's __get_address function. Signed-off-by: John Snow <jsnow@redhat.com> --- python/qemu/qmp/__init__.py | 26 ++++++++++++++++++++++++++ scripts/qmp/qmp-shell | 27 ++------------------------- 2 files changed, 28 insertions(+), 25 deletions(-)