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.
22 from typing
import Optional
, TextIO
24 from .machine
import QEMUMachine
27 class QEMUQtestProtocol
:
29 QEMUQtestProtocol implements a connection to a qtest socket.
31 :param address: QEMU address, can be either a unix socket path (string)
32 or a tuple in the form ( address, port ) for a TCP
34 :param server: server mode, listens on the socket (bool)
35 :raise socket.error: on socket connection errors
38 No conection is estabalished by __init__(), this is done
39 by the connect() or accept() methods.
41 def __init__(self
, address
, server
=False):
42 self
._address
= address
43 self
._sock
= self
._get
_sock
()
44 self
._sockfile
: Optional
[TextIO
] = None
46 self
._sock
.bind(self
._address
)
50 if isinstance(self
._address
, tuple):
51 family
= socket
.AF_INET
53 family
= socket
.AF_UNIX
54 return socket
.socket(family
, socket
.SOCK_STREAM
)
58 Connect to the qtest socket.
60 @raise socket.error on socket connection errors
62 self
._sock
.connect(self
._address
)
63 self
._sockfile
= self
._sock
.makefile(mode
='r')
67 Await connection from QEMU.
69 @raise socket.error on socket connection errors
71 self
._sock
, _
= self
._sock
.accept()
72 self
._sockfile
= self
._sock
.makefile(mode
='r')
74 def cmd(self
, qtest_cmd
):
76 Send a qtest command on the wire.
78 @param qtest_cmd: qtest command text to be sent
80 assert self
._sockfile
is not None
81 self
._sock
.sendall((qtest_cmd
+ "\n").encode('utf-8'))
82 resp
= self
._sockfile
.readline()
86 """Close this socket."""
89 self
._sockfile
.close()
92 def settimeout(self
, timeout
):
93 """Set a timeout, in seconds."""
94 self
._sock
.settimeout(timeout
)
97 class QEMUQtestMachine(QEMUMachine
):
99 A QEMU VM, with a qtest socket available.
102 def __init__(self
, binary
, args
=None, name
=None, test_dir
="/var/tmp",
103 socket_scm_helper
=None, sock_dir
=None):
105 name
= "qemu-%d" % os
.getpid()
108 super().__init
__(binary
, args
, name
=name
, test_dir
=test_dir
,
109 socket_scm_helper
=socket_scm_helper
,
112 self
._qtest
_path
= os
.path
.join(sock_dir
, name
+ "-qtest.sock")
114 def _base_args(self
):
115 args
= super()._base
_args
()
116 args
.extend(['-qtest', 'unix:path=' + self
._qtest
_path
,
120 def _pre_launch(self
):
121 super()._pre
_launch
()
122 self
._qtest
= QEMUQtestProtocol(self
._qtest
_path
, server
=True)
124 def _post_launch(self
) -> None:
125 assert self
._qtest
is not None
126 super()._post
_launch
()
129 def _post_shutdown(self
):
130 super()._post
_shutdown
()
131 self
._remove
_if
_exists
(self
._qtest
_path
)
133 def qtest(self
, cmd
: str) -> str:
135 Send a qtest command to the guest.
137 :param cmd: qtest command to send
138 :return: qtest server response
140 if self
._qtest
is None:
141 raise RuntimeError("qtest socket not available")
142 return self
._qtest
.cmd(cmd
)