]>
git.proxmox.com Git - mirror_qemu.git/blob - python/qemu/machine/qtest.py
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.
30 from qemu
.qmp
import SocketAddrT
32 from .machine
import QEMUMachine
35 class QEMUQtestProtocol
:
37 QEMUQtestProtocol implements a connection to a qtest socket.
39 :param address: QEMU address, can be either a unix socket path (string)
40 or a tuple in the form ( address, port ) for a TCP
42 :param sock: An existing socket can be provided as an alternative to
43 an address. One of address or sock must be provided.
44 :param server: server mode, listens on the socket. Only meaningful
45 in conjunction with an address and not an existing
48 :raise socket.error: on socket connection errors
51 No connection is established by __init__(), this is done
52 by the connect() or accept() methods.
55 address
: Optional
[SocketAddrT
] = None,
56 sock
: Optional
[socket
.socket
] = None,
57 server
: bool = False):
58 if address
is None and sock
is None:
59 raise ValueError("Either 'address' or 'sock' must be specified")
60 if address
is not None and sock
is not None:
62 "Either 'address' or 'sock' must be specified, but not both")
63 if sock
is not None and server
:
64 raise ValueError("server=True is meaningless when passing socket")
66 self
._address
= address
67 self
._sock
= sock
or self
._get
_sock
()
68 self
._sockfile
: Optional
[TextIO
] = None
71 assert self
._address
is not None
72 self
._sock
.bind(self
._address
)
75 def _get_sock(self
) -> socket
.socket
:
76 assert self
._address
is not None
77 if isinstance(self
._address
, tuple):
78 family
= socket
.AF_INET
80 family
= socket
.AF_UNIX
81 return socket
.socket(family
, socket
.SOCK_STREAM
)
83 def connect(self
) -> None:
85 Connect to the qtest socket.
87 @raise socket.error on socket connection errors
89 if self
._address
is not None:
90 self
._sock
.connect(self
._address
)
91 self
._sockfile
= self
._sock
.makefile(mode
='r')
93 def accept(self
) -> None:
95 Await connection from QEMU.
97 @raise socket.error on socket connection errors
99 self
._sock
, _
= self
._sock
.accept()
100 self
._sockfile
= self
._sock
.makefile(mode
='r')
102 def cmd(self
, qtest_cmd
: str) -> str:
104 Send a qtest command on the wire.
106 @param qtest_cmd: qtest command text to be sent
108 assert self
._sockfile
is not None
109 self
._sock
.sendall((qtest_cmd
+ "\n").encode('utf-8'))
110 resp
= self
._sockfile
.readline()
113 def close(self
) -> None:
119 self
._sockfile
.close()
120 self
._sockfile
= None
122 def settimeout(self
, timeout
: Optional
[float]) -> None:
123 """Set a timeout, in seconds."""
124 self
._sock
.settimeout(timeout
)
127 class QEMUQtestMachine(QEMUMachine
):
129 A QEMU VM, with a qtest socket available.
134 args
: Sequence
[str] = (),
135 wrapper
: Sequence
[str] = (),
136 name
: Optional
[str] = None,
137 base_temp_dir
: str = "/var/tmp",
138 sock_dir
: Optional
[str] = None,
139 qmp_timer
: Optional
[float] = None):
140 # pylint: disable=too-many-arguments
143 name
= "qemu-%d" % os
.getpid()
145 sock_dir
= base_temp_dir
146 super().__init
__(binary
, args
, wrapper
=wrapper
, name
=name
,
147 base_temp_dir
=base_temp_dir
,
148 sock_dir
=sock_dir
, qmp_timer
=qmp_timer
)
149 self
._qtest
: Optional
[QEMUQtestProtocol
] = None
150 self
._qtest
_sock
_pair
: Optional
[
151 Tuple
[socket
.socket
, socket
.socket
]] = None
154 def _base_args(self
) -> List
[str]:
155 args
= super()._base
_args
156 assert self
._qtest
_sock
_pair
is not None
157 fd
= self
._qtest
_sock
_pair
[0].fileno()
159 '-chardev', f
"socket,id=qtest,fd={fd}",
160 '-qtest', 'chardev:qtest',
165 def _pre_launch(self
) -> None:
166 self
._qtest
_sock
_pair
= socket
.socketpair()
167 os
.set_inheritable(self
._qtest
_sock
_pair
[0].fileno(), True)
168 super()._pre
_launch
()
169 self
._qtest
= QEMUQtestProtocol(sock
=self
._qtest
_sock
_pair
[1])
171 def _post_launch(self
) -> None:
172 assert self
._qtest
is not None
173 super()._post
_launch
()
174 if self
._qtest
_sock
_pair
:
175 self
._qtest
_sock
_pair
[0].close()
176 self
._qtest
.connect()
178 def _post_shutdown(self
) -> None:
179 if self
._qtest
_sock
_pair
:
180 self
._qtest
_sock
_pair
[0].close()
181 self
._qtest
_sock
_pair
[1].close()
182 self
._qtest
_sock
_pair
= None
183 super()._post
_shutdown
()
185 def qtest(self
, cmd
: str) -> str:
187 Send a qtest command to the guest.
189 :param cmd: qtest command to send
190 :return: qtest server response
192 if self
._qtest
is None:
193 raise RuntimeError("qtest socket not available")
194 return self
._qtest
.cmd(cmd
)