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