xref: /openbmc/qemu/tests/qemu-iotests/iotests.py (revision 1770b2f2)
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 argparse
20import atexit
21import bz2
22from collections import OrderedDict
23import faulthandler
24import json
25import logging
26import os
27import re
28import shutil
29import signal
30import struct
31import subprocess
32import sys
33import time
34from typing import (Any, Callable, Dict, Iterable, Iterator,
35                    List, Optional, Sequence, TextIO, Tuple, Type, TypeVar)
36import unittest
37
38from contextlib import contextmanager
39
40from qemu.machine import qtest
41from qemu.qmp.legacy import QMPMessage, QEMUMonitorProtocol
42from qemu.utils import VerboseProcessError
43
44# Use this logger for logging messages directly from the iotests module
45logger = logging.getLogger('qemu.iotests')
46logger.addHandler(logging.NullHandler())
47
48# Use this logger for messages that ought to be used for diff output.
49test_logger = logging.getLogger('qemu.iotests.diff_io')
50
51
52faulthandler.enable()
53
54# This will not work if arguments contain spaces but is necessary if we
55# want to support the override options that ./check supports.
56qemu_img_args = [os.environ.get('QEMU_IMG_PROG', 'qemu-img')]
57if os.environ.get('QEMU_IMG_OPTIONS'):
58    qemu_img_args += os.environ['QEMU_IMG_OPTIONS'].strip().split(' ')
59
60qemu_io_args = [os.environ.get('QEMU_IO_PROG', 'qemu-io')]
61if os.environ.get('QEMU_IO_OPTIONS'):
62    qemu_io_args += os.environ['QEMU_IO_OPTIONS'].strip().split(' ')
63
64qemu_io_args_no_fmt = [os.environ.get('QEMU_IO_PROG', 'qemu-io')]
65if os.environ.get('QEMU_IO_OPTIONS_NO_FMT'):
66    qemu_io_args_no_fmt += \
67        os.environ['QEMU_IO_OPTIONS_NO_FMT'].strip().split(' ')
68
69qemu_nbd_prog = os.environ.get('QEMU_NBD_PROG', 'qemu-nbd')
70qemu_nbd_args = [qemu_nbd_prog]
71if os.environ.get('QEMU_NBD_OPTIONS'):
72    qemu_nbd_args += os.environ['QEMU_NBD_OPTIONS'].strip().split(' ')
73
74qemu_prog = os.environ.get('QEMU_PROG', 'qemu')
75qemu_opts = os.environ.get('QEMU_OPTIONS', '').strip().split(' ')
76
77qsd_prog = os.environ.get('QSD_PROG', 'qemu-storage-daemon')
78
79gdb_qemu_env = os.environ.get('GDB_OPTIONS')
80qemu_gdb = []
81if gdb_qemu_env:
82    qemu_gdb = ['gdbserver'] + gdb_qemu_env.strip().split(' ')
83
84qemu_print = os.environ.get('PRINT_QEMU', False)
85
86imgfmt = os.environ.get('IMGFMT', 'raw')
87imgproto = os.environ.get('IMGPROTO', 'file')
88
89try:
90    test_dir = os.environ['TEST_DIR']
91    sock_dir = os.environ['SOCK_DIR']
92    cachemode = os.environ['CACHEMODE']
93    aiomode = os.environ['AIOMODE']
94    qemu_default_machine = os.environ['QEMU_DEFAULT_MACHINE']
95except KeyError:
96    # We are using these variables as proxies to indicate that we're
97    # not being run via "check". There may be other things set up by
98    # "check" that individual test cases rely on.
99    sys.stderr.write('Please run this test via the "check" script\n')
100    sys.exit(os.EX_USAGE)
101
102qemu_valgrind = []
103if os.environ.get('VALGRIND_QEMU') == "y" and \
104    os.environ.get('NO_VALGRIND') != "y":
105    valgrind_logfile = "--log-file=" + test_dir
106    # %p allows to put the valgrind process PID, since
107    # we don't know it a priori (subprocess.Popen is
108    # not yet invoked)
109    valgrind_logfile += "/%p.valgrind"
110
111    qemu_valgrind = ['valgrind', valgrind_logfile, '--error-exitcode=99']
112
113luks_default_secret_object = 'secret,id=keysec0,data=' + \
114                             os.environ.get('IMGKEYSECRET', '')
115luks_default_key_secret_opt = 'key-secret=keysec0'
116
117sample_img_dir = os.environ['SAMPLE_IMG_DIR']
118
119
120@contextmanager
121def change_log_level(
122        logger_name: str, level: int = logging.CRITICAL) -> Iterator[None]:
123    """
124    Utility function for temporarily changing the log level of a logger.
125
126    This can be used to silence errors that are expected or uninteresting.
127    """
128    _logger = logging.getLogger(logger_name)
129    current_level = _logger.level
130    _logger.setLevel(level)
131
132    try:
133        yield
134    finally:
135        _logger.setLevel(current_level)
136
137
138def unarchive_sample_image(sample, fname):
139    sample_fname = os.path.join(sample_img_dir, sample + '.bz2')
140    with bz2.open(sample_fname) as f_in, open(fname, 'wb') as f_out:
141        shutil.copyfileobj(f_in, f_out)
142
143
144def qemu_tool_popen(args: Sequence[str],
145                    connect_stderr: bool = True) -> 'subprocess.Popen[str]':
146    stderr = subprocess.STDOUT if connect_stderr else None
147    # pylint: disable=consider-using-with
148    return subprocess.Popen(args,
149                            stdout=subprocess.PIPE,
150                            stderr=stderr,
151                            universal_newlines=True)
152
153
154def qemu_tool_pipe_and_status(tool: str, args: Sequence[str],
155                              connect_stderr: bool = True,
156                              drop_successful_output: bool = False) \
157        -> Tuple[str, int]:
158    """
159    Run a tool and return both its output and its exit code
160    """
161    with qemu_tool_popen(args, connect_stderr) as subp:
162        output = subp.communicate()[0]
163        if subp.returncode < 0:
164            cmd = ' '.join(args)
165            sys.stderr.write(f'{tool} received signal \
166                               {-subp.returncode}: {cmd}\n')
167        if drop_successful_output and subp.returncode == 0:
168            output = ''
169        return (output, subp.returncode)
170
171def qemu_img_create_prepare_args(args: List[str]) -> List[str]:
172    if not args or args[0] != 'create':
173        return list(args)
174    args = args[1:]
175
176    p = argparse.ArgumentParser(allow_abbrev=False)
177    # -o option may be specified several times
178    p.add_argument('-o', action='append', default=[])
179    p.add_argument('-f')
180    parsed, remaining = p.parse_known_args(args)
181
182    opts_list = parsed.o
183
184    result = ['create']
185    if parsed.f is not None:
186        result += ['-f', parsed.f]
187
188    # IMGOPTS most probably contain options specific for the selected format,
189    # like extended_l2 or compression_type for qcow2. Test may want to create
190    # additional images in other formats that doesn't support these options.
191    # So, use IMGOPTS only for images created in imgfmt format.
192    imgopts = os.environ.get('IMGOPTS')
193    if imgopts and parsed.f == imgfmt:
194        opts_list.insert(0, imgopts)
195
196    # default luks support
197    if parsed.f == 'luks' and \
198            all('key-secret' not in opts for opts in opts_list):
199        result += ['--object', luks_default_secret_object]
200        opts_list.append(luks_default_key_secret_opt)
201
202    for opts in opts_list:
203        result += ['-o', opts]
204
205    result += remaining
206
207    return result
208
209
210def qemu_tool(*args: str, check: bool = True, combine_stdio: bool = True
211              ) -> 'subprocess.CompletedProcess[str]':
212    """
213    Run a qemu tool and return its status code and console output.
214
215    :param args: full command line to run.
216    :param check: Enforce a return code of zero.
217    :param combine_stdio: set to False to keep stdout/stderr separated.
218
219    :raise VerboseProcessError:
220        When the return code is negative, or on any non-zero exit code
221        when 'check=True' was provided (the default). This exception has
222        'stdout', 'stderr', and 'returncode' properties that may be
223        inspected to show greater detail. If this exception is not
224        handled, the command-line, return code, and all console output
225        will be included at the bottom of the stack trace.
226
227    :return:
228        a CompletedProcess. This object has args, returncode, and stdout
229        properties. If streams are not combined, it will also have a
230        stderr property.
231    """
232    subp = subprocess.run(
233        args,
234        stdout=subprocess.PIPE,
235        stderr=subprocess.STDOUT if combine_stdio else subprocess.PIPE,
236        universal_newlines=True,
237        check=False
238    )
239
240    if check and subp.returncode or (subp.returncode < 0):
241        raise VerboseProcessError(
242            subp.returncode, args,
243            output=subp.stdout,
244            stderr=subp.stderr,
245        )
246
247    return subp
248
249
250def qemu_img(*args: str, check: bool = True, combine_stdio: bool = True
251             ) -> 'subprocess.CompletedProcess[str]':
252    """
253    Run QEMU_IMG_PROG and return its status code and console output.
254
255    This function always prepends QEMU_IMG_OPTIONS and may further alter
256    the args for 'create' commands.
257
258    See `qemu_tool()` for greater detail.
259    """
260    full_args = qemu_img_args + qemu_img_create_prepare_args(list(args))
261    return qemu_tool(*full_args, check=check, combine_stdio=combine_stdio)
262
263
264def ordered_qmp(qmsg, conv_keys=True):
265    # Dictionaries are not ordered prior to 3.6, therefore:
266    if isinstance(qmsg, list):
267        return [ordered_qmp(atom) for atom in qmsg]
268    if isinstance(qmsg, dict):
269        od = OrderedDict()
270        for k, v in sorted(qmsg.items()):
271            if conv_keys:
272                k = k.replace('_', '-')
273            od[k] = ordered_qmp(v, conv_keys=False)
274        return od
275    return qmsg
276
277def qemu_img_create(*args: str) -> 'subprocess.CompletedProcess[str]':
278    return qemu_img('create', *args)
279
280def qemu_img_json(*args: str) -> Any:
281    """
282    Run qemu-img and return its output as deserialized JSON.
283
284    :raise CalledProcessError:
285        When qemu-img crashes, or returns a non-zero exit code without
286        producing a valid JSON document to stdout.
287    :raise JSONDecoderError:
288        When qemu-img returns 0, but failed to produce a valid JSON document.
289
290    :return: A deserialized JSON object; probably a dict[str, Any].
291    """
292    try:
293        res = qemu_img(*args, combine_stdio=False)
294    except subprocess.CalledProcessError as exc:
295        # Terminated due to signal. Don't bother.
296        if exc.returncode < 0:
297            raise
298
299        # Commands like 'check' can return failure (exit codes 2 and 3)
300        # to indicate command completion, but with errors found. For
301        # multi-command flexibility, ignore the exact error codes and
302        # *try* to load JSON.
303        try:
304            return json.loads(exc.stdout)
305        except json.JSONDecodeError:
306            # Nope. This thing is toast. Raise the /process/ error.
307            pass
308        raise
309
310    return json.loads(res.stdout)
311
312def qemu_img_measure(*args: str) -> Any:
313    return qemu_img_json("measure", "--output", "json", *args)
314
315def qemu_img_check(*args: str) -> Any:
316    return qemu_img_json("check", "--output", "json", *args)
317
318def qemu_img_info(*args: str) -> Any:
319    return qemu_img_json('info', "--output", "json", *args)
320
321def qemu_img_map(*args: str) -> Any:
322    return qemu_img_json('map', "--output", "json", *args)
323
324def qemu_img_log(*args: str, check: bool = True
325                 ) -> 'subprocess.CompletedProcess[str]':
326    result = qemu_img(*args, check=check)
327    log(result.stdout, filters=[filter_testfiles])
328    return result
329
330def img_info_log(filename: str, filter_path: Optional[str] = None,
331                 use_image_opts: bool = False, extra_args: Sequence[str] = (),
332                 check: bool = True, drop_child_info: bool = True,
333                 ) -> None:
334    args = ['info']
335    if use_image_opts:
336        args.append('--image-opts')
337    else:
338        args += ['-f', imgfmt]
339    args += extra_args
340    args.append(filename)
341
342    output = qemu_img(*args, check=check).stdout
343    if not filter_path:
344        filter_path = filename
345    log(filter_img_info(output, filter_path, drop_child_info))
346
347def qemu_io_wrap_args(args: Sequence[str]) -> List[str]:
348    if '-f' in args or '--image-opts' in args:
349        return qemu_io_args_no_fmt + list(args)
350    else:
351        return qemu_io_args + list(args)
352
353def qemu_io_popen(*args):
354    return qemu_tool_popen(qemu_io_wrap_args(args))
355
356def qemu_io(*args: str, check: bool = True, combine_stdio: bool = True
357            ) -> 'subprocess.CompletedProcess[str]':
358    """
359    Run QEMU_IO_PROG and return the status code and console output.
360
361    This function always prepends either QEMU_IO_OPTIONS or
362    QEMU_IO_OPTIONS_NO_FMT.
363    """
364    return qemu_tool(*qemu_io_wrap_args(args),
365                     check=check, combine_stdio=combine_stdio)
366
367def qemu_io_log(*args: str, check: bool = True
368                ) -> 'subprocess.CompletedProcess[str]':
369    result = qemu_io(*args, check=check)
370    log(result.stdout, filters=[filter_testfiles, filter_qemu_io])
371    return result
372
373class QemuIoInteractive:
374    def __init__(self, *args):
375        self.args = qemu_io_wrap_args(args)
376        # We need to keep the Popen objext around, and not
377        # close it immediately. Therefore, disable the pylint check:
378        # pylint: disable=consider-using-with
379        self._p = subprocess.Popen(self.args, stdin=subprocess.PIPE,
380                                   stdout=subprocess.PIPE,
381                                   stderr=subprocess.STDOUT,
382                                   universal_newlines=True)
383        out = self._p.stdout.read(9)
384        if out != 'qemu-io> ':
385            # Most probably qemu-io just failed to start.
386            # Let's collect the whole output and exit.
387            out += self._p.stdout.read()
388            self._p.wait(timeout=1)
389            raise ValueError(out)
390
391    def close(self):
392        self._p.communicate('q\n')
393
394    def _read_output(self):
395        pattern = 'qemu-io> '
396        n = len(pattern)
397        pos = 0
398        s = []
399        while pos != n:
400            c = self._p.stdout.read(1)
401            # check unexpected EOF
402            assert c != ''
403            s.append(c)
404            if c == pattern[pos]:
405                pos += 1
406            else:
407                pos = 0
408
409        return ''.join(s[:-n])
410
411    def cmd(self, cmd):
412        # quit command is in close(), '\n' is added automatically
413        assert '\n' not in cmd
414        cmd = cmd.strip()
415        assert cmd not in ('q', 'quit')
416        self._p.stdin.write(cmd + '\n')
417        self._p.stdin.flush()
418        return self._read_output()
419
420
421class QemuStorageDaemon:
422    _qmp: Optional[QEMUMonitorProtocol] = None
423    _qmpsock: Optional[str] = None
424    # Python < 3.8 would complain if this type were not a string literal
425    # (importing `annotations` from `__future__` would work; but not on <= 3.6)
426    _p: 'Optional[subprocess.Popen[bytes]]' = None
427
428    def __init__(self, *args: str, instance_id: str = 'a', qmp: bool = False):
429        assert '--pidfile' not in args
430        self.pidfile = os.path.join(test_dir, f'qsd-{instance_id}-pid')
431        all_args = [qsd_prog] + list(args) + ['--pidfile', self.pidfile]
432
433        if qmp:
434            self._qmpsock = os.path.join(sock_dir, f'qsd-{instance_id}.sock')
435            all_args += ['--chardev',
436                         f'socket,id=qmp-sock,path={self._qmpsock}',
437                         '--monitor', 'qmp-sock']
438
439            self._qmp = QEMUMonitorProtocol(self._qmpsock, server=True)
440
441        # Cannot use with here, we want the subprocess to stay around
442        # pylint: disable=consider-using-with
443        self._p = subprocess.Popen(all_args)
444        if self._qmp is not None:
445            self._qmp.accept()
446        while not os.path.exists(self.pidfile):
447            if self._p.poll() is not None:
448                cmd = ' '.join(all_args)
449                raise RuntimeError(
450                    'qemu-storage-daemon terminated with exit code ' +
451                    f'{self._p.returncode}: {cmd}')
452
453            time.sleep(0.01)
454
455        with open(self.pidfile, encoding='utf-8') as f:
456            self._pid = int(f.read().strip())
457
458        assert self._pid == self._p.pid
459
460    def qmp(self, cmd: str, args: Optional[Dict[str, object]] = None) \
461            -> QMPMessage:
462        assert self._qmp is not None
463        return self._qmp.cmd(cmd, args)
464
465    def stop(self, kill_signal=15):
466        self._p.send_signal(kill_signal)
467        self._p.wait()
468        self._p = None
469
470        if self._qmp:
471            self._qmp.close()
472
473        if self._qmpsock is not None:
474            try:
475                os.remove(self._qmpsock)
476            except OSError:
477                pass
478        try:
479            os.remove(self.pidfile)
480        except OSError:
481            pass
482
483    def __del__(self):
484        if self._p is not None:
485            self.stop(kill_signal=9)
486
487
488def qemu_nbd(*args):
489    '''Run qemu-nbd in daemon mode and return the parent's exit code'''
490    return subprocess.call(qemu_nbd_args + ['--fork'] + list(args))
491
492def qemu_nbd_early_pipe(*args: str) -> Tuple[int, str]:
493    '''Run qemu-nbd in daemon mode and return both the parent's exit code
494       and its output in case of an error'''
495    full_args = qemu_nbd_args + ['--fork'] + list(args)
496    output, returncode = qemu_tool_pipe_and_status('qemu-nbd', full_args,
497                                                   connect_stderr=False)
498    return returncode, output if returncode else ''
499
500def qemu_nbd_list_log(*args: str) -> str:
501    '''Run qemu-nbd to list remote exports'''
502    full_args = [qemu_nbd_prog, '-L'] + list(args)
503    output, _ = qemu_tool_pipe_and_status('qemu-nbd', full_args)
504    log(output, filters=[filter_testfiles, filter_nbd_exports])
505    return output
506
507@contextmanager
508def qemu_nbd_popen(*args):
509    '''Context manager running qemu-nbd within the context'''
510    pid_file = file_path("qemu_nbd_popen-nbd-pid-file")
511
512    assert not os.path.exists(pid_file)
513
514    cmd = list(qemu_nbd_args)
515    cmd.extend(('--persistent', '--pid-file', pid_file))
516    cmd.extend(args)
517
518    log('Start NBD server')
519    with subprocess.Popen(cmd) as p:
520        try:
521            while not os.path.exists(pid_file):
522                if p.poll() is not None:
523                    raise RuntimeError(
524                        "qemu-nbd terminated with exit code {}: {}"
525                        .format(p.returncode, ' '.join(cmd)))
526
527                time.sleep(0.01)
528            yield
529        finally:
530            if os.path.exists(pid_file):
531                os.remove(pid_file)
532            log('Kill NBD server')
533            p.kill()
534            p.wait()
535
536def compare_images(img1: str, img2: str,
537                   fmt1: str = imgfmt, fmt2: str = imgfmt) -> bool:
538    """
539    Compare two images with QEMU_IMG; return True if they are identical.
540
541    :raise CalledProcessError:
542        when qemu-img crashes or returns a status code of anything other
543        than 0 (identical) or 1 (different).
544    """
545    try:
546        qemu_img('compare', '-f', fmt1, '-F', fmt2, img1, img2)
547        return True
548    except subprocess.CalledProcessError as exc:
549        if exc.returncode == 1:
550            return False
551        raise
552
553def create_image(name, size):
554    '''Create a fully-allocated raw image with sector markers'''
555    with open(name, 'wb') as file:
556        i = 0
557        while i < size:
558            sector = struct.pack('>l504xl', i // 512, i // 512)
559            file.write(sector)
560            i = i + 512
561
562def image_size(img: str) -> int:
563    """Return image's virtual size"""
564    value = qemu_img_info('-f', imgfmt, img)['virtual-size']
565    if not isinstance(value, int):
566        type_name = type(value).__name__
567        raise TypeError("Expected 'int' for 'virtual-size', "
568                        f"got '{value}' of type '{type_name}'")
569    return value
570
571def is_str(val):
572    return isinstance(val, str)
573
574test_dir_re = re.compile(r"%s" % test_dir)
575def filter_test_dir(msg):
576    return test_dir_re.sub("TEST_DIR", msg)
577
578win32_re = re.compile(r"\r")
579def filter_win32(msg):
580    return win32_re.sub("", msg)
581
582qemu_io_re = re.compile(r"[0-9]* ops; [0-9\/:. sec]* "
583                        r"\([0-9\/.inf]* [EPTGMKiBbytes]*\/sec "
584                        r"and [0-9\/.inf]* ops\/sec\)")
585def filter_qemu_io(msg):
586    msg = filter_win32(msg)
587    return qemu_io_re.sub("X ops; XX:XX:XX.X "
588                          "(XXX YYY/sec and XXX ops/sec)", msg)
589
590chown_re = re.compile(r"chown [0-9]+:[0-9]+")
591def filter_chown(msg):
592    return chown_re.sub("chown UID:GID", msg)
593
594def filter_qmp_event(event):
595    '''Filter a QMP event dict'''
596    event = dict(event)
597    if 'timestamp' in event:
598        event['timestamp']['seconds'] = 'SECS'
599        event['timestamp']['microseconds'] = 'USECS'
600    return event
601
602def filter_qmp(qmsg, filter_fn):
603    '''Given a string filter, filter a QMP object's values.
604    filter_fn takes a (key, value) pair.'''
605    # Iterate through either lists or dicts;
606    if isinstance(qmsg, list):
607        items = enumerate(qmsg)
608    elif isinstance(qmsg, dict):
609        items = qmsg.items()
610    else:
611        return filter_fn(None, qmsg)
612
613    for k, v in items:
614        if isinstance(v, (dict, list)):
615            qmsg[k] = filter_qmp(v, filter_fn)
616        else:
617            qmsg[k] = filter_fn(k, v)
618    return qmsg
619
620def filter_testfiles(msg):
621    pref1 = os.path.join(test_dir, "%s-" % (os.getpid()))
622    pref2 = os.path.join(sock_dir, "%s-" % (os.getpid()))
623    return msg.replace(pref1, 'TEST_DIR/PID-').replace(pref2, 'SOCK_DIR/PID-')
624
625def filter_qmp_testfiles(qmsg):
626    def _filter(_key, value):
627        if is_str(value):
628            return filter_testfiles(value)
629        return value
630    return filter_qmp(qmsg, _filter)
631
632def filter_virtio_scsi(output: str) -> str:
633    return re.sub(r'(virtio-scsi)-(ccw|pci)', r'\1', output)
634
635def filter_qmp_virtio_scsi(qmsg):
636    def _filter(_key, value):
637        if is_str(value):
638            return filter_virtio_scsi(value)
639        return value
640    return filter_qmp(qmsg, _filter)
641
642def filter_generated_node_ids(msg):
643    return re.sub("#block[0-9]+", "NODE_NAME", msg)
644
645def filter_img_info(output: str, filename: str,
646                    drop_child_info: bool = True) -> str:
647    lines = []
648    drop_indented = False
649    for line in output.split('\n'):
650        if 'disk size' in line or 'actual-size' in line:
651            continue
652
653        # Drop child node info
654        if drop_indented:
655            if line.startswith(' '):
656                continue
657            drop_indented = False
658        if drop_child_info and "Child node '/" in line:
659            drop_indented = True
660            continue
661
662        line = line.replace(filename, 'TEST_IMG')
663        line = filter_testfiles(line)
664        line = line.replace(imgfmt, 'IMGFMT')
665        line = re.sub('iters: [0-9]+', 'iters: XXX', line)
666        line = re.sub('uuid: [-a-f0-9]+',
667                      'uuid: XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX',
668                      line)
669        line = re.sub('cid: [0-9]+', 'cid: XXXXXXXXXX', line)
670        line = re.sub('(compression type: )(zlib|zstd)', r'\1COMPRESSION_TYPE',
671                      line)
672        lines.append(line)
673    return '\n'.join(lines)
674
675def filter_imgfmt(msg):
676    return msg.replace(imgfmt, 'IMGFMT')
677
678def filter_qmp_imgfmt(qmsg):
679    def _filter(_key, value):
680        if is_str(value):
681            return filter_imgfmt(value)
682        return value
683    return filter_qmp(qmsg, _filter)
684
685def filter_nbd_exports(output: str) -> str:
686    return re.sub(r'((min|opt|max) block): [0-9]+', r'\1: XXX', output)
687
688
689Msg = TypeVar('Msg', Dict[str, Any], List[Any], str)
690
691def log(msg: Msg,
692        filters: Iterable[Callable[[Msg], Msg]] = (),
693        indent: Optional[int] = None) -> None:
694    """
695    Logs either a string message or a JSON serializable message (like QMP).
696    If indent is provided, JSON serializable messages are pretty-printed.
697    """
698    for flt in filters:
699        msg = flt(msg)
700    if isinstance(msg, (dict, list)):
701        # Don't sort if it's already sorted
702        do_sort = not isinstance(msg, OrderedDict)
703        test_logger.info(json.dumps(msg, sort_keys=do_sort, indent=indent))
704    else:
705        test_logger.info(msg)
706
707class Timeout:
708    def __init__(self, seconds, errmsg="Timeout"):
709        self.seconds = seconds
710        self.errmsg = errmsg
711    def __enter__(self):
712        if qemu_gdb or qemu_valgrind:
713            return self
714        signal.signal(signal.SIGALRM, self.timeout)
715        signal.setitimer(signal.ITIMER_REAL, self.seconds)
716        return self
717    def __exit__(self, exc_type, value, traceback):
718        if qemu_gdb or qemu_valgrind:
719            return False
720        signal.setitimer(signal.ITIMER_REAL, 0)
721        return False
722    def timeout(self, signum, frame):
723        raise TimeoutError(self.errmsg)
724
725def file_pattern(name):
726    return "{0}-{1}".format(os.getpid(), name)
727
728class FilePath:
729    """
730    Context manager generating multiple file names. The generated files are
731    removed when exiting the context.
732
733    Example usage:
734
735        with FilePath('a.img', 'b.img') as (img_a, img_b):
736            # Use img_a and img_b here...
737
738        # a.img and b.img are automatically removed here.
739
740    By default images are created in iotests.test_dir. To create sockets use
741    iotests.sock_dir:
742
743       with FilePath('a.sock', base_dir=iotests.sock_dir) as sock:
744
745    For convenience, calling with one argument yields a single file instead of
746    a tuple with one item.
747
748    """
749    def __init__(self, *names, base_dir=test_dir):
750        self.paths = [os.path.join(base_dir, file_pattern(name))
751                      for name in names]
752
753    def __enter__(self):
754        if len(self.paths) == 1:
755            return self.paths[0]
756        else:
757            return self.paths
758
759    def __exit__(self, exc_type, exc_val, exc_tb):
760        for path in self.paths:
761            try:
762                os.remove(path)
763            except OSError:
764                pass
765        return False
766
767
768def try_remove(img):
769    try:
770        os.remove(img)
771    except OSError:
772        pass
773
774def file_path_remover():
775    for path in reversed(file_path_remover.paths):
776        try_remove(path)
777
778
779def file_path(*names, base_dir=test_dir):
780    ''' Another way to get auto-generated filename that cleans itself up.
781
782    Use is as simple as:
783
784    img_a, img_b = file_path('a.img', 'b.img')
785    sock = file_path('socket')
786    '''
787
788    if not hasattr(file_path_remover, 'paths'):
789        file_path_remover.paths = []
790        atexit.register(file_path_remover)
791
792    paths = []
793    for name in names:
794        filename = file_pattern(name)
795        path = os.path.join(base_dir, filename)
796        file_path_remover.paths.append(path)
797        paths.append(path)
798
799    return paths[0] if len(paths) == 1 else paths
800
801def remote_filename(path):
802    if imgproto == 'file':
803        return path
804    elif imgproto == 'ssh':
805        return "ssh://%s@127.0.0.1:22%s" % (os.environ.get('USER'), path)
806    else:
807        raise ValueError("Protocol %s not supported" % (imgproto))
808
809class VM(qtest.QEMUQtestMachine):
810    '''A QEMU VM'''
811
812    def __init__(self, path_suffix=''):
813        name = "qemu%s-%d" % (path_suffix, os.getpid())
814        timer = 15.0 if not (qemu_gdb or qemu_valgrind) else None
815        if qemu_gdb and qemu_valgrind:
816            sys.stderr.write('gdb and valgrind are mutually exclusive\n')
817            sys.exit(1)
818        wrapper = qemu_gdb if qemu_gdb else qemu_valgrind
819        super().__init__(qemu_prog, qemu_opts, wrapper=wrapper,
820                         name=name,
821                         base_temp_dir=test_dir,
822                         sock_dir=sock_dir, qmp_timer=timer)
823        self._num_drives = 0
824
825    def _post_shutdown(self) -> None:
826        super()._post_shutdown()
827        if not qemu_valgrind or not self._popen:
828            return
829        valgrind_filename = f"{test_dir}/{self._popen.pid}.valgrind"
830        if self.exitcode() == 99:
831            with open(valgrind_filename, encoding='utf-8') as f:
832                print(f.read())
833        else:
834            os.remove(valgrind_filename)
835
836    def _pre_launch(self) -> None:
837        super()._pre_launch()
838        if qemu_print:
839            # set QEMU binary output to stdout
840            self._close_qemu_log_file()
841
842    def add_object(self, opts):
843        self._args.append('-object')
844        self._args.append(opts)
845        return self
846
847    def add_device(self, opts):
848        self._args.append('-device')
849        self._args.append(opts)
850        return self
851
852    def add_drive_raw(self, opts):
853        self._args.append('-drive')
854        self._args.append(opts)
855        return self
856
857    def add_drive(self, path, opts='', interface='virtio', img_format=imgfmt):
858        '''Add a virtio-blk drive to the VM'''
859        options = ['if=%s' % interface,
860                   'id=drive%d' % self._num_drives]
861
862        if path is not None:
863            options.append('file=%s' % path)
864            options.append('format=%s' % img_format)
865            options.append('cache=%s' % cachemode)
866            options.append('aio=%s' % aiomode)
867
868        if opts:
869            options.append(opts)
870
871        if img_format == 'luks' and 'key-secret' not in opts:
872            # default luks support
873            if luks_default_secret_object not in self._args:
874                self.add_object(luks_default_secret_object)
875
876            options.append(luks_default_key_secret_opt)
877
878        self._args.append('-drive')
879        self._args.append(','.join(options))
880        self._num_drives += 1
881        return self
882
883    def add_blockdev(self, opts):
884        self._args.append('-blockdev')
885        if isinstance(opts, str):
886            self._args.append(opts)
887        else:
888            self._args.append(','.join(opts))
889        return self
890
891    def add_incoming(self, addr):
892        self._args.append('-incoming')
893        self._args.append(addr)
894        return self
895
896    def hmp(self, command_line: str, use_log: bool = False) -> QMPMessage:
897        cmd = 'human-monitor-command'
898        kwargs: Dict[str, Any] = {'command-line': command_line}
899        if use_log:
900            return self.qmp_log(cmd, **kwargs)
901        else:
902            return self.qmp(cmd, **kwargs)
903
904    def pause_drive(self, drive: str, event: Optional[str] = None) -> None:
905        """Pause drive r/w operations"""
906        if not event:
907            self.pause_drive(drive, "read_aio")
908            self.pause_drive(drive, "write_aio")
909            return
910        self.hmp(f'qemu-io {drive} "break {event} bp_{drive}"')
911
912    def resume_drive(self, drive: str) -> None:
913        """Resume drive r/w operations"""
914        self.hmp(f'qemu-io {drive} "remove_break bp_{drive}"')
915
916    def hmp_qemu_io(self, drive: str, cmd: str,
917                    use_log: bool = False, qdev: bool = False) -> QMPMessage:
918        """Write to a given drive using an HMP command"""
919        d = '-d ' if qdev else ''
920        return self.hmp(f'qemu-io {d}{drive} "{cmd}"', use_log=use_log)
921
922    def flatten_qmp_object(self, obj, output=None, basestr=''):
923        if output is None:
924            output = {}
925        if isinstance(obj, list):
926            for i, item in enumerate(obj):
927                self.flatten_qmp_object(item, output, basestr + str(i) + '.')
928        elif isinstance(obj, dict):
929            for key in obj:
930                self.flatten_qmp_object(obj[key], output, basestr + key + '.')
931        else:
932            output[basestr[:-1]] = obj # Strip trailing '.'
933        return output
934
935    def qmp_to_opts(self, obj):
936        obj = self.flatten_qmp_object(obj)
937        output_list = []
938        for key in obj:
939            output_list += [key + '=' + obj[key]]
940        return ','.join(output_list)
941
942    def get_qmp_events_filtered(self, wait=60.0):
943        result = []
944        for ev in self.get_qmp_events(wait=wait):
945            result.append(filter_qmp_event(ev))
946        return result
947
948    def qmp_log(self, cmd, filters=(), indent=None, **kwargs):
949        full_cmd = OrderedDict((
950            ("execute", cmd),
951            ("arguments", ordered_qmp(kwargs))
952        ))
953        log(full_cmd, filters, indent=indent)
954        result = self.qmp(cmd, **kwargs)
955        log(result, filters, indent=indent)
956        return result
957
958    # Returns None on success, and an error string on failure
959    def run_job(self, job: str, auto_finalize: bool = True,
960                auto_dismiss: bool = False,
961                pre_finalize: Optional[Callable[[], None]] = None,
962                cancel: bool = False, wait: float = 60.0,
963                filters: Iterable[Callable[[Any], Any]] = (),
964                ) -> Optional[str]:
965        """
966        run_job moves a job from creation through to dismissal.
967
968        :param job: String. ID of recently-launched job
969        :param auto_finalize: Bool. True if the job was launched with
970                              auto_finalize. Defaults to True.
971        :param auto_dismiss: Bool. True if the job was launched with
972                             auto_dismiss=True. Defaults to False.
973        :param pre_finalize: Callback. A callable that takes no arguments to be
974                             invoked prior to issuing job-finalize, if any.
975        :param cancel: Bool. When true, cancels the job after the pre_finalize
976                       callback.
977        :param wait: Float. Timeout value specifying how long to wait for any
978                     event, in seconds. Defaults to 60.0.
979        """
980        match_device = {'data': {'device': job}}
981        match_id = {'data': {'id': job}}
982        events = [
983            ('BLOCK_JOB_COMPLETED', match_device),
984            ('BLOCK_JOB_CANCELLED', match_device),
985            ('BLOCK_JOB_ERROR', match_device),
986            ('BLOCK_JOB_READY', match_device),
987            ('BLOCK_JOB_PENDING', match_id),
988            ('JOB_STATUS_CHANGE', match_id)
989        ]
990        error = None
991        while True:
992            ev = filter_qmp_event(self.events_wait(events, timeout=wait))
993            if ev['event'] != 'JOB_STATUS_CHANGE':
994                log(ev, filters=filters)
995                continue
996            status = ev['data']['status']
997            if status == 'aborting':
998                result = self.qmp('query-jobs')
999                for j in result['return']:
1000                    if j['id'] == job:
1001                        error = j['error']
1002                        log('Job failed: %s' % (j['error']), filters=filters)
1003            elif status == 'ready':
1004                self.qmp_log('job-complete', id=job, filters=filters)
1005            elif status == 'pending' and not auto_finalize:
1006                if pre_finalize:
1007                    pre_finalize()
1008                if cancel:
1009                    self.qmp_log('job-cancel', id=job, filters=filters)
1010                else:
1011                    self.qmp_log('job-finalize', id=job, filters=filters)
1012            elif status == 'concluded' and not auto_dismiss:
1013                self.qmp_log('job-dismiss', id=job, filters=filters)
1014            elif status == 'null':
1015                return error
1016
1017    # Returns None on success, and an error string on failure
1018    def blockdev_create(self, options, job_id='job0', filters=None):
1019        if filters is None:
1020            filters = [filter_qmp_testfiles]
1021        result = self.qmp_log('blockdev-create', filters=filters,
1022                              job_id=job_id, options=options)
1023
1024        if 'return' in result:
1025            assert result['return'] == {}
1026            job_result = self.run_job(job_id, filters=filters)
1027        else:
1028            job_result = result['error']
1029
1030        log("")
1031        return job_result
1032
1033    def enable_migration_events(self, name):
1034        log('Enabling migration QMP events on %s...' % name)
1035        log(self.qmp('migrate-set-capabilities', capabilities=[
1036            {
1037                'capability': 'events',
1038                'state': True
1039            }
1040        ]))
1041
1042    def wait_migration(self, expect_runstate: Optional[str]) -> bool:
1043        while True:
1044            event = self.event_wait('MIGRATION')
1045            # We use the default timeout, and with a timeout, event_wait()
1046            # never returns None
1047            assert event
1048
1049            log(event, filters=[filter_qmp_event])
1050            if event['data']['status'] in ('completed', 'failed'):
1051                break
1052
1053        if event['data']['status'] == 'completed':
1054            # The event may occur in finish-migrate, so wait for the expected
1055            # post-migration runstate
1056            runstate = None
1057            while runstate != expect_runstate:
1058                runstate = self.qmp('query-status')['return']['status']
1059            return True
1060        else:
1061            return False
1062
1063    def node_info(self, node_name):
1064        nodes = self.qmp('query-named-block-nodes')
1065        for x in nodes['return']:
1066            if x['node-name'] == node_name:
1067                return x
1068        return None
1069
1070    def query_bitmaps(self):
1071        res = self.qmp("query-named-block-nodes")
1072        return {device['node-name']: device['dirty-bitmaps']
1073                for device in res['return'] if 'dirty-bitmaps' in device}
1074
1075    def get_bitmap(self, node_name, bitmap_name, recording=None, bitmaps=None):
1076        """
1077        get a specific bitmap from the object returned by query_bitmaps.
1078        :param recording: If specified, filter results by the specified value.
1079        :param bitmaps: If specified, use it instead of call query_bitmaps()
1080        """
1081        if bitmaps is None:
1082            bitmaps = self.query_bitmaps()
1083
1084        for bitmap in bitmaps[node_name]:
1085            if bitmap.get('name', '') == bitmap_name:
1086                if recording is None or bitmap.get('recording') == recording:
1087                    return bitmap
1088        return None
1089
1090    def check_bitmap_status(self, node_name, bitmap_name, fields):
1091        ret = self.get_bitmap(node_name, bitmap_name)
1092
1093        return fields.items() <= ret.items()
1094
1095    def assert_block_path(self, root, path, expected_node, graph=None):
1096        """
1097        Check whether the node under the given path in the block graph
1098        is @expected_node.
1099
1100        @root is the node name of the node where the @path is rooted.
1101
1102        @path is a string that consists of child names separated by
1103        slashes.  It must begin with a slash.
1104
1105        Examples for @root + @path:
1106          - root="qcow2-node", path="/backing/file"
1107          - root="quorum-node", path="/children.2/file"
1108
1109        Hypothetically, @path could be empty, in which case it would
1110        point to @root.  However, in practice this case is not useful
1111        and hence not allowed.
1112
1113        @expected_node may be None.  (All elements of the path but the
1114        leaf must still exist.)
1115
1116        @graph may be None or the result of an x-debug-query-block-graph
1117        call that has already been performed.
1118        """
1119        if graph is None:
1120            graph = self.qmp('x-debug-query-block-graph')['return']
1121
1122        iter_path = iter(path.split('/'))
1123
1124        # Must start with a /
1125        assert next(iter_path) == ''
1126
1127        node = next((node for node in graph['nodes'] if node['name'] == root),
1128                    None)
1129
1130        # An empty @path is not allowed, so the root node must be present
1131        assert node is not None, 'Root node %s not found' % root
1132
1133        for child_name in iter_path:
1134            assert node is not None, 'Cannot follow path %s%s' % (root, path)
1135
1136            try:
1137                node_id = next(edge['child'] for edge in graph['edges']
1138                               if (edge['parent'] == node['id'] and
1139                                   edge['name'] == child_name))
1140
1141                node = next(node for node in graph['nodes']
1142                            if node['id'] == node_id)
1143
1144            except StopIteration:
1145                node = None
1146
1147        if node is None:
1148            assert expected_node is None, \
1149                   'No node found under %s (but expected %s)' % \
1150                   (path, expected_node)
1151        else:
1152            assert node['name'] == expected_node, \
1153                   'Found node %s under %s (but expected %s)' % \
1154                   (node['name'], path, expected_node)
1155
1156index_re = re.compile(r'([^\[]+)\[([^\]]+)\]')
1157
1158class QMPTestCase(unittest.TestCase):
1159    '''Abstract base class for QMP test cases'''
1160
1161    def __init__(self, *args, **kwargs):
1162        super().__init__(*args, **kwargs)
1163        # Many users of this class set a VM property we rely on heavily
1164        # in the methods below.
1165        self.vm = None
1166
1167    def dictpath(self, d, path):
1168        '''Traverse a path in a nested dict'''
1169        for component in path.split('/'):
1170            m = index_re.match(component)
1171            if m:
1172                component, idx = m.groups()
1173                idx = int(idx)
1174
1175            if not isinstance(d, dict) or component not in d:
1176                self.fail(f'failed path traversal for "{path}" in "{d}"')
1177            d = d[component]
1178
1179            if m:
1180                if not isinstance(d, list):
1181                    self.fail(f'path component "{component}" in "{path}" '
1182                              f'is not a list in "{d}"')
1183                try:
1184                    d = d[idx]
1185                except IndexError:
1186                    self.fail(f'invalid index "{idx}" in path "{path}" '
1187                              f'in "{d}"')
1188        return d
1189
1190    def assert_qmp_absent(self, d, path):
1191        try:
1192            result = self.dictpath(d, path)
1193        except AssertionError:
1194            return
1195        self.fail('path "%s" has value "%s"' % (path, str(result)))
1196
1197    def assert_qmp(self, d, path, value):
1198        '''Assert that the value for a specific path in a QMP dict
1199           matches.  When given a list of values, assert that any of
1200           them matches.'''
1201
1202        result = self.dictpath(d, path)
1203
1204        # [] makes no sense as a list of valid values, so treat it as
1205        # an actual single value.
1206        if isinstance(value, list) and value != []:
1207            for v in value:
1208                if result == v:
1209                    return
1210            self.fail('no match for "%s" in %s' % (str(result), str(value)))
1211        else:
1212            self.assertEqual(result, value,
1213                             '"%s" is "%s", expected "%s"'
1214                             % (path, str(result), str(value)))
1215
1216    def assert_no_active_block_jobs(self):
1217        result = self.vm.qmp('query-block-jobs')
1218        self.assert_qmp(result, 'return', [])
1219
1220    def assert_has_block_node(self, node_name=None, file_name=None):
1221        """Issue a query-named-block-nodes and assert node_name and/or
1222        file_name is present in the result"""
1223        def check_equal_or_none(a, b):
1224            return a is None or b is None or a == b
1225        assert node_name or file_name
1226        result = self.vm.qmp('query-named-block-nodes')
1227        for x in result["return"]:
1228            if check_equal_or_none(x.get("node-name"), node_name) and \
1229                    check_equal_or_none(x.get("file"), file_name):
1230                return
1231        self.fail("Cannot find %s %s in result:\n%s" %
1232                  (node_name, file_name, result))
1233
1234    def assert_json_filename_equal(self, json_filename, reference):
1235        '''Asserts that the given filename is a json: filename and that its
1236           content is equal to the given reference object'''
1237        self.assertEqual(json_filename[:5], 'json:')
1238        self.assertEqual(
1239            self.vm.flatten_qmp_object(json.loads(json_filename[5:])),
1240            self.vm.flatten_qmp_object(reference)
1241        )
1242
1243    def cancel_and_wait(self, drive='drive0', force=False,
1244                        resume=False, wait=60.0):
1245        '''Cancel a block job and wait for it to finish, returning the event'''
1246        result = self.vm.qmp('block-job-cancel', device=drive, force=force)
1247        self.assert_qmp(result, 'return', {})
1248
1249        if resume:
1250            self.vm.resume_drive(drive)
1251
1252        cancelled = False
1253        result = None
1254        while not cancelled:
1255            for event in self.vm.get_qmp_events(wait=wait):
1256                if event['event'] == 'BLOCK_JOB_COMPLETED' or \
1257                   event['event'] == 'BLOCK_JOB_CANCELLED':
1258                    self.assert_qmp(event, 'data/device', drive)
1259                    result = event
1260                    cancelled = True
1261                elif event['event'] == 'JOB_STATUS_CHANGE':
1262                    self.assert_qmp(event, 'data/id', drive)
1263
1264
1265        self.assert_no_active_block_jobs()
1266        return result
1267
1268    def wait_until_completed(self, drive='drive0', check_offset=True,
1269                             wait=60.0, error=None):
1270        '''Wait for a block job to finish, returning the event'''
1271        while True:
1272            for event in self.vm.get_qmp_events(wait=wait):
1273                if event['event'] == 'BLOCK_JOB_COMPLETED':
1274                    self.assert_qmp(event, 'data/device', drive)
1275                    if error is None:
1276                        self.assert_qmp_absent(event, 'data/error')
1277                        if check_offset:
1278                            self.assert_qmp(event, 'data/offset',
1279                                            event['data']['len'])
1280                    else:
1281                        self.assert_qmp(event, 'data/error', error)
1282                    self.assert_no_active_block_jobs()
1283                    return event
1284                if event['event'] == 'JOB_STATUS_CHANGE':
1285                    self.assert_qmp(event, 'data/id', drive)
1286
1287    def wait_ready(self, drive='drive0'):
1288        """Wait until a BLOCK_JOB_READY event, and return the event."""
1289        return self.vm.events_wait([
1290            ('BLOCK_JOB_READY',
1291             {'data': {'type': 'mirror', 'device': drive}}),
1292            ('BLOCK_JOB_READY',
1293             {'data': {'type': 'commit', 'device': drive}})
1294        ])
1295
1296    def wait_ready_and_cancel(self, drive='drive0'):
1297        self.wait_ready(drive=drive)
1298        event = self.cancel_and_wait(drive=drive)
1299        self.assertEqual(event['event'], 'BLOCK_JOB_COMPLETED')
1300        self.assert_qmp(event, 'data/type', 'mirror')
1301        self.assert_qmp(event, 'data/offset', event['data']['len'])
1302
1303    def complete_and_wait(self, drive='drive0', wait_ready=True,
1304                          completion_error=None):
1305        '''Complete a block job and wait for it to finish'''
1306        if wait_ready:
1307            self.wait_ready(drive=drive)
1308
1309        result = self.vm.qmp('block-job-complete', device=drive)
1310        self.assert_qmp(result, 'return', {})
1311
1312        event = self.wait_until_completed(drive=drive, error=completion_error)
1313        self.assertTrue(event['data']['type'] in ['mirror', 'commit'])
1314
1315    def pause_wait(self, job_id='job0'):
1316        with Timeout(3, "Timeout waiting for job to pause"):
1317            while True:
1318                result = self.vm.qmp('query-block-jobs')
1319                found = False
1320                for job in result['return']:
1321                    if job['device'] == job_id:
1322                        found = True
1323                        if job['paused'] and not job['busy']:
1324                            return job
1325                        break
1326                assert found
1327
1328    def pause_job(self, job_id='job0', wait=True):
1329        result = self.vm.qmp('block-job-pause', device=job_id)
1330        self.assert_qmp(result, 'return', {})
1331        if wait:
1332            return self.pause_wait(job_id)
1333        return result
1334
1335    def case_skip(self, reason):
1336        '''Skip this test case'''
1337        case_notrun(reason)
1338        self.skipTest(reason)
1339
1340
1341def notrun(reason):
1342    '''Skip this test suite'''
1343    # Each test in qemu-iotests has a number ("seq")
1344    seq = os.path.basename(sys.argv[0])
1345
1346    with open('%s/%s.notrun' % (test_dir, seq), 'w', encoding='utf-8') \
1347            as outfile:
1348        outfile.write(reason + '\n')
1349    logger.warning("%s not run: %s", seq, reason)
1350    sys.exit(0)
1351
1352def case_notrun(reason):
1353    '''Mark this test case as not having been run (without actually
1354    skipping it, that is left to the caller).  See
1355    QMPTestCase.case_skip() for a variant that actually skips the
1356    current test case.'''
1357
1358    # Each test in qemu-iotests has a number ("seq")
1359    seq = os.path.basename(sys.argv[0])
1360
1361    with open('%s/%s.casenotrun' % (test_dir, seq), 'a', encoding='utf-8') \
1362            as outfile:
1363        outfile.write('    [case not run] ' + reason + '\n')
1364
1365def _verify_image_format(supported_fmts: Sequence[str] = (),
1366                         unsupported_fmts: Sequence[str] = ()) -> None:
1367    if 'generic' in supported_fmts and \
1368            os.environ.get('IMGFMT_GENERIC', 'true') == 'true':
1369        # similar to
1370        #   _supported_fmt generic
1371        # for bash tests
1372        supported_fmts = ()
1373
1374    not_sup = supported_fmts and (imgfmt not in supported_fmts)
1375    if not_sup or (imgfmt in unsupported_fmts):
1376        notrun('not suitable for this image format: %s' % imgfmt)
1377
1378    if imgfmt == 'luks':
1379        verify_working_luks()
1380
1381def _verify_protocol(supported: Sequence[str] = (),
1382                     unsupported: Sequence[str] = ()) -> None:
1383    assert not (supported and unsupported)
1384
1385    if 'generic' in supported:
1386        return
1387
1388    not_sup = supported and (imgproto not in supported)
1389    if not_sup or (imgproto in unsupported):
1390        notrun('not suitable for this protocol: %s' % imgproto)
1391
1392def _verify_platform(supported: Sequence[str] = (),
1393                     unsupported: Sequence[str] = ()) -> None:
1394    if any((sys.platform.startswith(x) for x in unsupported)):
1395        notrun('not suitable for this OS: %s' % sys.platform)
1396
1397    if supported:
1398        if not any((sys.platform.startswith(x) for x in supported)):
1399            notrun('not suitable for this OS: %s' % sys.platform)
1400
1401def _verify_cache_mode(supported_cache_modes: Sequence[str] = ()) -> None:
1402    if supported_cache_modes and (cachemode not in supported_cache_modes):
1403        notrun('not suitable for this cache mode: %s' % cachemode)
1404
1405def _verify_aio_mode(supported_aio_modes: Sequence[str] = ()) -> None:
1406    if supported_aio_modes and (aiomode not in supported_aio_modes):
1407        notrun('not suitable for this aio mode: %s' % aiomode)
1408
1409def _verify_formats(required_formats: Sequence[str] = ()) -> None:
1410    usf_list = list(set(required_formats) - set(supported_formats()))
1411    if usf_list:
1412        notrun(f'formats {usf_list} are not whitelisted')
1413
1414
1415def _verify_virtio_blk() -> None:
1416    out = qemu_pipe('-M', 'none', '-device', 'help')
1417    if 'virtio-blk' not in out:
1418        notrun('Missing virtio-blk in QEMU binary')
1419
1420def _verify_virtio_scsi_pci_or_ccw() -> None:
1421    out = qemu_pipe('-M', 'none', '-device', 'help')
1422    if 'virtio-scsi-pci' not in out and 'virtio-scsi-ccw' not in out:
1423        notrun('Missing virtio-scsi-pci or virtio-scsi-ccw in QEMU binary')
1424
1425
1426def _verify_imgopts(unsupported: Sequence[str] = ()) -> None:
1427    imgopts = os.environ.get('IMGOPTS')
1428    # One of usage examples for IMGOPTS is "data_file=$TEST_IMG.ext_data_file"
1429    # but it supported only for bash tests. We don't have a concept of global
1430    # TEST_IMG in iotests.py, not saying about somehow parsing $variables.
1431    # So, for simplicity let's just not support any IMGOPTS with '$' inside.
1432    unsup = list(unsupported) + ['$']
1433    if imgopts and any(x in imgopts for x in unsup):
1434        notrun(f'not suitable for this imgopts: {imgopts}')
1435
1436
1437def supports_quorum() -> bool:
1438    return 'quorum' in qemu_img('--help').stdout
1439
1440def verify_quorum():
1441    '''Skip test suite if quorum support is not available'''
1442    if not supports_quorum():
1443        notrun('quorum support missing')
1444
1445def has_working_luks() -> Tuple[bool, str]:
1446    """
1447    Check whether our LUKS driver can actually create images
1448    (this extends to LUKS encryption for qcow2).
1449
1450    If not, return the reason why.
1451    """
1452
1453    img_file = f'{test_dir}/luks-test.luks'
1454    res = qemu_img('create', '-f', 'luks',
1455                   '--object', luks_default_secret_object,
1456                   '-o', luks_default_key_secret_opt,
1457                   '-o', 'iter-time=10',
1458                   img_file, '1G',
1459                   check=False)
1460    try:
1461        os.remove(img_file)
1462    except OSError:
1463        pass
1464
1465    if res.returncode:
1466        reason = res.stdout
1467        for line in res.stdout.splitlines():
1468            if img_file + ':' in line:
1469                reason = line.split(img_file + ':', 1)[1].strip()
1470                break
1471
1472        return (False, reason)
1473    else:
1474        return (True, '')
1475
1476def verify_working_luks():
1477    """
1478    Skip test suite if LUKS does not work
1479    """
1480    (working, reason) = has_working_luks()
1481    if not working:
1482        notrun(reason)
1483
1484def supports_qcow2_zstd_compression() -> bool:
1485    img_file = f'{test_dir}/qcow2-zstd-test.qcow2'
1486    res = qemu_img('create', '-f', 'qcow2', '-o', 'compression_type=zstd',
1487                   img_file, '0',
1488                   check=False)
1489    try:
1490        os.remove(img_file)
1491    except OSError:
1492        pass
1493
1494    if res.returncode == 1 and \
1495            "'compression-type' does not accept value 'zstd'" in res.stdout:
1496        return False
1497    else:
1498        return True
1499
1500def verify_qcow2_zstd_compression():
1501    if not supports_qcow2_zstd_compression():
1502        notrun('zstd compression not supported')
1503
1504def qemu_pipe(*args: str) -> str:
1505    """
1506    Run qemu with an option to print something and exit (e.g. a help option).
1507
1508    :return: QEMU's stdout output.
1509    """
1510    full_args = [qemu_prog] + qemu_opts + list(args)
1511    output, _ = qemu_tool_pipe_and_status('qemu', full_args)
1512    return output
1513
1514def supported_formats(read_only=False):
1515    '''Set 'read_only' to True to check ro-whitelist
1516       Otherwise, rw-whitelist is checked'''
1517
1518    if not hasattr(supported_formats, "formats"):
1519        supported_formats.formats = {}
1520
1521    if read_only not in supported_formats.formats:
1522        format_message = qemu_pipe("-drive", "format=help")
1523        line = 1 if read_only else 0
1524        supported_formats.formats[read_only] = \
1525            format_message.splitlines()[line].split(":")[1].split()
1526
1527    return supported_formats.formats[read_only]
1528
1529def skip_if_unsupported(required_formats=(), read_only=False):
1530    '''Skip Test Decorator
1531       Runs the test if all the required formats are whitelisted'''
1532    def skip_test_decorator(func):
1533        def func_wrapper(test_case: QMPTestCase, *args: List[Any],
1534                         **kwargs: Dict[str, Any]) -> None:
1535            if callable(required_formats):
1536                fmts = required_formats(test_case)
1537            else:
1538                fmts = required_formats
1539
1540            usf_list = list(set(fmts) - set(supported_formats(read_only)))
1541            if usf_list:
1542                msg = f'{test_case}: formats {usf_list} are not whitelisted'
1543                test_case.case_skip(msg)
1544            else:
1545                func(test_case, *args, **kwargs)
1546        return func_wrapper
1547    return skip_test_decorator
1548
1549def skip_for_formats(formats: Sequence[str] = ()) \
1550    -> Callable[[Callable[[QMPTestCase, List[Any], Dict[str, Any]], None]],
1551                Callable[[QMPTestCase, List[Any], Dict[str, Any]], None]]:
1552    '''Skip Test Decorator
1553       Skips the test for the given formats'''
1554    def skip_test_decorator(func):
1555        def func_wrapper(test_case: QMPTestCase, *args: List[Any],
1556                         **kwargs: Dict[str, Any]) -> None:
1557            if imgfmt in formats:
1558                msg = f'{test_case}: Skipped for format {imgfmt}'
1559                test_case.case_skip(msg)
1560            else:
1561                func(test_case, *args, **kwargs)
1562        return func_wrapper
1563    return skip_test_decorator
1564
1565def skip_if_user_is_root(func):
1566    '''Skip Test Decorator
1567       Runs the test only without root permissions'''
1568    def func_wrapper(*args, **kwargs):
1569        if os.getuid() == 0:
1570            case_notrun('{}: cannot be run as root'.format(args[0]))
1571            return None
1572        else:
1573            return func(*args, **kwargs)
1574    return func_wrapper
1575
1576# We need to filter out the time taken from the output so that
1577# qemu-iotest can reliably diff the results against master output,
1578# and hide skipped tests from the reference output.
1579
1580class ReproducibleTestResult(unittest.TextTestResult):
1581    def addSkip(self, test, reason):
1582        # Same as TextTestResult, but print dot instead of "s"
1583        unittest.TestResult.addSkip(self, test, reason)
1584        if self.showAll:
1585            self.stream.writeln("skipped {0!r}".format(reason))
1586        elif self.dots:
1587            self.stream.write(".")
1588            self.stream.flush()
1589
1590class ReproducibleStreamWrapper:
1591    def __init__(self, stream: TextIO):
1592        self.stream = stream
1593
1594    def __getattr__(self, attr):
1595        if attr in ('stream', '__getstate__'):
1596            raise AttributeError(attr)
1597        return getattr(self.stream, attr)
1598
1599    def write(self, arg=None):
1600        arg = re.sub(r'Ran (\d+) tests? in [\d.]+s', r'Ran \1 tests', arg)
1601        arg = re.sub(r' \(skipped=\d+\)', r'', arg)
1602        self.stream.write(arg)
1603
1604class ReproducibleTestRunner(unittest.TextTestRunner):
1605    def __init__(self, stream: Optional[TextIO] = None,
1606                 resultclass: Type[unittest.TestResult] =
1607                 ReproducibleTestResult,
1608                 **kwargs: Any) -> None:
1609        rstream = ReproducibleStreamWrapper(stream or sys.stdout)
1610        super().__init__(stream=rstream,           # type: ignore
1611                         descriptions=True,
1612                         resultclass=resultclass,
1613                         **kwargs)
1614
1615def execute_unittest(argv: List[str], debug: bool = False) -> None:
1616    """Executes unittests within the calling module."""
1617
1618    # Some tests have warnings, especially ResourceWarnings for unclosed
1619    # files and sockets.  Ignore them for now to ensure reproducibility of
1620    # the test output.
1621    unittest.main(argv=argv,
1622                  testRunner=ReproducibleTestRunner,
1623                  verbosity=2 if debug else 1,
1624                  warnings=None if sys.warnoptions else 'ignore')
1625
1626def execute_setup_common(supported_fmts: Sequence[str] = (),
1627                         supported_platforms: Sequence[str] = (),
1628                         supported_cache_modes: Sequence[str] = (),
1629                         supported_aio_modes: Sequence[str] = (),
1630                         unsupported_fmts: Sequence[str] = (),
1631                         supported_protocols: Sequence[str] = (),
1632                         unsupported_protocols: Sequence[str] = (),
1633                         required_fmts: Sequence[str] = (),
1634                         unsupported_imgopts: Sequence[str] = ()) -> bool:
1635    """
1636    Perform necessary setup for either script-style or unittest-style tests.
1637
1638    :return: Bool; Whether or not debug mode has been requested via the CLI.
1639    """
1640    # Note: Python 3.6 and pylint do not like 'Collection' so use 'Sequence'.
1641
1642    debug = '-d' in sys.argv
1643    if debug:
1644        sys.argv.remove('-d')
1645    logging.basicConfig(level=(logging.DEBUG if debug else logging.WARN))
1646
1647    _verify_image_format(supported_fmts, unsupported_fmts)
1648    _verify_protocol(supported_protocols, unsupported_protocols)
1649    _verify_platform(supported=supported_platforms)
1650    _verify_cache_mode(supported_cache_modes)
1651    _verify_aio_mode(supported_aio_modes)
1652    _verify_formats(required_fmts)
1653    _verify_virtio_blk()
1654    _verify_imgopts(unsupported_imgopts)
1655
1656    return debug
1657
1658def execute_test(*args, test_function=None, **kwargs):
1659    """Run either unittest or script-style tests."""
1660
1661    debug = execute_setup_common(*args, **kwargs)
1662    if not test_function:
1663        execute_unittest(sys.argv, debug)
1664    else:
1665        test_function()
1666
1667def activate_logging():
1668    """Activate iotests.log() output to stdout for script-style tests."""
1669    handler = logging.StreamHandler(stream=sys.stdout)
1670    formatter = logging.Formatter('%(message)s')
1671    handler.setFormatter(formatter)
1672    test_logger.addHandler(handler)
1673    test_logger.setLevel(logging.INFO)
1674    test_logger.propagate = False
1675
1676# This is called from script-style iotests without a single point of entry
1677def script_initialize(*args, **kwargs):
1678    """Initialize script-style tests without running any tests."""
1679    activate_logging()
1680    execute_setup_common(*args, **kwargs)
1681
1682# This is called from script-style iotests with a single point of entry
1683def script_main(test_function, *args, **kwargs):
1684    """Run script-style tests outside of the unittest framework"""
1685    activate_logging()
1686    execute_test(*args, test_function=test_function, **kwargs)
1687
1688# This is called from unittest style iotests
1689def main(*args, **kwargs):
1690    """Run tests using the unittest framework"""
1691    execute_test(*args, **kwargs)
1692