# Obviously we only work with the luks image format
-iotests.verify_image_format(supported_fmts=['luks'])
-iotests.verify_platform()
+iotests.script_initialize(supported_fmts=['luks'])
# We need sudo in order to run cryptsetup to create
# dm-crypt devices. This is safe to use on any
import iotests
-iotests.verify_image_format(supported_fmts=['qcow2', 'qed', 'raw'])
-iotests.verify_platform(['linux'])
+iotests.script_initialize(supported_fmts=['qcow2', 'qed', 'raw'],
+ supported_platforms=['linux'])
with iotests.FilePath('source.img') as source_img_path, \
iotests.FilePath('dest.img') as dest_img_path, \
import iotests
-iotests.verify_image_format(supported_fmts=['qcow2'])
-iotests.verify_platform(['linux'])
+iotests.script_initialize(supported_fmts=['qcow2'],
+ supported_platforms=['linux'])
with iotests.FilePath('disk0.img') as disk0_img_path, \
iotests.FilePath('disk1.img') as disk1_img_path, \
import iotests
-iotests.verify_image_format(supported_fmts=['qcow2'])
-iotests.verify_platform(['linux'])
+iotests.script_initialize(supported_fmts=['qcow2'],
+ supported_platforms=['linux'])
with iotests.FilePath('disk0.img') as disk0_img_path, \
iotests.FilePath('disk1.img') as disk1_img_path, \
import iotests
from iotests import imgfmt
-iotests.verify_image_format(supported_fmts=['qcow2'])
+iotests.script_initialize(supported_fmts=['qcow2'])
with iotests.FilePath('t.qcow2') as disk_path, \
iotests.FilePath('t.qcow2.base') as backing_path, \
import subprocess
import re
-iotests.verify_image_format(supported_fmts=['raw'])
-iotests.verify_protocol(supported=['ssh'])
+iotests.script_initialize(
+ supported_fmts=['raw'],
+ supported_protocols=['ssh'],
+)
def filter_hash(qmsg):
def _filter(key, value):
import iotests
-iotests.verify_image_format(supported_fmts=['generic'])
+iotests.script_initialize(supported_fmts=['generic'])
with iotests.FilePath('disk.img') as disk_img_path, \
iotests.FilePath('disk-snapshot.img') as disk_snapshot_img_path, \
from iotests import qemu_img_create, qemu_io, qemu_img_verbose, qemu_nbd, \
file_path
-iotests.verify_image_format(supported_fmts=['qcow2'])
+iotests.script_initialize(supported_fmts=['qcow2'])
disk = file_path('disk')
nbd_sock = file_path('nbd-sock', base_dir=iotests.sock_dir)
import iotests
from iotests import imgfmt
-iotests.verify_image_format(supported_fmts=['luks'])
-iotests.verify_protocol(supported=['file'])
+iotests.script_initialize(
+ supported_fmts=['luks'],
+ supported_protocols=['file'],
+)
with iotests.FilePath('t.luks') as disk_path, \
iotests.VM() as vm:
import iotests
from iotests import imgfmt
-iotests.verify_image_format(supported_fmts=['vdi'])
-iotests.verify_protocol(supported=['file'])
+iotests.script_initialize(
+ supported_fmts=['vdi'],
+ supported_protocols=['file'],
+)
def blockdev_create(vm, options):
error = vm.blockdev_create(options)
import iotests
from iotests import imgfmt
-iotests.verify_image_format(supported_fmts=['parallels'])
-iotests.verify_protocol(supported=['file'])
+iotests.script_initialize(
+ supported_fmts=['parallels'],
+ supported_protocols=['file'],
+)
with iotests.FilePath('t.parallels') as disk_path, \
iotests.VM() as vm:
import iotests
from iotests import imgfmt
-iotests.verify_image_format(supported_fmts=['vhdx'])
-iotests.verify_protocol(supported=['file'])
+iotests.script_initialize(
+ supported_fmts=['vhdx'],
+ supported_protocols=['file'],
+)
with iotests.FilePath('t.vhdx') as disk_path, \
iotests.VM() as vm:
from iotests import log, qemu_img, qemu_io_silent
# Need backing file support
-iotests.verify_image_format(supported_fmts=['qcow2', 'qcow', 'qed', 'vmdk'])
-iotests.verify_platform(['linux'])
+iotests.script_initialize(supported_fmts=['qcow2', 'qcow', 'qed', 'vmdk'],
+ supported_platforms=['linux'])
log('')
log('=== Copy-on-read across nodes ===')
import iotests
from iotests import log, qemu_img, qemu_io_silent
-iotests.verify_image_format(supported_fmts=['qcow2', 'raw'])
+iotests.script_initialize(supported_fmts=['qcow2', 'raw'])
# Launches the VM, adds two null-co nodes (source and target), and
import iotests
-iotests.verify_image_format(supported_fmts=['qcow2'])
+iotests.script_initialize(supported_fmts=['qcow2'])
img_size = 4 * 1024 * 1024
import iotests
from iotests import log, qemu_img, qemu_io, qemu_io_silent
-iotests.verify_platform(['linux'])
-iotests.verify_image_format(supported_fmts=['qcow2', 'qcow', 'qed', 'vmdk',
- 'vhdx', 'raw'])
+iotests.script_initialize(
+ supported_fmts=['qcow2', 'qcow', 'qed', 'vmdk', 'vhdx', 'raw'],
+ supported_platforms=['linux'],
+)
patterns = [("0x5d", "0", "64k"),
("0xd5", "1M", "64k"),
import json
# Need backing file support (for arbitrary backing formats)
-iotests.verify_image_format(supported_fmts=['qcow2', 'qcow', 'qed'])
-iotests.verify_platform(['linux'])
+iotests.script_initialize(supported_fmts=['qcow2', 'qcow', 'qed'],
+ supported_platforms=['linux'])
# There are two variations of this test:
filter_qmp_testfiles, filter_qmp_imgfmt
# Need backing file and change-backing-file support
-iotests.verify_image_format(supported_fmts=['qcow2', 'qed'])
-iotests.verify_platform(['linux'])
+iotests.script_initialize(
+ supported_fmts=['qcow2', 'qed'],
+ supported_platforms=['linux'],
+)
def log_node_info(node):
import iotests
import os
-iotests.verify_image_format(supported_fmts=['qcow2'])
-iotests.verify_platform(['linux'])
+iotests.script_initialize(supported_fmts=['qcow2'],
+ supported_platforms=['linux'])
with iotests.FilePath('img') as img_path, \
iotests.FilePath('backing') as backing_path, \
from qemu.machine import QEMUMachine
+iotests.script_initialize(supported_fmts=['qcow2'])
+
# Note:
# This test was added to check that mirror dead-lock was fixed (see previous
# commit before this test addition).
size = 1 * 1024 * 1024 * 1024
-iotests.verify_image_format(supported_fmts=['qcow2'])
-
disk = file_path('disk')
# prepare source image
import iotests
from iotests import log
-iotests.verify_image_format(supported_fmts=['generic'])
+iotests.script_initialize(supported_fmts=['generic'])
size = 64 * 1024 * 1024
granularity = 64 * 1024
import iotests
from iotests import imgfmt
-iotests.verify_image_format(supported_fmts=['vmdk'])
+iotests.script_initialize(supported_fmts=['vmdk'])
with iotests.FilePath('t.vmdk') as disk_path, \
iotests.FilePath('t.vmdk.1') as extent1_path, \
import iotests
from iotests import log
+iotests.script_initialize()
+
virtio_scsi_device = iotests.get_virtio_scsi_device()
vm = iotests.VM()
from iotests import qemu_img_create, qemu_io, qemu_img_pipe, \
file_path, img_info_log, log, filter_qemu_io
-iotests.verify_image_format(supported_fmts=['qcow2'])
+iotests.script_initialize(supported_fmts=['qcow2'])
disk = file_path('disk')
chunk = 256 * 1024
import iotests
from iotests import log
-iotests.verify_image_format(supported_fmts=['qcow2'])
+iotests.script_initialize(supported_fmts=['qcow2'])
size = 64 * 1024 * 1024 * 1024
gran_small = 32 * 1024
gran_large = 128 * 1024
import iotests
from iotests import qemu_img_create, qemu_io, file_path, filter_qmp_testfiles
-iotests.verify_image_format(supported_fmts=['qcow2'])
+iotests.script_initialize(supported_fmts=['qcow2'])
source, target = file_path('source', 'target')
size = 5 * 1024 * 1024
import iotests
from iotests import qemu_img_create, file_path, log
-iotests.verify_image_format(supported_fmts=['qcow2'])
+iotests.script_initialize(supported_fmts=['qcow2'])
disk, top = file_path('disk', 'top')
size = 1024 * 1024
import iotests
from iotests import imgfmt
-iotests.verify_image_format(supported_fmts=['qcow2'])
+iotests.script_initialize(supported_fmts=['qcow2'])
iotests.log('Finishing a commit job with background reads')
iotests.log('============================================')
import iotests
from iotests import log
-iotests.verify_image_format(supported_fmts=['qcow2'])
+iotests.script_initialize(supported_fmts=['qcow2'])
size = 64 * 1024 * 1024
with iotests.FilePath('img0') as img0_path, \
filter_qmp_testfiles, filter_qmp_imgfmt
# Need backing file and change-backing-file support
-iotests.verify_image_format(supported_fmts=['qcow2', 'qed'])
-iotests.verify_platform(['linux'])
-
+iotests.script_initialize(
+ supported_fmts=['qcow2', 'qed'],
+ supported_platforms=['linux'],
+)
# Returns a node for blockdev-add
def node(node_name, path, backing=None, fmt=None, throttle=None):
import iotests
from iotests import qemu_img_create, file_path, log, filter_qmp_event
-iotests.verify_image_format(supported_fmts=['qcow2'])
+iotests.script_initialize(
+ supported_fmts=['qcow2']
+)
base, top = file_path('base', 'top')
size = 64 * 1024 * 3
import iotests
import os
-iotests.verify_image_format(supported_fmts=['qcow2'])
-iotests.verify_platform(['linux'])
+iotests.script_initialize(supported_fmts=['qcow2'],
+ supported_platforms=['linux'])
with iotests.FilePath('img') as img_path, \
iotests.FilePath('mig_fifo') as fifo, \
from iotests import qemu_img_create, qemu_io_silent_check, file_path, \
qemu_nbd_popen, log
-iotests.verify_image_format(supported_fmts=['qcow2'])
+iotests.script_initialize(
+ supported_fmts=['qcow2'],
+)
disk_a, disk_b, nbd_sock = file_path('disk_a', 'disk_b', 'nbd-sock')
nbd_uri = 'nbd+unix:///?socket=' + nbd_sock
import iotests
-iotests.verify_image_format(supported_fmts=['qcow2'])
-iotests.verify_platform(['linux'])
+iotests.script_initialize(supported_fmts=['qcow2'],
+ supported_platforms=['linux'])
size_short = 1 * 1024 * 1024
size_long = 2 * 1024 * 1024
import iotests
from iotests import file_path, log
+iotests.script_initialize()
+
nbd_sock, conf_file = file_path('nbd-sock', 'nbd-fault-injector.conf')
import iotests
import os
-iotests.verify_image_format(supported_fmts=['qcow2'])
-iotests.verify_protocol(supported=['file'])
-iotests.verify_platform(['linux'])
+iotests.script_initialize(
+ supported_fmts=['qcow2'],
+ supported_protocols=['file'],
+ supported_platforms=['linux'],
+)
with iotests.FilePath('base') as base_path , \
iotests.FilePath('top') as top_path, \
import iotests
# The test is unrelated to formats, restrict it to qcow2 to avoid extra runs
-iotests.verify_image_format(supported_fmts=['qcow2'])
+iotests.script_initialize(
+ supported_fmts=['qcow2'],
+)
size = 1024 * 1024
import struct
import subprocess
import sys
-from typing import (Any, Callable, Dict, Iterable, List, Optional, TypeVar)
+from typing import (Any, Callable, Dict, Iterable,
+ List, Optional, Sequence, TypeVar)
import unittest
# pylint: disable=import-error, wrong-import-position
if not_sup or (imgproto in unsupported):
notrun('not suitable for this protocol: %s' % imgproto)
-def verify_platform(supported=None, unsupported=None):
- if unsupported is not None:
- if any((sys.platform.startswith(x) for x in unsupported)):
- notrun('not suitable for this OS: %s' % sys.platform)
+def verify_platform(supported=(), unsupported=()):
+ if any((sys.platform.startswith(x) for x in unsupported)):
+ notrun('not suitable for this OS: %s' % sys.platform)
- if supported is not None:
+ if supported:
if not any((sys.platform.startswith(x) for x in supported)):
notrun('not suitable for this OS: %s' % sys.platform)
return func(*args, **kwargs)
return func_wrapper
-def execute_unittest(output, verbosity, debug):
+def execute_unittest(debug=False):
+ """Executes unittests within the calling module."""
+
+ verbosity = 2 if debug else 1
+
+ if debug:
+ output = sys.stdout
+ else:
+ # We need to filter out the time taken from the output so that
+ # qemu-iotest can reliably diff the results against master output.
+ output = io.StringIO()
+
runner = unittest.TextTestRunner(stream=output, descriptions=True,
verbosity=verbosity)
try:
# exception
unittest.main(testRunner=runner)
finally:
+ # We need to filter out the time taken from the output so that
+ # qemu-iotest can reliably diff the results against master output.
if not debug:
out = output.getvalue()
out = re.sub(r'Ran (\d+) tests? in [\d.]+s', r'Ran \1 tests', out)
sys.stderr.write(out)
-def execute_test(test_function=None,
- supported_fmts=(),
- supported_platforms=None,
- supported_cache_modes=(), supported_aio_modes=(),
- unsupported_fmts=(), supported_protocols=(),
- unsupported_protocols=()):
- """Run either unittest or script-style tests."""
+def execute_setup_common(supported_fmts: Sequence[str] = (),
+ supported_platforms: Sequence[str] = (),
+ supported_cache_modes: Sequence[str] = (),
+ supported_aio_modes: Sequence[str] = (),
+ unsupported_fmts: Sequence[str] = (),
+ supported_protocols: Sequence[str] = (),
+ unsupported_protocols: Sequence[str] = ()) -> bool:
+ """
+ Perform necessary setup for either script-style or unittest-style tests.
+
+ :return: Bool; Whether or not debug mode has been requested via the CLI.
+ """
+ # Note: Python 3.6 and pylint do not like 'Collection' so use 'Sequence'.
# We are using TEST_DIR and QEMU_DEFAULT_MACHINE as proxies to
# indicate that we're not being run via "check". There may be
sys.stderr.write('Please run this test via the "check" script\n')
sys.exit(os.EX_USAGE)
- debug = '-d' in sys.argv
- verbosity = 1
verify_image_format(supported_fmts, unsupported_fmts)
verify_protocol(supported_protocols, unsupported_protocols)
verify_platform(supported=supported_platforms)
verify_cache_mode(supported_cache_modes)
verify_aio_mode(supported_aio_modes)
+ debug = '-d' in sys.argv
if debug:
- output = sys.stdout
- verbosity = 2
sys.argv.remove('-d')
- else:
- # We need to filter out the time taken from the output so that
- # qemu-iotest can reliably diff the results against master output.
- output = io.StringIO()
-
logging.basicConfig(level=(logging.DEBUG if debug else logging.WARN))
+ return debug
+
+def execute_test(*args, test_function=None, **kwargs):
+ """Run either unittest or script-style tests."""
+
+ debug = execute_setup_common(*args, **kwargs)
if not test_function:
- execute_unittest(output, verbosity, debug)
+ execute_unittest(debug)
else:
test_function()
+# This is called from script-style iotests without a single point of entry
+def script_initialize(*args, **kwargs):
+ """Initialize script-style tests without running any tests."""
+ execute_setup_common(*args, **kwargs)
+
+# This is called from script-style iotests with a single point of entry
def script_main(test_function, *args, **kwargs):
"""Run script-style tests outside of the unittest framework"""
- execute_test(test_function, *args, **kwargs)
+ execute_test(*args, test_function=test_function, **kwargs)
+# This is called from unittest style iotests
def main(*args, **kwargs):
"""Run tests using the unittest framework"""
- execute_test(None, *args, **kwargs)
+ execute_test(*args, **kwargs)
projects / qemu.git / commitdiff