Lines Matching +full:- +full:- +full:disable +full:- +full:attr
1 # Common utilities and Python wrappers for qemu-iotests
56 qemu_img_args = [os.environ.get('QEMU_IMG_PROG', 'qemu-img')]
60 qemu_io_args = [os.environ.get('QEMU_IO_PROG', 'qemu-io')]
64 qemu_io_args_no_fmt = [os.environ.get('QEMU_IO_PROG', 'qemu-io')]
69 qemu_nbd_prog = os.environ.get('QEMU_NBD_PROG', 'qemu-nbd')
77 qsd_prog = os.environ.get('QSD_PROG', 'qemu-storage-daemon')
105 valgrind_logfile = "--log-file=" + test_dir
111 qemu_valgrind = ['valgrind', valgrind_logfile, '--error-exitcode=99']
115 luks_default_key_secret_opt = 'key-secret=keysec0'
122 logger_name: str, level: int = logging.CRITICAL) -> Iterator[None]:
145 connect_stderr: bool = True) -> 'subprocess.Popen[str]':
147 # pylint: disable=consider-using-with
157 -> Tuple[str, int]:
166 {-subp.returncode}: {cmd}\n')
171 def qemu_img_create_prepare_args(args: List[str]) -> List[str]:
177 # -o option may be specified several times
178 p.add_argument('-o', action='append', default=[])
179 p.add_argument('-f')
186 result += ['-f', parsed.f]
198 all('key-secret' not in opts for opts in opts_list):
199 result += ['--object', luks_default_secret_object]
203 result += ['-o', opts]
211 ) -> 'subprocess.CompletedProcess[str]':
220 When the return code is negative, or on any non-zero exit code
224 handled, the command-line, return code, and all console output
251 ) -> 'subprocess.CompletedProcess[str]':
272 k = k.replace('_', '-')
277 def qemu_img_create(*args: str) -> 'subprocess.CompletedProcess[str]':
280 def qemu_img_json(*args: str) -> Any:
282 Run qemu-img and return its output as deserialized JSON.
285 When qemu-img crashes, or returns a non-zero exit code without
288 When qemu-img returns 0, but failed to produce a valid JSON document.
301 # multi-command flexibility, ignore the exact error codes and
312 def qemu_img_measure(*args: str) -> Any:
313 return qemu_img_json("measure", "--output", "json", *args)
315 def qemu_img_check(*args: str) -> Any:
316 return qemu_img_json("check", "--output", "json", *args)
318 def qemu_img_info(*args: str) -> Any:
319 return qemu_img_json('info', "--output", "json", *args)
321 def qemu_img_map(*args: str) -> Any:
322 return qemu_img_json('map', "--output", "json", *args)
325 ) -> 'subprocess.CompletedProcess[str]':
333 ) -> None:
336 args.append('--image-opts')
338 args += ['-f', imgfmt]
347 def qemu_io_wrap_args(args: Sequence[str]) -> List[str]:
348 if '-f' in args or '--image-opts' in args:
357 ) -> 'subprocess.CompletedProcess[str]':
368 ) -> 'subprocess.CompletedProcess[str]':
377 # close it immediately. Therefore, disable the pylint check:
378 # pylint: disable=consider-using-with
384 if out != 'qemu-io> ':
385 # Most probably qemu-io just failed to start.
395 pattern = 'qemu-io> '
409 return ''.join(s[:-n])
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]
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']
442 # pylint: disable=consider-using-with
450 'qemu-storage-daemon terminated with exit code ' +
455 with open(self.pidfile, encoding='utf-8') as f:
461 -> QMPMessage:
465 def get_qmp(self) -> QEMUMonitorProtocol:
470 -> QMPReturnValue:
498 '''Run qemu-nbd in daemon mode and return the parent's exit code'''
499 return subprocess.call(qemu_nbd_args + ['--fork'] + list(args))
501 def qemu_nbd_early_pipe(*args: str) -> Tuple[int, str]:
502 '''Run qemu-nbd in daemon mode and return both the parent's exit code
504 full_args = qemu_nbd_args + ['--fork'] + list(args)
505 output, returncode = qemu_tool_pipe_and_status('qemu-nbd', full_args,
509 def qemu_nbd_list_log(*args: str) -> str:
510 '''Run qemu-nbd to list remote exports'''
511 full_args = [qemu_nbd_prog, '-L'] + list(args)
512 output, _ = qemu_tool_pipe_and_status('qemu-nbd', full_args)
518 '''Context manager running qemu-nbd within the context'''
519 pid_file = file_path("qemu_nbd_popen-nbd-pid-file")
524 cmd.extend(('--persistent', '--pid-file', pid_file))
533 "qemu-nbd terminated with exit code {}: {}"
546 fmt1: str = imgfmt, fmt2: str = imgfmt) -> bool:
551 when qemu-img crashes or returns a status code of anything other
555 qemu_img('compare', '-f', fmt1, '-F', fmt2, img1, img2)
563 '''Create a fully-allocated raw image with sector markers'''
571 def image_size(img: str) -> int:
573 value = qemu_img_info('-f', imgfmt, img)['virtual-size']
576 raise TypeError("Expected 'int' for 'virtual-size', "
591 qemu_io_re = re.compile(r"[0-9]* ops; [0-9\/:. sec]* "
592 r"\([0-9\/.inf]* [EPTGMKiBbytes]*\/sec "
593 r"and [0-9\/.inf]* ops\/sec\)")
599 chown_re = re.compile(r"chown [0-9]+:[0-9]+")
640 pref1 = os.path.join(test_dir, "%s-" % (os.getpid()))
641 pref2 = os.path.join(sock_dir, "%s-" % (os.getpid()))
642 return msg.replace(pref1, 'TEST_DIR/PID-').replace(pref2, 'SOCK_DIR/PID-')
651 def filter_virtio_scsi(output: str) -> str:
652 return re.sub(r'(virtio-scsi)-(ccw|pci)', r'\1', output)
662 return re.sub("#block[0-9]+", "NODE_NAME", msg)
672 drop_child_info: bool = True) -> str:
676 if 'disk size' in line or 'actual-size' in line:
691 line = re.sub('iters: [0-9]+', 'iters: XXX', line)
692 line = re.sub('uuid: [-a-f0-9]+',
693 'uuid: XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX',
695 line = re.sub('cid: [0-9]+', 'cid: XXXXXXXXXX', line)
711 def filter_nbd_exports(output: str) -> str:
712 return re.sub(r'((min|opt|max) block): [0-9]+', r'\1: XXX', output)
714 def filter_qtest(output: str) -> str:
723 indent: Optional[int] = None) -> None:
726 If indent is provided, JSON serializable messages are pretty-printed.
756 return "{0}-{1}".format(os.getpid(), name)
810 ''' Another way to get auto-generated filename that cleans itself up.
843 name = "qemu%s-%d" % (path_suffix, os.getpid())
855 def _post_shutdown(self) -> None:
861 with open(valgrind_filename, encoding='utf-8') as f:
866 def _pre_launch(self) -> None:
873 self._args.append('-object')
878 self._args.append('-device')
883 self._args.append('-drive')
888 '''Add a virtio-blk drive to the VM'''
901 if img_format == 'luks' and 'key-secret' not in opts:
908 self._args.append('-drive')
914 self._args.append('-blockdev')
922 self._args.append('-incoming')
927 self._args.append('-S')
930 def hmp(self, command_line: str, use_log: bool = False) -> QMPMessage:
931 cmd = 'human-monitor-command'
932 kwargs: Dict[str, Any] = {'command-line': command_line}
938 def pause_drive(self, drive: str, event: Optional[str] = None) -> None:
944 self.hmp(f'qemu-io {drive} "break {event} bp_{drive}"')
946 def resume_drive(self, drive: str) -> None:
948 self.hmp(f'qemu-io {drive} "remove_break bp_{drive}"')
951 use_log: bool = False, qdev: bool = False) -> QMPMessage:
953 d = '-d ' if qdev else ''
954 return self.hmp(f'qemu-io {d}{drive} "{cmd}"', use_log=use_log)
966 output[basestr[:-1]] = obj # Strip trailing '.'
998 ) -> Optional[str]:
1002 :param job: String. ID of recently-launched job
1008 invoked prior to issuing job-finalize, if any.
1032 result = self.qmp('query-jobs')
1038 self.qmp_log('job-complete', id=job, filters=filters)
1043 self.qmp_log('job-cancel', id=job, filters=filters)
1045 self.qmp_log('job-finalize', id=job, filters=filters)
1047 self.qmp_log('job-dismiss', id=job, filters=filters)
1055 result = self.qmp_log('blockdev-create', filters=filters,
1069 log(self.qmp('migrate-set-capabilities', capabilities=[
1076 def wait_migration(self, expect_runstate: Optional[str]) -> bool:
1088 # The event may occur in finish-migrate, so wait for the expected
1089 # post-migration runstate
1092 runstate = self.qmp('query-status')['return']['status']
1098 nodes = self.qmp('query-named-block-nodes')
1100 if x['node-name'] == node_name:
1105 res = self.qmp("query-named-block-nodes")
1106 return {device['node-name']: device['dirty-bitmaps']
1107 for device in res['return'] if 'dirty-bitmaps' in device}
1140 - root="qcow2-node", path="/backing/file"
1141 - root="quorum-node", path="/children.2/file"
1150 @graph may be None or the result of an x-debug-query-block-graph
1154 graph = self.qmp('x-debug-query-block-graph')['return']
1251 result = self.vm.qmp('query-block-jobs')
1255 """Issue a query-named-block-nodes and assert node_name and/or
1260 result = self.vm.qmp('query-named-block-nodes')
1262 if check_equal_or_none(x.get("node-name"), node_name) and \
1280 self.vm.cmd('block-job-cancel', device=drive, force=force)
1342 self.vm.cmd('block-job-complete', device=drive)
1350 result = self.vm.qmp('query-block-jobs')
1361 self.vm.cmd('block-job-pause', device=job_id)
1373 # Each test in qemu-iotests has a number ("seq")
1376 with open('%s/%s.notrun' % (test_dir, seq), 'w', encoding='utf-8') \
1388 # Each test in qemu-iotests has a number ("seq")
1391 with open('%s/%s.casenotrun' % (test_dir, seq), 'a', encoding='utf-8') \
1396 unsupported_fmts: Sequence[str] = ()) -> None:
1412 unsupported: Sequence[str] = ()) -> None:
1423 unsupported: Sequence[str] = ()) -> None:
1431 def _verify_cache_mode(supported_cache_modes: Sequence[str] = ()) -> None:
1435 def _verify_aio_mode(supported_aio_modes: Sequence[str] = ()) -> None:
1439 def _verify_formats(required_formats: Sequence[str] = ()) -> None:
1440 usf_list = list(set(required_formats) - set(supported_formats()))
1445 def _verify_virtio_blk() -> None:
1446 out = qemu_pipe('-M', 'none', '-device', 'help')
1447 if 'virtio-blk' not in out:
1448 notrun('Missing virtio-blk in QEMU binary')
1450 def verify_virtio_scsi_pci_or_ccw() -> None:
1451 out = qemu_pipe('-M', 'none', '-device', 'help')
1452 if 'virtio-scsi-pci' not in out and 'virtio-scsi-ccw' not in out:
1453 notrun('Missing virtio-scsi-pci or virtio-scsi-ccw in QEMU binary')
1456 def _verify_imgopts(unsupported: Sequence[str] = ()) -> None:
1467 def supports_quorum() -> bool:
1468 return 'quorum' in qemu_img('--help').stdout
1475 def has_working_luks() -> Tuple[bool, str]:
1483 img_file = f'{test_dir}/luks-test.luks'
1484 res = qemu_img('create', '-f', 'luks',
1485 '--object', luks_default_secret_object,
1486 '-o', luks_default_key_secret_opt,
1487 '-o', 'iter-time=10',
1514 def supports_qcow2_zstd_compression() -> bool:
1515 img_file = f'{test_dir}/qcow2-zstd-test.qcow2'
1516 res = qemu_img('create', '-f', 'qcow2', '-o', 'compression_type=zstd',
1525 "'compression-type' does not accept value 'zstd'" in res.stdout:
1534 def qemu_pipe(*args: str) -> str:
1545 '''Set 'read_only' to True to check ro-whitelist
1546 Otherwise, rw-whitelist is checked'''
1552 format_message = qemu_pipe("-drive", "format=help")
1564 **kwargs: Dict[str, Any]) -> None:
1570 usf_list = list(set(fmts) - set(supported_formats(read_only)))
1580 -> Callable[[Callable[[QMPTestCase, List[Any], Dict[str, Any]], None]],
1586 **kwargs: Dict[str, Any]) -> None:
1607 # qemu-iotest can reliably diff the results against master output,
1624 def __getattr__(self, attr): argument
1625 if attr in ('stream', '__getstate__'):
1626 raise AttributeError(attr)
1627 return getattr(self.stream, attr)
1641 ) -> None:
1648 def execute_unittest(argv: List[str], debug: bool = False) -> None:
1667 unsupported_imgopts: Sequence[str] = ()) -> bool:
1669 Perform necessary setup for either script-style or unittest-style tests.
1675 debug = '-d' in sys.argv
1677 sys.argv.remove('-d')
1692 """Run either unittest or script-style tests."""
1701 """Activate iotests.log() output to stdout for script-style tests."""
1709 # This is called from script-style iotests without a single point of entry
1711 """Initialize script-style tests without running any tests."""
1715 # This is called from script-style iotests with a single point of entry
1717 """Run script-style tests outside of the unittest framework"""