diff options
Diffstat (limited to 'python/qemu/qtest.py')
| -rw-r--r-- | python/qemu/qtest.py | 55 |
1 files changed, 36 insertions, 19 deletions
diff --git a/python/qemu/qtest.py b/python/qemu/qtest.py index 888c8bd2f6..39a0cf62fe 100644 --- a/python/qemu/qtest.py +++ b/python/qemu/qtest.py @@ -17,11 +17,17 @@ subclass of QEMUMachine, respectively. # Based on qmp.py. # -import socket import os -from typing import Optional, TextIO +import socket +from typing import ( + List, + Optional, + Sequence, + TextIO, +) from .machine import QEMUMachine +from .qmp import SocketAddrT class QEMUQtestProtocol: @@ -38,7 +44,8 @@ class QEMUQtestProtocol: No conection is estabalished by __init__(), this is done by the connect() or accept() methods. """ - def __init__(self, address, server=False): + def __init__(self, address: SocketAddrT, + server: bool = False): self._address = address self._sock = self._get_sock() self._sockfile: Optional[TextIO] = None @@ -46,14 +53,14 @@ class QEMUQtestProtocol: self._sock.bind(self._address) self._sock.listen(1) - def _get_sock(self): + def _get_sock(self) -> socket.socket: if isinstance(self._address, tuple): family = socket.AF_INET else: family = socket.AF_UNIX return socket.socket(family, socket.SOCK_STREAM) - def connect(self): + def connect(self) -> None: """ Connect to the qtest socket. @@ -62,7 +69,7 @@ class QEMUQtestProtocol: self._sock.connect(self._address) self._sockfile = self._sock.makefile(mode='r') - def accept(self): + def accept(self) -> None: """ Await connection from QEMU. @@ -71,7 +78,7 @@ class QEMUQtestProtocol: self._sock, _ = self._sock.accept() self._sockfile = self._sock.makefile(mode='r') - def cmd(self, qtest_cmd): + def cmd(self, qtest_cmd: str) -> str: """ Send a qtest command on the wire. @@ -82,14 +89,16 @@ class QEMUQtestProtocol: resp = self._sockfile.readline() return resp - def close(self): - """Close this socket.""" + def close(self) -> None: + """ + Close this socket. + """ self._sock.close() if self._sockfile: self._sockfile.close() self._sockfile = None - def settimeout(self, timeout): + def settimeout(self, timeout: Optional[float]) -> None: """Set a timeout, in seconds.""" self._sock.settimeout(timeout) @@ -99,8 +108,13 @@ class QEMUQtestMachine(QEMUMachine): A QEMU VM, with a qtest socket available. """ - def __init__(self, binary, args=None, name=None, test_dir="/var/tmp", - socket_scm_helper=None, sock_dir=None): + def __init__(self, + binary: str, + args: Sequence[str] = (), + name: Optional[str] = None, + test_dir: str = "/var/tmp", + socket_scm_helper: Optional[str] = None, + sock_dir: Optional[str] = None): if name is None: name = "qemu-%d" % os.getpid() if sock_dir is None: @@ -108,16 +122,19 @@ class QEMUQtestMachine(QEMUMachine): super().__init__(binary, args, name=name, test_dir=test_dir, socket_scm_helper=socket_scm_helper, sock_dir=sock_dir) - self._qtest = None + self._qtest: Optional[QEMUQtestProtocol] = None self._qtest_path = os.path.join(sock_dir, name + "-qtest.sock") - def _base_args(self): - args = super()._base_args() - args.extend(['-qtest', 'unix:path=' + self._qtest_path, - '-accel', 'qtest']) + @property + def _base_args(self) -> List[str]: + args = super()._base_args + args.extend([ + '-qtest', f"unix:path={self._qtest_path}", + '-accel', 'qtest' + ]) return args - def _pre_launch(self): + def _pre_launch(self) -> None: super()._pre_launch() self._qtest = QEMUQtestProtocol(self._qtest_path, server=True) @@ -126,7 +143,7 @@ class QEMUQtestMachine(QEMUMachine): super()._post_launch() self._qtest.accept() - def _post_shutdown(self): + def _post_shutdown(self) -> None: super()._post_shutdown() self._remove_if_exists(self._qtest_path) |