tests/qemu-iotests/iotests.py

   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
  19 import errno
  20 import os
  21 import re
  22 import subprocess
  23 import string
  24 import unittest
  25 import sys
  26 sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'scripts'))
  27 import qtest
  28 import struct
  29 import json
  30
  31
  32 # This will not work if arguments contain spaces but is necessary if we
  33 # want to support the override options that ./check supports.
  34 qemu_img_args = [os.environ.get('QEMU_IMG_PROG', 'qemu-img')]
  35 if os.environ.get('QEMU_IMG_OPTIONS'):
  36     qemu_img_args += os.environ['QEMU_IMG_OPTIONS'].strip().split(' ')
  37
  38 qemu_io_args = [os.environ.get('QEMU_IO_PROG', 'qemu-io')]
  39 if os.environ.get('QEMU_IO_OPTIONS'):
  40     qemu_io_args += os.environ['QEMU_IO_OPTIONS'].strip().split(' ')
  41
  42 qemu_prog = os.environ.get('QEMU_PROG', 'qemu')
  43 qemu_opts = os.environ.get('QEMU_OPTIONS', '').strip().split(' ')
  44
  45 imgfmt = os.environ.get('IMGFMT', 'raw')
  46 imgproto = os.environ.get('IMGPROTO', 'file')
  47 test_dir = os.environ.get('TEST_DIR')
  48 output_dir = os.environ.get('OUTPUT_DIR', '.')
  49 cachemode = os.environ.get('CACHEMODE')
  50 qemu_default_machine = os.environ.get('QEMU_DEFAULT_MACHINE')
  51
  52 socket_scm_helper = os.environ.get('SOCKET_SCM_HELPER', 'socket_scm_helper')
  53 debug = False
  54
  55 def qemu_img(*args):
  56     '''Run qemu-img and return the exit code'''
  57     devnull = open('/dev/null', 'r+')
  58     exitcode = subprocess.call(qemu_img_args + list(args), stdin=devnull, stdout=devnull)
  59     if exitcode < 0:
  60         sys.stderr.write('qemu-img received signal %i: %s\n' % (-exitcode, ' '.join(qemu_img_args + list(args))))
  61     return exitcode
  62
  63 def qemu_img_verbose(*args):
  64     '''Run qemu-img without suppressing its output and return the exit code'''
  65     exitcode = subprocess.call(qemu_img_args + list(args))
  66     if exitcode < 0:
  67         sys.stderr.write('qemu-img received signal %i: %s\n' % (-exitcode, ' '.join(qemu_img_args + list(args))))
  68     return exitcode
  69
  70 def qemu_img_pipe(*args):
  71     '''Run qemu-img and return its output'''
  72     subp = subprocess.Popen(qemu_img_args + list(args),
  73                             stdout=subprocess.PIPE,
  74                             stderr=subprocess.STDOUT)
  75     exitcode = subp.wait()
  76     if exitcode < 0:
  77         sys.stderr.write('qemu-img received signal %i: %s\n' % (-exitcode, ' '.join(qemu_img_args + list(args))))
  78     return subp.communicate()[0]
  79
  80 def qemu_io(*args):
  81     '''Run qemu-io and return the stdout data'''
  82     args = qemu_io_args + list(args)
  83     subp = subprocess.Popen(args, stdout=subprocess.PIPE,
  84                             stderr=subprocess.STDOUT)
  85     exitcode = subp.wait()
  86     if exitcode < 0:
  87         sys.stderr.write('qemu-io received signal %i: %s\n' % (-exitcode, ' '.join(args)))
  88     return subp.communicate()[0]
  89
  90 def compare_images(img1, img2, fmt1=imgfmt, fmt2=imgfmt):
  91     '''Return True if two image files are identical'''
  92     return qemu_img('compare', '-f', fmt1,
  93                     '-F', fmt2, img1, img2) == 0
  94
  95 def create_image(name, size):
  96     '''Create a fully-allocated raw image with sector markers'''
  97     file = open(name, 'w')
  98     i = 0
  99     while i < size:
 100         sector = struct.pack('>l504xl', i / 512, i / 512)
 101         file.write(sector)
 102         i = i + 512
 103     file.close()
 104
 105 def image_size(img):
 106     '''Return image's virtual size'''
 107     r = qemu_img_pipe('info', '--output=json', '-f', imgfmt, img)
 108     return json.loads(r)['virtual-size']
 109
 110 test_dir_re = re.compile(r"%s" % test_dir)
 111 def filter_test_dir(msg):
 112     return test_dir_re.sub("TEST_DIR", msg)
 113
 114 win32_re = re.compile(r"\r")
 115 def filter_win32(msg):
 116     return win32_re.sub("", msg)
 117
 118 qemu_io_re = re.compile(r"[0-9]* ops; [0-9\/:. sec]* \([0-9\/.inf]* [EPTGMKiBbytes]*\/sec and [0-9\/.inf]* ops\/sec\)")
 119 def filter_qemu_io(msg):
 120     msg = filter_win32(msg)
 121     return qemu_io_re.sub("X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec)", msg)
 122
 123 chown_re = re.compile(r"chown [0-9]+:[0-9]+")
 124 def filter_chown(msg):
 125     return chown_re.sub("chown UID:GID", msg)
 126
 127 def log(msg, filters=[]):
 128     for flt in filters:
 129         msg = flt(msg)
 130     print msg
 131
 132 class VM(qtest.QEMUQtestMachine):
 133     '''A QEMU VM'''
 134
 135     def __init__(self):
 136         super(VM, self).__init__(qemu_prog, qemu_opts, test_dir=test_dir,
 137                                  socket_scm_helper=socket_scm_helper)
 138         if debug:
 139             self._debug = True
 140         self._num_drives = 0
 141
 142     def add_drive_raw(self, opts):
 143         self._args.append('-drive')
 144         self._args.append(opts)
 145         return self
 146
 147     def add_drive(self, path, opts='', interface='virtio', format=imgfmt):
 148         '''Add a virtio-blk drive to the VM'''
 149         options = ['if=%s' % interface,
 150                    'id=drive%d' % self._num_drives]
 151
 152         if path is not None:
 153             options.append('file=%s' % path)
 154             options.append('format=%s' % format)
 155             options.append('cache=%s' % cachemode)
 156
 157         if opts:
 158             options.append(opts)
 159
 160         self._args.append('-drive')
 161         self._args.append(','.join(options))
 162         self._num_drives += 1
 163         return self
 164
 165     def pause_drive(self, drive, event=None):
 166         '''Pause drive r/w operations'''
 167         if not event:
 168             self.pause_drive(drive, "read_aio")
 169             self.pause_drive(drive, "write_aio")
 170             return
 171         self.qmp('human-monitor-command',
 172                     command_line='qemu-io %s "break %s bp_%s"' % (drive, event, drive))
 173
 174     def resume_drive(self, drive):
 175         self.qmp('human-monitor-command',
 176                     command_line='qemu-io %s "remove_break bp_%s"' % (drive, drive))
 177
 178     def hmp_qemu_io(self, drive, cmd):
 179         '''Write to a given drive using an HMP command'''
 180         return self.qmp('human-monitor-command',
 181                         command_line='qemu-io %s "%s"' % (drive, cmd))
 182
 183
 184 index_re = re.compile(r'([^\[]+)\[([^\]]+)\]')
 185
 186 class QMPTestCase(unittest.TestCase):
 187     '''Abstract base class for QMP test cases'''
 188
 189     def dictpath(self, d, path):
 190         '''Traverse a path in a nested dict'''
 191         for component in path.split('/'):
 192             m = index_re.match(component)
 193             if m:
 194                 component, idx = m.groups()
 195                 idx = int(idx)
 196
 197             if not isinstance(d, dict) or component not in d:
 198                 self.fail('failed path traversal for "%s" in "%s"' % (path, str(d)))
 199             d = d[component]
 200
 201             if m:
 202                 if not isinstance(d, list):
 203                     self.fail('path component "%s" in "%s" is not a list in "%s"' % (component, path, str(d)))
 204                 try:
 205                     d = d[idx]
 206                 except IndexError:
 207                     self.fail('invalid index "%s" in path "%s" in "%s"' % (idx, path, str(d)))
 208         return d
 209
 210     def assert_qmp_absent(self, d, path):
 211         try:
 212             result = self.dictpath(d, path)
 213         except AssertionError:
 214             return
 215         self.fail('path "%s" has value "%s"' % (path, str(result)))
 216
 217     def assert_qmp(self, d, path, value):
 218         '''Assert that the value for a specific path in a QMP dict matches'''
 219         result = self.dictpath(d, path)
 220         self.assertEqual(result, value, 'values not equal "%s" and "%s"' % (str(result), str(value)))
 221
 222     def assert_no_active_block_jobs(self):
 223         result = self.vm.qmp('query-block-jobs')
 224         self.assert_qmp(result, 'return', [])
 225
 226     def assert_has_block_node(self, node_name=None, file_name=None):
 227         """Issue a query-named-block-nodes and assert node_name and/or
 228         file_name is present in the result"""
 229         def check_equal_or_none(a, b):
 230             return a == None or b == None or a == b
 231         assert node_name or file_name
 232         result = self.vm.qmp('query-named-block-nodes')
 233         for x in result["return"]:
 234             if check_equal_or_none(x.get("node-name"), node_name) and \
 235                     check_equal_or_none(x.get("file"), file_name):
 236                 return
 237         self.assertTrue(False, "Cannot find %s %s in result:\n%s" % \
 238                 (node_name, file_name, result))
 239
 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', {})
 244
 245         if resume:
 246             self.vm.resume_drive(drive)
 247
 248         cancelled = False
 249         result = None
 250         while not cancelled:
 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)
 255                     result = event
 256                     cancelled = True
 257
 258         self.assert_no_active_block_jobs()
 259         return result
 260
 261     def wait_until_completed(self, drive='drive0', check_offset=True):
 262         '''Wait for a block job to finish, returning the event'''
 263         completed = False
 264         while not completed:
 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')
 269                     if check_offset:
 270                         self.assert_qmp(event, 'data/offset', event['data']['len'])
 271                     completed = True
 272
 273         self.assert_no_active_block_jobs()
 274         return event
 275
 276     def wait_ready(self, drive='drive0'):
 277         '''Wait until a block job BLOCK_JOB_READY event'''
 278         f = {'data': {'type': 'mirror', 'device': drive } }
 279         event = self.vm.event_wait(name='BLOCK_JOB_READY', match=f)
 280
 281     def wait_ready_and_cancel(self, drive='drive0'):
 282         self.wait_ready(drive=drive)
 283         event = self.cancel_and_wait(drive=drive)
 284         self.assertEquals(event['event'], 'BLOCK_JOB_COMPLETED')
 285         self.assert_qmp(event, 'data/type', 'mirror')
 286         self.assert_qmp(event, 'data/offset', event['data']['len'])
 287
 288     def complete_and_wait(self, drive='drive0', wait_ready=True):
 289         '''Complete a block job and wait for it to finish'''
 290         if wait_ready:
 291             self.wait_ready(drive=drive)
 292
 293         result = self.vm.qmp('block-job-complete', device=drive)
 294         self.assert_qmp(result, 'return', {})
 295
 296         event = self.wait_until_completed(drive=drive)
 297         self.assert_qmp(event, 'data/type', 'mirror')
 298
 299 def notrun(reason):
 300     '''Skip this test suite'''
 301     # Each test in qemu-iotests has a number ("seq")
 302     seq = os.path.basename(sys.argv[0])
 303
 304     open('%s/%s.notrun' % (output_dir, seq), 'wb').write(reason + '\n')
 305     print '%s not run: %s' % (seq, reason)
 306     sys.exit(0)
 307
 308 def verify_image_format(supported_fmts=[]):
 309     if supported_fmts and (imgfmt not in supported_fmts):
 310         notrun('not suitable for this image format: %s' % imgfmt)
 311
 312 def verify_platform(supported_oses=['linux']):
 313     if True not in [sys.platform.startswith(x) for x in supported_oses]:
 314         notrun('not suitable for this OS: %s' % sys.platform)
 315
 316 def verify_quorum():
 317     '''Skip test suite if quorum support is not available'''
 318     if 'quorum' not in qemu_img_pipe('--help'):
 319         notrun('quorum support missing')
 320
 321 def main(supported_fmts=[], supported_oses=['linux']):
 322     '''Run tests'''
 323
 324     global debug
 325
 326     # We are using TEST_DIR and QEMU_DEFAULT_MACHINE as proxies to
 327     # indicate that we're not being run via "check". There may be
 328     # other things set up by "check" that individual test cases rely
 329     # on.
 330     if test_dir is None or qemu_default_machine is None:
 331         sys.stderr.write('Please run this test via the "check" script\n')
 332         sys.exit(os.EX_USAGE)
 333
 334     debug = '-d' in sys.argv
 335     verbosity = 1
 336     verify_image_format(supported_fmts)
 337     verify_platform(supported_oses)
 338
 339     # We need to filter out the time taken from the output so that qemu-iotest
 340     # can reliably diff the results against master output.
 341     import StringIO
 342     if debug:
 343         output = sys.stdout
 344         verbosity = 2
 345         sys.argv.remove('-d')
 346     else:
 347         output = StringIO.StringIO()
 348
 349     class MyTestRunner(unittest.TextTestRunner):
 350         def __init__(self, stream=output, descriptions=True, verbosity=verbosity):
 351             unittest.TextTestRunner.__init__(self, stream, descriptions, verbosity)
 352
 353     # unittest.main() will use sys.exit() so expect a SystemExit exception
 354     try:
 355         unittest.main(testRunner=MyTestRunner)
 356     finally:
 357         if not debug:
 358             sys.stderr.write(re.sub(r'Ran (\d+) tests? in [\d.]+s', r'Ran \1 tests', output.getvalue()))