[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[Qemu-block] [PATCH 2/6] iotests.py: move the generic QMPTestCase to qte
From: |
Philippe Mathieu-Daudé |
Subject: |
[Qemu-block] [PATCH 2/6] iotests.py: move the generic QMPTestCase to qtest.py |
Date: |
Wed, 13 Dec 2017 18:35:53 -0300 |
All those methods are generic and can be reused.
Add a few 'from qtest import ...' to keep the patch short.
Signed-off-by: Philippe Mathieu-Daudé <address@hidden>
---
scripts/qtest.py | 102 ++++++++++++++++++++++++++++++++++++++++++
tests/qemu-iotests/iotests.py | 86 +++--------------------------------
2 files changed, 107 insertions(+), 81 deletions(-)
diff --git a/scripts/qtest.py b/scripts/qtest.py
index df0daf26ca..78db8d02c0 100644
--- a/scripts/qtest.py
+++ b/scripts/qtest.py
@@ -12,10 +12,33 @@
#
import socket
+import sys
import os
import qemu
+import re
+import unittest
+import logging
+qemu_prog = os.environ.get('QEMU_PROG', 'qemu')
+qemu_opts = os.environ.get('QEMU_OPTIONS', '').strip().split(' ')
+
+test_dir = os.environ.get('TEST_DIR')
+output_dir = os.environ.get('OUTPUT_DIR', '.')
+qemu_default_machine = os.environ.get('QEMU_DEFAULT_MACHINE')
+
+socket_scm_helper = os.environ.get('SOCKET_SCM_HELPER', 'socket_scm_helper')
+debug = False
+
+
+def filter_qmp_event(event):
+ '''Filter a QMP event dict'''
+ event = dict(event)
+ if 'timestamp' in event:
+ event['timestamp']['seconds'] = 'SECS'
+ event['timestamp']['microseconds'] = 'USECS'
+ return event
+
class QEMUQtestProtocol(object):
def __init__(self, address, server=False):
"""
@@ -107,3 +130,82 @@ class QEMUQtestMachine(qemu.QEMUMachine):
def qtest(self, cmd):
'''Send a qtest command to guest'''
return self._qtest.cmd(cmd)
+
+
+def createQtestMachine(path_suffix=''):
+ name = "qemu%s-%d" % (path_suffix, os.getpid())
+ return QEMUQtestMachine(qemu_prog, qemu_opts, name=name,
+ test_dir=test_dir,
+ socket_scm_helper=socket_scm_helper)
+
+class QMPTestCase(unittest.TestCase):
+ '''Abstract base class for QMP test cases'''
+
+ index_re = re.compile(r'([^\[]+)\[([^\]]+)\]')
+
+ def dictpath(self, d, path):
+ '''Traverse a path in a nested dict'''
+ for component in path.split('/'):
+ m = QMPTestCase.index_re.match(component)
+ if m:
+ component, idx = m.groups()
+ idx = int(idx)
+
+ if not isinstance(d, dict) or component not in d:
+ self.fail('failed path traversal for "%s" in "%s"' % (path,
+ str(d)))
+ d = d[component]
+
+ if m:
+ if not isinstance(d, list):
+ self.fail(('path component "%s" in "%s" is not a list ' +
+ 'in "%s"') % (component, path, str(d)))
+ try:
+ d = d[idx]
+ except IndexError:
+ self.fail(('invalid index "%s" in path "%s" ' +
+ 'in "%s"') % (idx, path, str(d)))
+ return d
+
+ def flatten_qmp_object(self, obj, output=None, basestr=''):
+ if output is None:
+ output = dict()
+ if isinstance(obj, list):
+ for i in range(len(obj)):
+ self.flatten_qmp_object(obj[i], output, basestr + str(i) + '.')
+ elif isinstance(obj, dict):
+ for key in obj:
+ self.flatten_qmp_object(obj[key], output, basestr + key + '.')
+ else:
+ output[basestr[:-1]] = obj # Strip trailing '.'
+ return output
+
+ def qmp_to_opts(self, obj):
+ obj = self.flatten_qmp_object(obj)
+ output_list = list()
+ for key in obj:
+ output_list += [key + '=' + obj[key]]
+ return ','.join(output_list)
+
+ def assert_qmp_absent(self, d, path):
+ try:
+ result = self.dictpath(d, path)
+ except AssertionError:
+ return
+ self.fail('path "%s" has value "%s"' % (path, str(result)))
+
+ def assert_qmp(self, d, path, value):
+ '''Assert that the value for a specific path in a QMP dict matches'''
+ result = self.dictpath(d, path)
+ self.assertEqual(result, value, ('values not equal "%s" ' +
+ 'and "%s"') % (str(result), str(value)))
+
+
+def notrun(reason):
+ '''Skip this test suite'''
+ print '%s not run: %s' % (seq, reason)
+ sys.exit(0)
+
+def verify_platform(supported_oses=['linux']):
+ if True not in [sys.platform.startswith(x) for x in supported_oses]:
+ notrun('not suitable for this OS: %s' % sys.platform)
diff --git a/tests/qemu-iotests/iotests.py b/tests/qemu-iotests/iotests.py
index e1f892c46f..cb3d22855b 100644
--- a/tests/qemu-iotests/iotests.py
+++ b/tests/qemu-iotests/iotests.py
@@ -20,7 +20,6 @@ import errno
import os
import re
import subprocess
-import string
import unittest
import sys
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'scripts'))
@@ -30,6 +29,10 @@ import json
import signal
import logging
+# use the following imports from qtest instead of iotests.
+from qtest import qemu_prog, qemu_opts, test_dir, output_dir
+from qtest import qemu_default_machine, socket_scm_helper
+from qtest import filter_qmp_event, verify_platform
# This will not work if arguments contain spaces but is necessary if we
# want to support the override options that ./check supports.
@@ -45,18 +48,10 @@ qemu_nbd_args = [os.environ.get('QEMU_NBD_PROG',
'qemu-nbd')]
if os.environ.get('QEMU_NBD_OPTIONS'):
qemu_nbd_args += os.environ['QEMU_NBD_OPTIONS'].strip().split(' ')
-qemu_prog = os.environ.get('QEMU_PROG', 'qemu')
-qemu_opts = os.environ.get('QEMU_OPTIONS', '').strip().split(' ')
-
imgfmt = os.environ.get('IMGFMT', 'raw')
imgproto = os.environ.get('IMGPROTO', 'file')
-test_dir = os.environ.get('TEST_DIR')
-output_dir = os.environ.get('OUTPUT_DIR', '.')
cachemode = os.environ.get('CACHEMODE')
-qemu_default_machine = os.environ.get('QEMU_DEFAULT_MACHINE')
-socket_scm_helper = os.environ.get('SOCKET_SCM_HELPER', 'socket_scm_helper')
-debug = False
def qemu_img(*args):
'''Run qemu-img and return the exit code'''
@@ -134,14 +129,6 @@ chown_re = re.compile(r"chown [0-9]+:[0-9]+")
def filter_chown(msg):
return chown_re.sub("chown UID:GID", msg)
-def filter_qmp_event(event):
- '''Filter a QMP event dict'''
- event = dict(event)
- if 'timestamp' in event:
- event['timestamp']['seconds'] = 'SECS'
- event['timestamp']['microseconds'] = 'USECS'
- return event
-
def log(msg, filters=[]):
for flt in filters:
msg = flt(msg)
@@ -257,66 +244,7 @@ class VM(qtest.QEMUQtestMachine):
command_line='qemu-io %s "%s"' % (drive, cmd))
-index_re = re.compile(r'([^\[]+)\[([^\]]+)\]')
-
-class QMPTestCase(unittest.TestCase):
- '''Abstract base class for QMP test cases'''
-
- def dictpath(self, d, path):
- '''Traverse a path in a nested dict'''
- for component in path.split('/'):
- m = index_re.match(component)
- if m:
- component, idx = m.groups()
- idx = int(idx)
-
- if not isinstance(d, dict) or component not in d:
- self.fail('failed path traversal for "%s" in "%s"' % (path,
str(d)))
- d = d[component]
-
- if m:
- if not isinstance(d, list):
- self.fail('path component "%s" in "%s" is not a list in
"%s"' % (component, path, str(d)))
- try:
- d = d[idx]
- except IndexError:
- self.fail('invalid index "%s" in path "%s" in "%s"' %
(idx, path, str(d)))
- return d
-
- def flatten_qmp_object(self, obj, output=None, basestr=''):
- if output is None:
- output = dict()
- if isinstance(obj, list):
- for i in range(len(obj)):
- self.flatten_qmp_object(obj[i], output, basestr + str(i) + '.')
- elif isinstance(obj, dict):
- for key in obj:
- self.flatten_qmp_object(obj[key], output, basestr + key + '.')
- else:
- output[basestr[:-1]] = obj # Strip trailing '.'
- return output
-
- def qmp_to_opts(self, obj):
- obj = self.flatten_qmp_object(obj)
- output_list = list()
- for key in obj:
- output_list += [key + '=' + obj[key]]
- return ','.join(output_list)
-
- def assert_qmp_absent(self, d, path):
- try:
- result = self.dictpath(d, path)
- except AssertionError:
- return
- self.fail('path "%s" has value "%s"' % (path, str(result)))
-
- def assert_qmp(self, d, path, value):
- '''Assert that the value for a specific path in a QMP dict matches'''
- result = self.dictpath(d, path)
- self.assertEqual(result, value, 'values not equal "%s" and "%s"' %
(str(result), str(value)))
-
-
-class BlockQMPTestCase(QMPTestCase):
+class BlockQMPTestCase(qtest.QMPTestCase):
'''Abstract base class for Block QMP test cases'''
def assert_no_active_block_jobs(self):
@@ -430,10 +358,6 @@ def verify_image_format(supported_fmts=[],
unsupported_fmts=[]):
if unsupported_fmts and (imgfmt in unsupported_fmts):
notrun('not suitable for this image format: %s' % imgfmt)
-def verify_platform(supported_oses=['linux']):
- if True not in [sys.platform.startswith(x) for x in supported_oses]:
- notrun('not suitable for this OS: %s' % sys.platform)
-
def supports_quorum():
return 'quorum' in qemu_img_pipe('--help')
--
2.15.1
- [Qemu-block] [PATCH 0/6] QTests: use Python to run complex tests, Philippe Mathieu-Daudé, 2017/12/13
- [Qemu-block] [PATCH 2/6] iotests.py: move the generic QMPTestCase to qtest.py,
Philippe Mathieu-Daudé <=
- [Qemu-block] [PATCH 1/6] iotests.py: split BlockQMPTestCase class of QMPTestCase, Philippe Mathieu-Daudé, 2017/12/13
- Re: [Qemu-block] [PATCH 0/6] QTests: use Python to run complex tests, Stefan Hajnoczi, 2017/12/14
- Re: [Qemu-block] [PATCH 0/6] QTests: use Python to run complex tests, Paolo Bonzini, 2017/12/14
- Re: [Qemu-block] [PATCH 0/6] QTests: use Python to run complex tests, Nir Soffer, 2017/12/14
- Re: [Qemu-block] [PATCH 0/6] QTests: use Python to run complex tests, Philippe Mathieu-Daudé, 2017/12/14
- Re: [Qemu-block] [Qemu-devel] [PATCH 0/6] QTests: use Python to run complex tests, Markus Armbruster, 2017/12/14
- Re: [Qemu-block] [Qemu-devel] [PATCH 0/6] QTests: use Python to run complex tests, Paolo Bonzini, 2017/12/14
- Re: [Qemu-block] [Qemu-devel] [PATCH 0/6] QTests: use Python to run complex tests, Peter Maydell, 2017/12/14
- Re: [Qemu-block] [Qemu-devel] [PATCH 0/6] QTests: use Python to run complex tests, Paolo Bonzini, 2017/12/14
Re: [Qemu-block] [PATCH 0/6] QTests: use Python to run complex tests, Philippe Mathieu-Daudé, 2017/12/14