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.
22 class QEMUQtestProtocol(object):
23 def __init__(self
, address
, server
=False):
25 Create a QEMUQtestProtocol object.
27 @param address: QEMU address, can be either a unix socket path (string)
28 or a tuple in the form ( address, port ) for a TCP
30 @param server: server mode, listens on the socket (bool)
31 @raise socket.error on socket connection errors
32 @note No connection is established, this is done by the connect() or
35 self
._address
= address
36 self
._sock
= self
._get
_sock
()
38 self
._sock
.bind(self
._address
)
42 if isinstance(self
._address
, tuple):
43 family
= socket
.AF_INET
45 family
= socket
.AF_UNIX
46 return socket
.socket(family
, socket
.SOCK_STREAM
)
50 Connect to the qtest socket.
52 @raise socket.error on socket connection errors
54 self
._sock
.connect(self
._address
)
58 Await connection from QEMU.
60 @raise socket.error on socket connection errors
62 self
._sock
, _
= self
._sock
.accept()
64 def cmd(self
, qtest_cmd
):
66 Send a qtest command on the wire.
68 @param qtest_cmd: qtest command text to be sent
70 self
._sock
.sendall(qtest_cmd
+ "\n")
75 def settimeout(self
, timeout
):
76 self
._sock
.settimeout(timeout
)
79 class QEMUQtestMachine(qemu
.QEMUMachine
):
82 def __init__(self
, binary
, args
=[], name
=None, test_dir
="/var/tmp"):
83 super(self
, QEMUQtestMachine
).__init
__(binary
, args
, name
, test_dir
)
84 self
._qtest
_path
= os
.path
.join(test_dir
, name
+ "-qtest.sock")
87 args
= super(self
, QEMUQtestMachine
)._base
_args
()
88 args
.extend(['-qtest', 'unix:path=' + self
._qtest
_path
])
91 def _pre_launch(self
):
92 super(self
, QEMUQtestMachine
)._pre
_launch
()
93 self
._qtest
= QEMUQtestProtocol(self
._qtest
_path
, server
=True)
95 def _post_launch(self
):
96 super(self
, QEMUQtestMachine
)._post
_launch
()
99 def _post_shutdown(self
):
100 super(self
, QEMUQtestMachine
)._post
_shutdown
()
101 self
._remove
_if
_exists
(self
._qtest
_path
)
103 def qtest(self
, cmd
):
104 '''Send a qtest command to guest'''
105 return self
._qtest
.cmd(cmd
)