xref: /qemu/tests/qemu-iotests/iotests.py (revision b3c09bde4842f1b157bc4431135aad4f2031e312)
1# Common utilities and Python wrappers for qemu-iotests
2#
3# Copyright (C) 2012 IBM Corp.
4#
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.
9#
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.
14#
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/>.
17#
18
19import os
20import re
21import subprocess
22import unittest
23import sys; sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'QMP'))
24import qmp
25
26__all__ = ['imgfmt', 'imgproto', 'test_dir' 'qemu_img', 'qemu_io',
27           'VM', 'QMPTestCase', 'notrun', 'main']
28
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.
31qemu_img_args = os.environ.get('QEMU_IMG', 'qemu-img').split(' ')
32qemu_io_args = os.environ.get('QEMU_IO', 'qemu-io').split(' ')
33qemu_args = os.environ.get('QEMU', 'qemu').split(' ')
34
35imgfmt = os.environ.get('IMGFMT', 'raw')
36imgproto = os.environ.get('IMGPROTO', 'file')
37test_dir = os.environ.get('TEST_DIR', '/var/tmp')
38
39def qemu_img(*args):
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)
43
44def qemu_io(*args):
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]
48
49class VM(object):
50    '''A QEMU VM'''
51
52    def __init__(self):
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']
58        self._num_drives = 0
59
60    def add_drive(self, path, opts=''):
61        '''Add a virtio-blk drive to the VM'''
62        options = ['if=virtio',
63                   'format=%s' % imgfmt,
64                   'cache=none',
65                   'file=%s' % path,
66                   'id=drive%d' % self._num_drives]
67        if opts:
68            options.append(opts)
69
70        self._args.append('-drive')
71        self._args.append(','.join(options))
72        self._num_drives += 1
73        return self
74
75    def launch(self):
76        '''Launch the VM and establish a QMP connection'''
77        devnull = open('/dev/null', 'rb')
78        qemulog = open(self._qemu_log_path, 'wb')
79        try:
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)
83            self._qmp.accept()
84        except:
85            os.remove(self._monitor_path)
86            raise
87
88    def shutdown(self):
89        '''Terminate the VM and clean up'''
90        self._qmp.cmd('quit')
91        self._popen.wait()
92        os.remove(self._monitor_path)
93        os.remove(self._qemu_log_path)
94
95    def qmp(self, cmd, **args):
96        '''Invoke a QMP command and return the result dict'''
97        return self._qmp.cmd(cmd, args=args)
98
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()
103        return events
104
105index_re = re.compile(r'([^\[]+)\[([^\]]+)\]')
106
107class QMPTestCase(unittest.TestCase):
108    '''Abstract base class for QMP test cases'''
109
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)
114            if m:
115                component, idx = m.groups()
116                idx = int(idx)
117
118            if not isinstance(d, dict) or component not in d:
119                self.fail('failed path traversal for "%s" in "%s"' % (path, str(d)))
120            d = d[component]
121
122            if m:
123                if not isinstance(d, list):
124                    self.fail('path component "%s" in "%s" is not a list in "%s"' % (component, path, str(d)))
125                try:
126                    d = d[idx]
127                except IndexError:
128                    self.fail('invalid index "%s" in path "%s" in "%s"' % (idx, path, str(d)))
129        return d
130
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)))
135
136def notrun(reason):
137    '''Skip this test suite'''
138    # Each test in qemu-iotests has a number ("seq")
139    seq = os.path.basename(sys.argv[0])
140
141    open('%s.notrun' % seq, 'wb').write(reason + '\n')
142    print '%s not run: %s' % (seq, reason)
143    sys.exit(0)
144
145def main(supported_fmts=[]):
146    '''Run tests'''
147
148    if supported_fmts and (imgfmt not in supported_fmts):
149        notrun('not suitable for this image format: %s' % imgfmt)
150
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.
153    import StringIO
154    output = StringIO.StringIO()
155
156    class MyTestRunner(unittest.TextTestRunner):
157        def __init__(self, stream=output, descriptions=True, verbosity=1):
158            unittest.TextTestRunner.__init__(self, stream, descriptions, verbosity)
159
160    # unittest.main() will use sys.exit() so expect a SystemExit exception
161    try:
162        unittest.main(testRunner=MyTestRunner)
163    finally:
164        sys.stderr.write(re.sub(r'Ran (\d+) test[s] in [\d.]+s', r'Ran \1 tests', output.getvalue()))
165