Return to iotests.py CVS log

Up to [Qemu by Fabrice Bellard] / qemu / tests / qemu-iotests

Annotation of qemu/tests/qemu-iotests/iotests.py, revision 1.1

1.1     ! root        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 os
        !            20: import re
        !            21: import subprocess
        !            22: import unittest
        !            23: import sys; sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'QMP'))
        !            24: import 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.
        !            31: qemu_img_args = os.environ.get('QEMU_IMG', 'qemu-img').strip().split(' ')
        !            32: qemu_io_args = os.environ.get('QEMU_IO', 'qemu-io').strip().split(' ')
        !            33: qemu_args = os.environ.get('QEMU', 'qemu').strip().split(' ')
        !            34: 
        !            35: imgfmt = os.environ.get('IMGFMT', 'raw')
        !            36: imgproto = os.environ.get('IMGPROTO', 'file')
        !            37: test_dir = os.environ.get('TEST_DIR', '/var/tmp')
        !            38: 
        !            39: def 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: 
        !            44: def 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: 
        !            49: class 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:         if not self._popen is None:
        !            91:             self._qmp.cmd('quit')
        !            92:             self._popen.wait()
        !            93:             os.remove(self._monitor_path)
        !            94:             os.remove(self._qemu_log_path)
        !            95:             self._popen = None
        !            96: 
        !            97:     def qmp(self, cmd, **args):
        !            98:         '''Invoke a QMP command and return the result dict'''
        !            99:         return self._qmp.cmd(cmd, args=args)
        !           100: 
        !           101:     def get_qmp_events(self, wait=False):
        !           102:         '''Poll for queued QMP events and return a list of dicts'''
        !           103:         events = self._qmp.get_events(wait=wait)
        !           104:         self._qmp.clear_events()
        !           105:         return events
        !           106: 
        !           107: index_re = re.compile(r'([^\[]+)\[([^\]]+)\]')
        !           108: 
        !           109: class QMPTestCase(unittest.TestCase):
        !           110:     '''Abstract base class for QMP test cases'''
        !           111: 
        !           112:     def dictpath(self, d, path):
        !           113:         '''Traverse a path in a nested dict'''
        !           114:         for component in path.split('/'):
        !           115:             m = index_re.match(component)
        !           116:             if m:
        !           117:                 component, idx = m.groups()
        !           118:                 idx = int(idx)
        !           119: 
        !           120:             if not isinstance(d, dict) or component not in d:
        !           121:                 self.fail('failed path traversal for "%s" in "%s"' % (path, str(d)))
        !           122:             d = d[component]
        !           123: 
        !           124:             if m:
        !           125:                 if not isinstance(d, list):
        !           126:                     self.fail('path component "%s" in "%s" is not a list in "%s"' % (component, path, str(d)))
        !           127:                 try:
        !           128:                     d = d[idx]
        !           129:                 except IndexError:
        !           130:                     self.fail('invalid index "%s" in path "%s" in "%s"' % (idx, path, str(d)))
        !           131:         return d
        !           132: 
        !           133:     def assert_qmp(self, d, path, value):
        !           134:         '''Assert that the value for a specific path in a QMP dict matches'''
        !           135:         result = self.dictpath(d, path)
        !           136:         self.assertEqual(result, value, 'values not equal "%s" and "%s"' % (str(result), str(value)))
        !           137: 
        !           138: def notrun(reason):
        !           139:     '''Skip this test suite'''
        !           140:     # Each test in qemu-iotests has a number ("seq")
        !           141:     seq = os.path.basename(sys.argv[0])
        !           142: 
        !           143:     open('%s.notrun' % seq, 'wb').write(reason + '\n')
        !           144:     print '%s not run: %s' % (seq, reason)
        !           145:     sys.exit(0)
        !           146: 
        !           147: def main(supported_fmts=[]):
        !           148:     '''Run tests'''
        !           149: 
        !           150:     if supported_fmts and (imgfmt not in supported_fmts):
        !           151:         notrun('not suitable for this image format: %s' % imgfmt)
        !           152: 
        !           153:     # We need to filter out the time taken from the output so that qemu-iotest
        !           154:     # can reliably diff the results against master output.
        !           155:     import StringIO
        !           156:     output = StringIO.StringIO()
        !           157: 
        !           158:     class MyTestRunner(unittest.TextTestRunner):
        !           159:         def __init__(self, stream=output, descriptions=True, verbosity=1):
        !           160:             unittest.TextTestRunner.__init__(self, stream, descriptions, verbosity)
        !           161: 
        !           162:     # unittest.main() will use sys.exit() so expect a SystemExit exception
        !           163:     try:
        !           164:         unittest.main(testRunner=MyTestRunner)
        !           165:     finally:
        !           166:         sys.stderr.write(re.sub(r'Ran (\d+) test[s] in [\d.]+s', r'Ran \1 tests', output.getvalue()))

This archive runs on limited infrastructure. Preserving old code on modern bandwidth. Automated agents are requested to crawl responsibly.