python/aqmp: take QMPBadPortError and parse_address from qemu.qmp

Shift these definitions over from the qmp package to the async qmp
package.

(Licensing: this is a lateral move, from GPLv2 (only) to GPLv2 (only))

Signed-off-by: John Snow <jsnow@redhat.com>
Reviewed-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
Reviewed-by: Beraldo Leal <bleal@redhat.com>
Message-id: 20220330172812.3427355-3-jsnow@redhat.com
Signed-off-by: John Snow <jsnow@redhat.com>

1 files changed, 26 insertions, 4 deletions

diff --git a/python/qemu/aqmp/legacy.py b/python/qemu/aqmp/legacy.py
index f86cb29804..f026274949 100644
--- a/python/qemu/aqmp/legacy.py
+++ b/python/qemu/aqmp/legacy.py
@@ -33,9 +33,6 @@ from .protocol import Runstate, SocketAddrT
 from .qmp_client import QMPClient
 
 
-# (Temporarily) Re-export QMPBadPortError
-QMPBadPortError = qemu.qmp.QMPBadPortError
-
 #: QMPMessage is an entire QMP message of any kind.
 QMPMessage = Dict[str, Any]
 
@@ -56,6 +53,12 @@ QMPObject = Dict[str, object]
 # pylint: disable=missing-docstring
 
 
+class QMPBadPortError(QMPError):
+    """
+    Unable to parse socket address: Port was non-numerical.
+    """
+
+
 class QEMUMonitorProtocol(qemu.qmp.QEMUMonitorProtocol):
     def __init__(self, address: SocketAddrT,
                  server: bool = False,
@@ -86,7 +89,26 @@ class QEMUMonitorProtocol(qemu.qmp.QEMUMonitorProtocol):
         return None
 
     # __enter__ and __exit__ need no changes
-    # parse_address needs no changes
+
+    @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]:
         self._aqmp.await_greeting = negotiate