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/>.
23 import sys
; sys
.path
.append(os
.path
.join(os
.path
.dirname(__file__
), '..', '..', 'QMP'))
26 __all__
= ['imgfmt', 'imgproto', 'test_dir' 'qemu_img', 'qemu_io',
27 'VM', 'QMPTestCase', 'notrun', 'main']
29 # This will not work if arguments or path contain spaces but is necessary if we
30 # want to support the override options that ./check supports.
31 qemu_img_args
= os
.environ
.get('QEMU_IMG', 'qemu-img').split(' ')
32 qemu_io_args
= os
.environ
.get('QEMU_IO', 'qemu-io').split(' ')
33 qemu_args
= os
.environ
.get('QEMU', 'qemu').split(' ')
35 imgfmt
= os
.environ
.get('IMGFMT', 'raw')
36 imgproto
= os
.environ
.get('IMGPROTO', 'file')
37 test_dir
= os
.environ
.get('TEST_DIR', '/var/tmp')
40 '''Run qemu-img and return the exit code'''
41 devnull
= open('/dev/null', 'r+')
42 return subprocess
.call(qemu_img_args
+ list(args
), stdin
=devnull
, stdout
=devnull
)
45 '''Run qemu-io and return the stdout data'''
46 args
= qemu_io_args
+ list(args
)
47 return subprocess
.Popen(args
, stdout
=subprocess
.PIPE
).communicate()[0]
53 self
._monitor
_path
= os
.path
.join(test_dir
, 'qemu-mon.%d' % os
.getpid())
54 self
._qemu
_log
_path
= os
.path
.join(test_dir
, 'qemu-log.%d' % os
.getpid())
55 self
._args
= qemu_args
+ ['-chardev',
56 'socket,id=mon,path=' + self
._monitor
_path
,
57 '-mon', 'chardev=mon,mode=control', '-nographic']
60 def add_drive(self
, path
, opts
=''):
61 '''Add a virtio-blk drive to the VM'''
62 options
= ['if=virtio',
66 'id=drive%d' % self
._num
_drives
]
70 self
._args
.append('-drive')
71 self
._args
.append(','.join(options
))
76 '''Launch the VM and establish a QMP connection'''
77 devnull
= open('/dev/null', 'rb')
78 qemulog
= open(self
._qemu
_log
_path
, 'wb')
80 self
._qmp
= qmp
.QEMUMonitorProtocol(self
._monitor
_path
, server
=True)
81 self
._popen
= subprocess
.Popen(self
._args
, stdin
=devnull
, stdout
=qemulog
,
82 stderr
=subprocess
.STDOUT
)
85 os
.remove(self
._monitor
_path
)
89 '''Terminate the VM and clean up'''
92 os
.remove(self
._monitor
_path
)
93 os
.remove(self
._qemu
_log
_path
)
95 def qmp(self
, cmd
, **args
):
96 '''Invoke a QMP command and return the result dict'''
97 return self
._qmp
.cmd(cmd
, args
=args
)
99 def get_qmp_events(self
, wait
=False):
100 '''Poll for queued QMP events and return a list of dicts'''
101 events
= self
._qmp
.get_events(wait
=wait
)
102 self
._qmp
.clear_events()
105 index_re
= re
.compile(r
'([^\[]+)\[([^\]]+)\]')
107 class QMPTestCase(unittest
.TestCase
):
108 '''Abstract base class for QMP test cases'''
110 def dictpath(self
, d
, path
):
111 '''Traverse a path in a nested dict'''
112 for component
in path
.split('/'):
113 m
= index_re
.match(component
)
115 component
, idx
= m
.groups()
118 if not isinstance(d
, dict) or component
not in d
:
119 self
.fail('failed path traversal for "%s" in "%s"' % (path
, str(d
)))
123 if not isinstance(d
, list):
124 self
.fail('path component "%s" in "%s" is not a list in "%s"' % (component
, path
, str(d
)))
128 self
.fail('invalid index "%s" in path "%s" in "%s"' % (idx
, path
, str(d
)))
131 def assert_qmp(self
, d
, path
, value
):
132 '''Assert that the value for a specific path in a QMP dict matches'''
133 result
= self
.dictpath(d
, path
)
134 self
.assertEqual(result
, value
, 'values not equal "%s" and "%s"' % (str(result
), str(value
)))
137 '''Skip this test suite'''
138 # Each test in qemu-iotests has a number ("seq")
139 seq
= os
.path
.basename(sys
.argv
[0])
141 open('%s.notrun' % seq
, 'wb').write(reason
+ '\n')
142 print '%s not run: %s' % (seq
, reason
)
145 def main(supported_fmts
=[]):
148 if supported_fmts
and (imgfmt
not in supported_fmts
):
149 notrun('not suitable for this image format: %s' % imgfmt
)
151 # We need to filter out the time taken from the output so that qemu-iotest
152 # can reliably diff the results against master output.
154 output
= StringIO
.StringIO()
156 class MyTestRunner(unittest
.TextTestRunner
):
157 def __init__(self
, stream
=output
, descriptions
=True, verbosity
=1):
158 unittest
.TextTestRunner
.__init
__(self
, stream
, descriptions
, verbosity
)
160 # unittest.main() will use sys.exit() so expect a SystemExit exception
162 unittest
.main(testRunner
=MyTestRunner
)
164 sys
.stderr
.write(re
.sub(r
'Ran (\d+) test[s] in [\d.]+s', r
'Ran \1 tests', output
.getvalue()))