1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
"""
QEMU qtest library
qtest offers the QEMUQtestProtocol and QEMUQTestMachine classes, which
offer a connection to QEMU's qtest protocol socket, and a qtest-enabled
subclass of QEMUMachine, respectively.
"""
# Copyright (C) 2015 Red Hat Inc.
#
# Authors:
# Fam Zheng <famz@redhat.com>
#
# This work is licensed under the terms of the GNU GPL, version 2. See
# the COPYING file in the top-level directory.
#
# Based on qmp.py.
#
import socket
import os
from .machine import QEMUMachine
class QEMUQtestProtocol:
"""
QEMUQtestProtocol implements a connection to a qtest socket.
:param address: QEMU address, can be either a unix socket path (string)
or a tuple in the form ( address, port ) for a TCP
connection
:param server: server mode, listens on the socket (bool)
:raise socket.error: on socket connection errors
.. note::
No conection is estabalished by __init__(), this is done
by the connect() or accept() methods.
"""
def __init__(self, address, server=False):
self._address = address
self._sock = self._get_sock()
self._sockfile = None
if server:
self._sock.bind(self._address)
self._sock.listen(1)
def _get_sock(self):
if isinstance(self._address, tuple):
family = socket.AF_INET
else:
family = socket.AF_UNIX
return socket.socket(family, socket.SOCK_STREAM)
def connect(self):
"""
Connect to the qtest socket.
@raise socket.error on socket connection errors
"""
self._sock.connect(self._address)
self._sockfile = self._sock.makefile()
def accept(self):
"""
Await connection from QEMU.
@raise socket.error on socket connection errors
"""
self._sock, _ = self._sock.accept()
self._sockfile = self._sock.makefile()
def cmd(self, qtest_cmd):
"""
Send a qtest command on the wire.
@param qtest_cmd: qtest command text to be sent
"""
self._sock.sendall((qtest_cmd + "\n").encode('utf-8'))
resp = self._sockfile.readline()
return resp
def close(self):
"""Close this socket."""
self._sock.close()
self._sockfile.close()
def settimeout(self, timeout):
"""Set a timeout, in seconds."""
self._sock.settimeout(timeout)
class QEMUQtestMachine(QEMUMachine):
"""
A QEMU VM, with a qtest socket available.
"""
def __init__(self, binary, args=None, name=None, test_dir="/var/tmp",
socket_scm_helper=None, sock_dir=None):
if name is None:
name = "qemu-%d" % os.getpid()
if sock_dir is None:
sock_dir = test_dir
super(QEMUQtestMachine,
self).__init__(binary, args, name=name, test_dir=test_dir,
socket_scm_helper=socket_scm_helper,
sock_dir=sock_dir)
self._qtest = None
self._qtest_path = os.path.join(sock_dir, name + "-qtest.sock")
def _base_args(self):
args = super(QEMUQtestMachine, self)._base_args()
args.extend(['-qtest', 'unix:path=' + self._qtest_path,
'-accel', 'qtest'])
return args
def _pre_launch(self):
super(QEMUQtestMachine, self)._pre_launch()
self._qtest = QEMUQtestProtocol(self._qtest_path, server=True)
def _post_launch(self):
super(QEMUQtestMachine, self)._post_launch()
self._qtest.accept()
def _post_shutdown(self):
super(QEMUQtestMachine, self)._post_shutdown()
self._remove_if_exists(self._qtest_path)
def qtest(self, cmd):
'''Send a qtest command to guest'''
return self._qtest.cmd(cmd)
|
