@@ -24,6 +24,7 @@
Optional,
Sequence,
TextIO,
+ Tuple,
)
from qemu.qmp import SocketAddrT
@@ -38,23 +39,41 @@ class QEMUQtestProtocol:
:param address: QEMU address, can be either a unix socket path (string)
or a tuple in the form ( address, port ) for a TCP
connection
- :param server: server mode, listens on the socket (bool)
+ :param sock: An existing socket can be provided as an alternative to
+ an address. One of address or sock must be provided.
+ :param server: server mode, listens on the socket. Only meaningful
+ in conjunction with an address and not an existing
+ socket.
+
:raise socket.error: on socket connection errors
.. note::
No connection is established by __init__(), this is done
by the connect() or accept() methods.
"""
- def __init__(self, address: SocketAddrT,
+ def __init__(self,
+ address: Optional[SocketAddrT] = None,
+ sock: Optional[socket.socket] = None,
server: bool = False):
+ if address is None and sock is None:
+ raise ValueError("Either 'address' or 'sock' must be specified")
+ if address is not None and sock is not None:
+ raise ValueError(
+ "Either 'address' or 'sock' must be specified, but not both")
+ if sock is not None and server:
+ raise ValueError("server=True is meaningless when passing socket")
+
self._address = address
- self._sock = self._get_sock()
+ self._sock = sock or self._get_sock()
self._sockfile: Optional[TextIO] = None
+
if server:
+ assert self._address is not None
self._sock.bind(self._address)
self._sock.listen(1)
def _get_sock(self) -> socket.socket:
+ assert self._address is not None
if isinstance(self._address, tuple):
family = socket.AF_INET
else:
@@ -67,7 +86,8 @@ def connect(self) -> None:
@raise socket.error on socket connection errors
"""
- self._sock.connect(self._address)
+ if self._address is not None:
+ self._sock.connect(self._address)
self._sockfile = self._sock.makefile(mode='r')
def accept(self) -> None:
@@ -127,29 +147,40 @@ def __init__(self,
base_temp_dir=base_temp_dir,
sock_dir=sock_dir, qmp_timer=qmp_timer)
self._qtest: Optional[QEMUQtestProtocol] = None
- self._qtest_path = os.path.join(sock_dir, name + "-qtest.sock")
+ self._qtest_sock_pair: Optional[
+ Tuple[socket.socket, socket.socket]] = None
@property
def _base_args(self) -> List[str]:
args = super()._base_args
+ assert self._qtest_sock_pair is not None
+ fd = self._qtest_sock_pair[0].fileno()
args.extend([
- '-qtest', f"unix:path={self._qtest_path}",
+ '-chardev', f"socket,id=qtest,fd={fd}",
+ '-qtest', 'chardev:qtest',
'-accel', 'qtest'
])
return args
def _pre_launch(self) -> None:
+ self._qtest_sock_pair = socket.socketpair()
+ os.set_inheritable(self._qtest_sock_pair[0].fileno(), True)
super()._pre_launch()
- self._qtest = QEMUQtestProtocol(self._qtest_path, server=True)
+ self._qtest = QEMUQtestProtocol(sock=self._qtest_sock_pair[1])
def _post_launch(self) -> None:
assert self._qtest is not None
super()._post_launch()
- self._qtest.accept()
+ if self._qtest_sock_pair:
+ self._qtest_sock_pair[0].close()
+ self._qtest.connect()
def _post_shutdown(self) -> None:
+ if self._qtest_sock_pair:
+ self._qtest_sock_pair[0].close()
+ self._qtest_sock_pair[1].close()
+ self._qtest_sock_pair = None
super()._post_shutdown()
- self._remove_if_exists(self._qtest_path)
def qtest(self, cmd: str) -> str:
"""