Lines Matching +full:stdout +full:- +full:path
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]:
139 sample_fname = os.path.join(sample_img_dir, sample + '.bz2')
145 connect_stderr: bool = True) -> 'subprocess.Popen[str]':
146 stderr = subprocess.STDOUT if connect_stderr else None
147 # pylint: disable=consider-using-with
149 stdout=subprocess.PIPE,
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]':
217 :param combine_stdio: set to False to keep stdout/stderr separated.
220 When the return code is negative, or on any non-zero exit code
222 'stdout', 'stderr', and 'returncode' properties that may be
224 handled, the command-line, return code, and all console output
228 a CompletedProcess. This object has args, returncode, and stdout
234 stdout=subprocess.PIPE,
235 stderr=subprocess.STDOUT if combine_stdio else subprocess.PIPE,
243 output=subp.stdout,
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
286 producing a valid JSON document to stdout.
288 When qemu-img returns 0, but failed to produce a valid JSON document.
301 # multi-command flexibility, ignore the exact error codes and
304 return json.loads(exc.stdout)
310 return json.loads(res.stdout)
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]':
327 log(result.stdout, filters=[filter_testfiles])
333 ) -> None:
336 args.append('--image-opts')
338 args += ['-f', imgfmt]
342 output = qemu_img(*args, check=check).stdout
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]':
370 log(result.stdout, filters=[filter_testfiles, filter_qemu_io])
378 # pylint: disable=consider-using-with
380 stdout=subprocess.PIPE,
381 stderr=subprocess.STDOUT,
383 out = self._p.stdout.read(9)
384 if out != 'qemu-io> ':
385 # Most probably qemu-io just failed to start.
387 out += self._p.stdout.read()
395 pattern = 'qemu-io> '
400 c = self._p.stdout.read(1)
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
446 while not os.path.exists(self.pidfile):
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")
521 assert not os.path.exists(pid_file)
524 cmd.extend(('--persistent', '--pid-file', pid_file))
530 while not os.path.exists(pid_file):
533 "qemu-nbd terminated with exit code {}: {}"
539 if os.path.exists(pid_file):
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]+")
630 pref1 = os.path.join(test_dir, "%s-" % (os.getpid()))
631 pref2 = os.path.join(sock_dir, "%s-" % (os.getpid()))
632 return msg.replace(pref1, 'TEST_DIR/PID-').replace(pref2, 'SOCK_DIR/PID-')
641 def filter_virtio_scsi(output: str) -> str:
642 return re.sub(r'(virtio-scsi)-(ccw|pci)', r'\1', output)
652 return re.sub("#block[0-9]+", "NODE_NAME", msg)
662 drop_child_info: bool = True) -> str:
666 if 'disk size' in line or 'actual-size' in line:
681 line = re.sub('iters: [0-9]+', 'iters: XXX', line)
682 line = re.sub('uuid: [-a-f0-9]+',
683 'uuid: XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX',
685 line = re.sub('cid: [0-9]+', 'cid: XXXXXXXXXX', line)
701 def filter_nbd_exports(output: str) -> str:
702 return re.sub(r'((min|opt|max) block): [0-9]+', r'\1: XXX', output)
709 indent: Optional[int] = None) -> None:
712 If indent is provided, JSON serializable messages are pretty-printed.
742 return "{0}-{1}".format(os.getpid(), name)
766 self.paths = [os.path.join(base_dir, file_pattern(name))
776 for path in self.paths:
778 os.remove(path)
791 for path in reversed(file_path_remover.paths):
792 try_remove(path)
796 ''' Another way to get auto-generated filename that cleans itself up.
811 path = os.path.join(base_dir, filename)
812 file_path_remover.paths.append(path)
813 paths.append(path)
817 def remote_filename(path): argument
819 return path
821 return "ssh://%s@127.0.0.1:22%s" % (os.environ.get('USER'), path)
829 name = "qemu%s-%d" % (path_suffix, os.getpid())
841 def _post_shutdown(self) -> None:
847 with open(valgrind_filename, encoding='utf-8') as f:
852 def _pre_launch(self) -> None:
855 # set QEMU binary output to stdout
859 self._args.append('-object')
864 self._args.append('-device')
869 self._args.append('-drive')
873 def add_drive(self, path, opts='', interface='virtio', img_format=imgfmt): argument
874 '''Add a virtio-blk drive to the VM'''
878 if path is not None:
879 options.append('file=%s' % path)
887 if img_format == 'luks' and 'key-secret' not in opts:
894 self._args.append('-drive')
900 self._args.append('-blockdev')
908 self._args.append('-incoming')
912 def hmp(self, command_line: str, use_log: bool = False) -> QMPMessage:
913 cmd = 'human-monitor-command'
914 kwargs: Dict[str, Any] = {'command-line': command_line}
920 def pause_drive(self, drive: str, event: Optional[str] = None) -> None:
926 self.hmp(f'qemu-io {drive} "break {event} bp_{drive}"')
928 def resume_drive(self, drive: str) -> None:
930 self.hmp(f'qemu-io {drive} "remove_break bp_{drive}"')
933 use_log: bool = False, qdev: bool = False) -> QMPMessage:
935 d = '-d ' if qdev else ''
936 return self.hmp(f'qemu-io {d}{drive} "{cmd}"', use_log=use_log)
948 output[basestr[:-1]] = obj # Strip trailing '.'
980 ) -> Optional[str]:
984 :param job: String. ID of recently-launched job
990 invoked prior to issuing job-finalize, if any.
1014 result = self.qmp('query-jobs')
1020 self.qmp_log('job-complete', id=job, filters=filters)
1025 self.qmp_log('job-cancel', id=job, filters=filters)
1027 self.qmp_log('job-finalize', id=job, filters=filters)
1029 self.qmp_log('job-dismiss', id=job, filters=filters)
1037 result = self.qmp_log('blockdev-create', filters=filters,
1051 log(self.qmp('migrate-set-capabilities', capabilities=[
1058 def wait_migration(self, expect_runstate: Optional[str]) -> bool:
1070 # The event may occur in finish-migrate, so wait for the expected
1071 # post-migration runstate
1074 runstate = self.qmp('query-status')['return']['status']
1080 nodes = self.qmp('query-named-block-nodes')
1082 if x['node-name'] == node_name:
1087 res = self.qmp("query-named-block-nodes")
1088 return {device['node-name']: device['dirty-bitmaps']
1089 for device in res['return'] if 'dirty-bitmaps' in device}
1111 def assert_block_path(self, root, path, expected_node, graph=None): argument
1113 Check whether the node under the given path in the block graph
1116 @root is the node name of the node where the @path is rooted.
1118 @path is a string that consists of child names separated by
1121 Examples for @root + @path:
1122 - root="qcow2-node", path="/backing/file"
1123 - root="quorum-node", path="/children.2/file"
1125 Hypothetically, @path could be empty, in which case it would
1129 @expected_node may be None. (All elements of the path but the
1132 @graph may be None or the result of an x-debug-query-block-graph
1136 graph = self.qmp('x-debug-query-block-graph')['return']
1138 iter_path = iter(path.split('/'))
1146 # An empty @path is not allowed, so the root node must be present
1150 assert node is not None, 'Cannot follow path %s%s' % (root, path)
1166 (path, expected_node)
1170 (node['name'], path, expected_node)
1183 def dictpath(self, d, path): argument
1184 '''Traverse a path in a nested dict'''
1185 for component in path.split('/'):
1192 self.fail(f'failed path traversal for "{path}" in "{d}"')
1197 self.fail(f'path component "{component}" in "{path}" '
1202 self.fail(f'invalid index "{idx}" in path "{path}" '
1206 def assert_qmp_absent(self, d, path): argument
1208 result = self.dictpath(d, path)
1211 self.fail('path "%s" has value "%s"' % (path, str(result)))
1213 def assert_qmp(self, d, path, value): argument
1214 '''Assert that the value for a specific path in a QMP dict
1218 result = self.dictpath(d, path)
1230 % (path, str(result), str(value)))
1233 result = self.vm.qmp('query-block-jobs')
1237 """Issue a query-named-block-nodes and assert node_name and/or
1242 result = self.vm.qmp('query-named-block-nodes')
1244 if check_equal_or_none(x.get("node-name"), node_name) and \
1262 self.vm.cmd('block-job-cancel', device=drive, force=force)
1324 self.vm.cmd('block-job-complete', device=drive)
1332 result = self.vm.qmp('query-block-jobs')
1343 self.vm.cmd('block-job-pause', device=job_id)
1355 # Each test in qemu-iotests has a number ("seq")
1356 seq = os.path.basename(sys.argv[0])
1358 with open('%s/%s.notrun' % (test_dir, seq), 'w', encoding='utf-8') \
1370 # Each test in qemu-iotests has a number ("seq")
1371 seq = os.path.basename(sys.argv[0])
1373 with open('%s/%s.casenotrun' % (test_dir, seq), 'a', encoding='utf-8') \
1378 unsupported_fmts: Sequence[str] = ()) -> None:
1394 unsupported: Sequence[str] = ()) -> None:
1405 unsupported: Sequence[str] = ()) -> None:
1413 def _verify_cache_mode(supported_cache_modes: Sequence[str] = ()) -> None:
1417 def _verify_aio_mode(supported_aio_modes: Sequence[str] = ()) -> None:
1421 def _verify_formats(required_formats: Sequence[str] = ()) -> None:
1422 usf_list = list(set(required_formats) - set(supported_formats()))
1427 def _verify_virtio_blk() -> None:
1428 out = qemu_pipe('-M', 'none', '-device', 'help')
1429 if 'virtio-blk' not in out:
1430 notrun('Missing virtio-blk in QEMU binary')
1432 def verify_virtio_scsi_pci_or_ccw() -> None:
1433 out = qemu_pipe('-M', 'none', '-device', 'help')
1434 if 'virtio-scsi-pci' not in out and 'virtio-scsi-ccw' not in out:
1435 notrun('Missing virtio-scsi-pci or virtio-scsi-ccw in QEMU binary')
1438 def _verify_imgopts(unsupported: Sequence[str] = ()) -> None:
1449 def supports_quorum() -> bool:
1450 return 'quorum' in qemu_img('--help').stdout
1457 def has_working_luks() -> Tuple[bool, str]:
1465 img_file = f'{test_dir}/luks-test.luks'
1466 res = qemu_img('create', '-f', 'luks',
1467 '--object', luks_default_secret_object,
1468 '-o', luks_default_key_secret_opt,
1469 '-o', 'iter-time=10',
1478 reason = res.stdout
1479 for line in res.stdout.splitlines():
1496 def supports_qcow2_zstd_compression() -> bool:
1497 img_file = f'{test_dir}/qcow2-zstd-test.qcow2'
1498 res = qemu_img('create', '-f', 'qcow2', '-o', 'compression_type=zstd',
1507 "'compression-type' does not accept value 'zstd'" in res.stdout:
1516 def qemu_pipe(*args: str) -> str:
1520 :return: QEMU's stdout output.
1527 '''Set 'read_only' to True to check ro-whitelist
1528 Otherwise, rw-whitelist is checked'''
1534 format_message = qemu_pipe("-drive", "format=help")
1546 **kwargs: Dict[str, Any]) -> None:
1552 usf_list = list(set(fmts) - set(supported_formats(read_only)))
1562 -> Callable[[Callable[[QMPTestCase, List[Any], Dict[str, Any]], None]],
1568 **kwargs: Dict[str, Any]) -> None:
1589 # qemu-iotest can reliably diff the results against master output,
1623 ) -> None:
1624 rstream = ReproducibleStreamWrapper(stream or sys.stdout)
1630 def execute_unittest(argv: List[str], debug: bool = False) -> None:
1649 unsupported_imgopts: Sequence[str] = ()) -> bool:
1651 Perform necessary setup for either script-style or unittest-style tests.
1657 debug = '-d' in sys.argv
1659 sys.argv.remove('-d')
1674 """Run either unittest or script-style tests."""
1683 """Activate iotests.log() output to stdout for script-style tests."""
1684 handler = logging.StreamHandler(stream=sys.stdout)
1691 # This is called from script-style iotests without a single point of entry
1693 """Initialize script-style tests without running any tests."""
1697 # This is called from script-style iotests with a single point of entry
1699 """Run script-style tests outside of the unittest framework"""