1 # Common utilities and Python wrappers for qemu-iotests
3 # Copyright (C) 2012 IBM Corp.
5 # This program is free software; you can redistribute it and/or modify
6 # it under the terms of the GNU General Public License as published by
7 # the Free Software Foundation; either version 2 of the License, or
8 # (at your option) any later version.
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 # GNU General Public License for more details.
15 # You should have received a copy of the GNU General Public License
16 # along with this program. If not, see <http://www.gnu.org/licenses/>.
24 import sys
; sys
.path
.append(os
.path
.join(os
.path
.dirname(__file__
), '..', '..', 'scripts', 'qmp'))
28 __all__
= ['imgfmt', 'imgproto', 'test_dir' 'qemu_img', 'qemu_io',
29 'VM', 'QMPTestCase', 'notrun', 'main']
31 # This will not work if arguments or path contain spaces but is necessary if we
32 # want to support the override options that ./check supports.
33 qemu_img_args
= os
.environ
.get('QEMU_IMG', 'qemu-img').strip().split(' ')
34 qemu_io_args
= os
.environ
.get('QEMU_IO', 'qemu-io').strip().split(' ')
35 qemu_args
= os
.environ
.get('QEMU', 'qemu').strip().split(' ')
37 imgfmt
= os
.environ
.get('IMGFMT', 'raw')
38 imgproto
= os
.environ
.get('IMGPROTO', 'file')
39 test_dir
= os
.environ
.get('TEST_DIR', '/var/tmp')
40 output_dir
= os
.environ
.get('OUTPUT_DIR', '.')
41 cachemode
= os
.environ
.get('CACHEMODE')
43 socket_scm_helper
= os
.environ
.get('SOCKET_SCM_HELPER', 'socket_scm_helper')
46 '''Run qemu-img and return the exit code'''
47 devnull
= open('/dev/null', 'r+')
48 return subprocess
.call(qemu_img_args
+ list(args
), stdin
=devnull
, stdout
=devnull
)
50 def qemu_img_verbose(*args
):
51 '''Run qemu-img without suppressing its output and return the exit code'''
52 return subprocess
.call(qemu_img_args
+ list(args
))
54 def qemu_img_pipe(*args
):
55 '''Run qemu-img and return its output'''
56 return subprocess
.Popen(qemu_img_args
+ list(args
), stdout
=subprocess
.PIPE
).communicate()[0]
59 '''Run qemu-io and return the stdout data'''
60 args
= qemu_io_args
+ list(args
)
61 return subprocess
.Popen(args
, stdout
=subprocess
.PIPE
).communicate()[0]
63 def compare_images(img1
, img2
):
64 '''Return True if two image files are identical'''
65 return qemu_img('compare', '-f', imgfmt
,
66 '-F', imgfmt
, img1
, img2
) == 0
68 def create_image(name
, size
):
69 '''Create a fully-allocated raw image with sector markers'''
70 file = open(name
, 'w')
73 sector
= struct
.pack('>l504xl', i
/ 512, i
/ 512)
82 self
._monitor
_path
= os
.path
.join(test_dir
, 'qemu-mon.%d' % os
.getpid())
83 self
._qemu
_log
_path
= os
.path
.join(test_dir
, 'qemu-log.%d' % os
.getpid())
84 self
._args
= qemu_args
+ ['-chardev',
85 'socket,id=mon,path=' + self
._monitor
_path
,
86 '-mon', 'chardev=mon,mode=control',
87 '-qtest', 'stdio', '-machine', 'accel=qtest',
88 '-display', 'none', '-vga', 'none']
91 # This can be used to add an unused monitor instance.
92 def add_monitor_telnet(self
, ip
, port
):
93 args
= 'tcp:%s:%d,server,nowait,telnet' % (ip
, port
)
94 self
._args
.append('-monitor')
95 self
._args
.append(args
)
97 def add_drive(self
, path
, opts
=''):
98 '''Add a virtio-blk drive to the VM'''
99 options
= ['if=virtio',
100 'format=%s' % imgfmt
,
101 'cache=%s' % cachemode
,
103 'id=drive%d' % self
._num
_drives
]
107 self
._args
.append('-drive')
108 self
._args
.append(','.join(options
))
109 self
._num
_drives
+= 1
112 def pause_drive(self
, drive
, event
=None):
113 '''Pause drive r/w operations'''
115 self
.pause_drive(drive
, "read_aio")
116 self
.pause_drive(drive
, "write_aio")
118 self
.qmp('human-monitor-command',
119 command_line
='qemu-io %s "break %s bp_%s"' % (drive
, event
, drive
))
121 def resume_drive(self
, drive
):
122 self
.qmp('human-monitor-command',
123 command_line
='qemu-io %s "remove_break bp_%s"' % (drive
, drive
))
125 def hmp_qemu_io(self
, drive
, cmd
):
126 '''Write to a given drive using an HMP command'''
127 return self
.qmp('human-monitor-command',
128 command_line
='qemu-io %s "%s"' % (drive
, cmd
))
130 def add_fd(self
, fd
, fdset
, opaque
, opts
=''):
131 '''Pass a file descriptor to the VM'''
132 options
= ['fd=%d' % fd
,
134 'opaque=%s' % opaque
]
138 self
._args
.append('-add-fd')
139 self
._args
.append(','.join(options
))
142 def send_fd_scm(self
, fd_file_path
):
143 # In iotest.py, the qmp should always use unix socket.
144 assert self
._qmp
.is_scm_available()
145 bin
= socket_scm_helper
146 if os
.path
.exists(bin
) == False:
147 print "Scm help program does not present, path '%s'." % bin
149 fd_param
= ["%s" % bin
,
150 "%d" % self
._qmp
.get_sock_fd(),
152 devnull
= open('/dev/null', 'rb')
153 p
= subprocess
.Popen(fd_param
, stdin
=devnull
, stdout
=sys
.stdout
,
158 '''Launch the VM and establish a QMP connection'''
159 devnull
= open('/dev/null', 'rb')
160 qemulog
= open(self
._qemu
_log
_path
, 'wb')
162 self
._qmp
= qmp
.QEMUMonitorProtocol(self
._monitor
_path
, server
=True)
163 self
._popen
= subprocess
.Popen(self
._args
, stdin
=devnull
, stdout
=qemulog
,
164 stderr
=subprocess
.STDOUT
)
167 os
.remove(self
._monitor
_path
)
171 '''Terminate the VM and clean up'''
172 if not self
._popen
is None:
173 self
._qmp
.cmd('quit')
175 os
.remove(self
._monitor
_path
)
176 os
.remove(self
._qemu
_log
_path
)
179 underscore_to_dash
= string
.maketrans('_', '-')
180 def qmp(self
, cmd
, **args
):
181 '''Invoke a QMP command and return the result dict'''
183 for k
in args
.keys():
184 qmp_args
[k
.translate(self
.underscore_to_dash
)] = args
[k
]
186 return self
._qmp
.cmd(cmd
, args
=qmp_args
)
188 def get_qmp_event(self
, wait
=False):
189 '''Poll for one queued QMP events and return it'''
190 return self
._qmp
.pull_event(wait
=wait
)
192 def get_qmp_events(self
, wait
=False):
193 '''Poll for queued QMP events and return a list of dicts'''
194 events
= self
._qmp
.get_events(wait
=wait
)
195 self
._qmp
.clear_events()
198 index_re
= re
.compile(r
'([^\[]+)\[([^\]]+)\]')
200 class QMPTestCase(unittest
.TestCase
):
201 '''Abstract base class for QMP test cases'''
203 def dictpath(self
, d
, path
):
204 '''Traverse a path in a nested dict'''
205 for component
in path
.split('/'):
206 m
= index_re
.match(component
)
208 component
, idx
= m
.groups()
211 if not isinstance(d
, dict) or component
not in d
:
212 self
.fail('failed path traversal for "%s" in "%s"' % (path
, str(d
)))
216 if not isinstance(d
, list):
217 self
.fail('path component "%s" in "%s" is not a list in "%s"' % (component
, path
, str(d
)))
221 self
.fail('invalid index "%s" in path "%s" in "%s"' % (idx
, path
, str(d
)))
224 def assert_qmp_absent(self
, d
, path
):
226 result
= self
.dictpath(d
, path
)
227 except AssertionError:
229 self
.fail('path "%s" has value "%s"' % (path
, str(result
)))
231 def assert_qmp(self
, d
, path
, value
):
232 '''Assert that the value for a specific path in a QMP dict matches'''
233 result
= self
.dictpath(d
, path
)
234 self
.assertEqual(result
, value
, 'values not equal "%s" and "%s"' % (str(result
), str(value
)))
236 def assert_no_active_block_jobs(self
):
237 result
= self
.vm
.qmp('query-block-jobs')
238 self
.assert_qmp(result
, 'return', [])
240 def cancel_and_wait(self
, drive
='drive0', force
=False, resume
=False):
241 '''Cancel a block job and wait for it to finish, returning the event'''
242 result
= self
.vm
.qmp('block-job-cancel', device
=drive
, force
=force
)
243 self
.assert_qmp(result
, 'return', {})
246 self
.vm
.resume_drive(drive
)
251 for event
in self
.vm
.get_qmp_events(wait
=True):
252 if event
['event'] == 'BLOCK_JOB_COMPLETED' or \
253 event
['event'] == 'BLOCK_JOB_CANCELLED':
254 self
.assert_qmp(event
, 'data/device', drive
)
258 self
.assert_no_active_block_jobs()
261 def wait_until_completed(self
, drive
='drive0', check_offset
=True):
262 '''Wait for a block job to finish, returning the event'''
265 for event
in self
.vm
.get_qmp_events(wait
=True):
266 if event
['event'] == 'BLOCK_JOB_COMPLETED':
267 self
.assert_qmp(event
, 'data/device', drive
)
268 self
.assert_qmp_absent(event
, 'data/error')
270 self
.assert_qmp(event
, 'data/offset', self
.image_len
)
271 self
.assert_qmp(event
, 'data/len', self
.image_len
)
274 self
.assert_no_active_block_jobs()
278 '''Skip this test suite'''
279 # Each test in qemu-iotests has a number ("seq")
280 seq
= os
.path
.basename(sys
.argv
[0])
282 open('%s/%s.notrun' % (output_dir
, seq
), 'wb').write(reason
+ '\n')
283 print '%s not run: %s' % (seq
, reason
)
286 def main(supported_fmts
=[]):
289 if supported_fmts
and (imgfmt
not in supported_fmts
):
290 notrun('not suitable for this image format: %s' % imgfmt
)
292 # We need to filter out the time taken from the output so that qemu-iotest
293 # can reliably diff the results against master output.
295 output
= StringIO
.StringIO()
297 class MyTestRunner(unittest
.TextTestRunner
):
298 def __init__(self
, stream
=output
, descriptions
=True, verbosity
=1):
299 unittest
.TextTestRunner
.__init
__(self
, stream
, descriptions
, verbosity
)
301 # unittest.main() will use sys.exit() so expect a SystemExit exception
303 unittest
.main(testRunner
=MyTestRunner
)
305 sys
.stderr
.write(re
.sub(r
'Ran (\d+) tests? in [\d.]+s', r
'Ran \1 tests', output
.getvalue()))