3 # Copyright (C) 2015 Red Hat Inc.
6 # Fam Zheng <famz@redhat.com>
8 # This work is licensed under the terms of the GNU GPL, version 2. See
9 # the COPYING file in the top-level directory.
17 from .machine
import QEMUMachine
20 class QEMUQtestProtocol(object):
21 def __init__(self
, address
, server
=False):
23 Create a QEMUQtestProtocol object.
25 @param address: QEMU address, can be either a unix socket path (string)
26 or a tuple in the form ( address, port ) for a TCP
28 @param server: server mode, listens on the socket (bool)
29 @raise socket.error on socket connection errors
30 @note No connection is established, this is done by the connect() or
33 self
._address
= address
34 self
._sock
= self
._get
_sock
()
37 self
._sock
.bind(self
._address
)
41 if isinstance(self
._address
, tuple):
42 family
= socket
.AF_INET
44 family
= socket
.AF_UNIX
45 return socket
.socket(family
, socket
.SOCK_STREAM
)
49 Connect to the qtest socket.
51 @raise socket.error on socket connection errors
53 self
._sock
.connect(self
._address
)
54 self
._sockfile
= self
._sock
.makefile()
58 Await connection from QEMU.
60 @raise socket.error on socket connection errors
62 self
._sock
, _
= self
._sock
.accept()
63 self
._sockfile
= self
._sock
.makefile()
65 def cmd(self
, qtest_cmd
):
67 Send a qtest command on the wire.
69 @param qtest_cmd: qtest command text to be sent
71 self
._sock
.sendall((qtest_cmd
+ "\n").encode('utf-8'))
72 resp
= self
._sockfile
.readline()
77 self
._sockfile
.close()
79 def settimeout(self
, timeout
):
80 self
._sock
.settimeout(timeout
)
83 class QEMUQtestMachine(QEMUMachine
):
86 def __init__(self
, binary
, args
=None, name
=None, test_dir
="/var/tmp",
87 socket_scm_helper
=None, sock_dir
=None):
89 name
= "qemu-%d" % os
.getpid()
92 super(QEMUQtestMachine
,
93 self
).__init
__(binary
, args
, name
=name
, test_dir
=test_dir
,
94 socket_scm_helper
=socket_scm_helper
,
97 self
._qtest
_path
= os
.path
.join(sock_dir
, name
+ "-qtest.sock")
100 args
= super(QEMUQtestMachine
, self
)._base
_args
()
101 args
.extend(['-qtest', 'unix:path=' + self
._qtest
_path
,
105 def _pre_launch(self
):
106 super(QEMUQtestMachine
, self
)._pre
_launch
()
107 self
._qtest
= QEMUQtestProtocol(self
._qtest
_path
, server
=True)
109 def _post_launch(self
):
110 super(QEMUQtestMachine
, self
)._post
_launch
()
113 def _post_shutdown(self
):
114 super(QEMUQtestMachine
, self
)._post
_shutdown
()
115 self
._remove
_if
_exists
(self
._qtest
_path
)
117 def qtest(self
, cmd
):
118 '''Send a qtest command to guest'''
119 return self
._qtest
.cmd(cmd
)