4 qtest offers the QEMUQtestProtocol and QEMUQTestMachine classes, which
5 offer a connection to QEMU's qtest protocol socket, and a qtest-enabled
6 subclass of QEMUMachine, respectively.
9 # Copyright (C) 2015 Red Hat Inc.
12 # Fam Zheng <famz@redhat.com>
14 # This work is licensed under the terms of the GNU GPL, version 2. See
15 # the COPYING file in the top-level directory.
29 from .machine
import QEMUMachine
30 from .qmp
import SocketAddrT
33 class QEMUQtestProtocol
:
35 QEMUQtestProtocol implements a connection to a qtest socket.
37 :param address: QEMU address, can be either a unix socket path (string)
38 or a tuple in the form ( address, port ) for a TCP
40 :param server: server mode, listens on the socket (bool)
41 :raise socket.error: on socket connection errors
44 No conection is estabalished by __init__(), this is done
45 by the connect() or accept() methods.
47 def __init__(self
, address
: SocketAddrT
,
48 server
: bool = False):
49 self
._address
= address
50 self
._sock
= self
._get
_sock
()
51 self
._sockfile
: Optional
[TextIO
] = None
53 self
._sock
.bind(self
._address
)
56 def _get_sock(self
) -> socket
.socket
:
57 if isinstance(self
._address
, tuple):
58 family
= socket
.AF_INET
60 family
= socket
.AF_UNIX
61 return socket
.socket(family
, socket
.SOCK_STREAM
)
63 def connect(self
) -> None:
65 Connect to the qtest socket.
67 @raise socket.error on socket connection errors
69 self
._sock
.connect(self
._address
)
70 self
._sockfile
= self
._sock
.makefile(mode
='r')
72 def accept(self
) -> None:
74 Await connection from QEMU.
76 @raise socket.error on socket connection errors
78 self
._sock
, _
= self
._sock
.accept()
79 self
._sockfile
= self
._sock
.makefile(mode
='r')
81 def cmd(self
, qtest_cmd
: str) -> str:
83 Send a qtest command on the wire.
85 @param qtest_cmd: qtest command text to be sent
87 assert self
._sockfile
is not None
88 self
._sock
.sendall((qtest_cmd
+ "\n").encode('utf-8'))
89 resp
= self
._sockfile
.readline()
92 def close(self
) -> None:
98 self
._sockfile
.close()
101 def settimeout(self
, timeout
: Optional
[float]) -> None:
102 """Set a timeout, in seconds."""
103 self
._sock
.settimeout(timeout
)
106 class QEMUQtestMachine(QEMUMachine
):
108 A QEMU VM, with a qtest socket available.
113 args
: Sequence
[str] = (),
114 name
: Optional
[str] = None,
115 test_dir
: str = "/var/tmp",
116 socket_scm_helper
: Optional
[str] = None,
117 sock_dir
: Optional
[str] = None):
119 name
= "qemu-%d" % os
.getpid()
122 super().__init
__(binary
, args
, name
=name
, test_dir
=test_dir
,
123 socket_scm_helper
=socket_scm_helper
,
125 self
._qtest
: Optional
[QEMUQtestProtocol
] = None
126 self
._qtest
_path
= os
.path
.join(sock_dir
, name
+ "-qtest.sock")
129 def _base_args(self
) -> List
[str]:
130 args
= super()._base
_args
132 '-qtest', f
"unix:path={self._qtest_path}",
137 def _pre_launch(self
) -> None:
138 super()._pre
_launch
()
139 self
._qtest
= QEMUQtestProtocol(self
._qtest
_path
, server
=True)
141 def _post_launch(self
) -> None:
142 assert self
._qtest
is not None
143 super()._post
_launch
()
146 def _post_shutdown(self
) -> None:
147 super()._post
_shutdown
()
148 self
._remove
_if
_exists
(self
._qtest
_path
)
150 def qtest(self
, cmd
: str) -> str:
152 Send a qtest command to the guest.
154 :param cmd: qtest command to send
155 :return: qtest server response
157 if self
._qtest
is None:
158 raise RuntimeError("qtest socket not available")
159 return self
._qtest
.cmd(cmd
)