1 # Common utilities and Python wrappers for qemu-iotests
3 # Copyright (C) 2012 IBM Corp.
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.
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.
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/>.
20 from collections
import OrderedDict
31 from typing
import (Any
, Callable
, Dict
, Iterable
, List
, Optional
, TypeVar
)
34 # pylint: disable=import-error, wrong-import-position
35 sys
.path
.append(os
.path
.join(os
.path
.dirname(__file__
), '..', '..', 'python'))
36 from qemu
import qtest
38 assert sys
.version_info
>= (3, 6)
42 # This will not work if arguments contain spaces but is necessary if we
43 # want to support the override options that ./check supports.
44 qemu_img_args
= [os
.environ
.get('QEMU_IMG_PROG', 'qemu-img')]
45 if os
.environ
.get('QEMU_IMG_OPTIONS'):
46 qemu_img_args
+= os
.environ
['QEMU_IMG_OPTIONS'].strip().split(' ')
48 qemu_io_args
= [os
.environ
.get('QEMU_IO_PROG', 'qemu-io')]
49 if os
.environ
.get('QEMU_IO_OPTIONS'):
50 qemu_io_args
+= os
.environ
['QEMU_IO_OPTIONS'].strip().split(' ')
52 qemu_io_args_no_fmt
= [os
.environ
.get('QEMU_IO_PROG', 'qemu-io')]
53 if os
.environ
.get('QEMU_IO_OPTIONS_NO_FMT'):
54 qemu_io_args_no_fmt
+= \
55 os
.environ
['QEMU_IO_OPTIONS_NO_FMT'].strip().split(' ')
57 qemu_nbd_args
= [os
.environ
.get('QEMU_NBD_PROG', 'qemu-nbd')]
58 if os
.environ
.get('QEMU_NBD_OPTIONS'):
59 qemu_nbd_args
+= os
.environ
['QEMU_NBD_OPTIONS'].strip().split(' ')
61 qemu_prog
= os
.environ
.get('QEMU_PROG', 'qemu')
62 qemu_opts
= os
.environ
.get('QEMU_OPTIONS', '').strip().split(' ')
64 imgfmt
= os
.environ
.get('IMGFMT', 'raw')
65 imgproto
= os
.environ
.get('IMGPROTO', 'file')
66 test_dir
= os
.environ
.get('TEST_DIR')
67 sock_dir
= os
.environ
.get('SOCK_DIR')
68 output_dir
= os
.environ
.get('OUTPUT_DIR', '.')
69 cachemode
= os
.environ
.get('CACHEMODE')
70 aiomode
= os
.environ
.get('AIOMODE')
71 qemu_default_machine
= os
.environ
.get('QEMU_DEFAULT_MACHINE')
73 socket_scm_helper
= os
.environ
.get('SOCKET_SCM_HELPER', 'socket_scm_helper')
75 luks_default_secret_object
= 'secret,id=keysec0,data=' + \
76 os
.environ
.get('IMGKEYSECRET', '')
77 luks_default_key_secret_opt
= 'key-secret=keysec0'
81 '''Run qemu-img and return the exit code'''
82 devnull
= open('/dev/null', 'r+')
83 exitcode
= subprocess
.call(qemu_img_args
+ list(args
), stdin
=devnull
, stdout
=devnull
)
85 sys
.stderr
.write('qemu-img received signal %i: %s\n' % (-exitcode
, ' '.join(qemu_img_args
+ list(args
))))
88 def ordered_qmp(qmsg
, conv_keys
=True):
89 # Dictionaries are not ordered prior to 3.6, therefore:
90 if isinstance(qmsg
, list):
91 return [ordered_qmp(atom
) for atom
in qmsg
]
92 if isinstance(qmsg
, dict):
94 for k
, v
in sorted(qmsg
.items()):
96 k
= k
.replace('_', '-')
97 od
[k
] = ordered_qmp(v
, conv_keys
=False)
101 def qemu_img_create(*args
):
104 # default luks support
105 if '-f' in args
and args
[args
.index('-f') + 1] == 'luks':
108 if 'key-secret' not in args
[i
+ 1]:
109 args
[i
+ 1].append(luks_default_key_secret_opt
)
110 args
.insert(i
+ 2, '--object')
111 args
.insert(i
+ 3, luks_default_secret_object
)
113 args
= ['-o', luks_default_key_secret_opt
,
114 '--object', luks_default_secret_object
] + args
116 args
.insert(0, 'create')
118 return qemu_img(*args
)
120 def qemu_img_verbose(*args
):
121 '''Run qemu-img without suppressing its output and return the exit code'''
122 exitcode
= subprocess
.call(qemu_img_args
+ list(args
))
124 sys
.stderr
.write('qemu-img received signal %i: %s\n' % (-exitcode
, ' '.join(qemu_img_args
+ list(args
))))
127 def qemu_img_pipe(*args
):
128 '''Run qemu-img and return its output'''
129 subp
= subprocess
.Popen(qemu_img_args
+ list(args
),
130 stdout
=subprocess
.PIPE
,
131 stderr
=subprocess
.STDOUT
,
132 universal_newlines
=True)
133 exitcode
= subp
.wait()
135 sys
.stderr
.write('qemu-img received signal %i: %s\n' % (-exitcode
, ' '.join(qemu_img_args
+ list(args
))))
136 return subp
.communicate()[0]
138 def qemu_img_log(*args
):
139 result
= qemu_img_pipe(*args
)
140 log(result
, filters
=[filter_testfiles
])
143 def img_info_log(filename
, filter_path
=None, imgopts
=False, extra_args
=()):
146 args
.append('--image-opts')
148 args
+= ['-f', imgfmt
]
150 args
.append(filename
)
152 output
= qemu_img_pipe(*args
)
154 filter_path
= filename
155 log(filter_img_info(output
, filter_path
))
158 '''Run qemu-io and return the stdout data'''
159 args
= qemu_io_args
+ list(args
)
160 subp
= subprocess
.Popen(args
, stdout
=subprocess
.PIPE
,
161 stderr
=subprocess
.STDOUT
,
162 universal_newlines
=True)
163 exitcode
= subp
.wait()
165 sys
.stderr
.write('qemu-io received signal %i: %s\n' % (-exitcode
, ' '.join(args
)))
166 return subp
.communicate()[0]
168 def qemu_io_log(*args
):
169 result
= qemu_io(*args
)
170 log(result
, filters
=[filter_testfiles
, filter_qemu_io
])
173 def qemu_io_silent(*args
):
174 '''Run qemu-io and return the exit code, suppressing stdout'''
175 args
= qemu_io_args
+ list(args
)
176 exitcode
= subprocess
.call(args
, stdout
=open('/dev/null', 'w'))
178 sys
.stderr
.write('qemu-io received signal %i: %s\n' %
179 (-exitcode
, ' '.join(args
)))
182 def qemu_io_silent_check(*args
):
183 '''Run qemu-io and return the true if subprocess returned 0'''
184 args
= qemu_io_args
+ list(args
)
185 exitcode
= subprocess
.call(args
, stdout
=open('/dev/null', 'w'),
186 stderr
=subprocess
.STDOUT
)
189 def get_virtio_scsi_device():
190 if qemu_default_machine
== 's390-ccw-virtio':
191 return 'virtio-scsi-ccw'
192 return 'virtio-scsi-pci'
194 class QemuIoInteractive
:
195 def __init__(self
, *args
):
196 self
.args
= qemu_io_args
+ list(args
)
197 self
._p
= subprocess
.Popen(self
.args
, stdin
=subprocess
.PIPE
,
198 stdout
=subprocess
.PIPE
,
199 stderr
=subprocess
.STDOUT
,
200 universal_newlines
=True)
201 assert self
._p
.stdout
.read(9) == 'qemu-io> '
204 self
._p
.communicate('q\n')
206 def _read_output(self
):
207 pattern
= 'qemu-io> '
212 c
= self
._p
.stdout
.read(1)
213 # check unexpected EOF
216 if c
== pattern
[pos
]:
221 return ''.join(s
[:-n
])
224 # quit command is in close(), '\n' is added automatically
225 assert '\n' not in cmd
227 assert cmd
not in ('q', 'quit')
228 self
._p
.stdin
.write(cmd
+ '\n')
229 self
._p
.stdin
.flush()
230 return self
._read
_output
()
234 '''Run qemu-nbd in daemon mode and return the parent's exit code'''
235 return subprocess
.call(qemu_nbd_args
+ ['--fork'] + list(args
))
237 def qemu_nbd_early_pipe(*args
):
238 '''Run qemu-nbd in daemon mode and return both the parent's exit code
239 and its output in case of an error'''
240 subp
= subprocess
.Popen(qemu_nbd_args
+ ['--fork'] + list(args
),
241 stdout
=subprocess
.PIPE
,
242 stderr
=subprocess
.STDOUT
,
243 universal_newlines
=True)
244 exitcode
= subp
.wait()
246 sys
.stderr
.write('qemu-nbd received signal %i: %s\n' %
248 ' '.join(qemu_nbd_args
+ ['--fork'] + list(args
))))
250 return exitcode
, subp
.communicate()[0] if exitcode
else ''
252 def qemu_nbd_popen(*args
):
253 '''Run qemu-nbd in daemon mode and return the parent's exit code'''
254 return subprocess
.Popen(qemu_nbd_args
+ ['--persistent'] + list(args
))
256 def compare_images(img1
, img2
, fmt1
=imgfmt
, fmt2
=imgfmt
):
257 '''Return True if two image files are identical'''
258 return qemu_img('compare', '-f', fmt1
,
259 '-F', fmt2
, img1
, img2
) == 0
261 def create_image(name
, size
):
262 '''Create a fully-allocated raw image with sector markers'''
263 file = open(name
, 'wb')
266 sector
= struct
.pack('>l504xl', i
// 512, i
// 512)
272 '''Return image's virtual size'''
273 r
= qemu_img_pipe('info', '--output=json', '-f', imgfmt
, img
)
274 return json
.loads(r
)['virtual-size']
277 return isinstance(val
, str)
279 test_dir_re
= re
.compile(r
"%s" % test_dir
)
280 def filter_test_dir(msg
):
281 return test_dir_re
.sub("TEST_DIR", msg
)
283 win32_re
= re
.compile(r
"\r")
284 def filter_win32(msg
):
285 return win32_re
.sub("", msg
)
287 qemu_io_re
= re
.compile(r
"[0-9]* ops; [0-9\/:. sec]* \([0-9\/.inf]* [EPTGMKiBbytes]*\/sec and [0-9\/.inf]* ops\/sec\)")
288 def filter_qemu_io(msg
):
289 msg
= filter_win32(msg
)
290 return qemu_io_re
.sub("X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec)", msg
)
292 chown_re
= re
.compile(r
"chown [0-9]+:[0-9]+")
293 def filter_chown(msg
):
294 return chown_re
.sub("chown UID:GID", msg
)
296 def filter_qmp_event(event
):
297 '''Filter a QMP event dict'''
299 if 'timestamp' in event
:
300 event
['timestamp']['seconds'] = 'SECS'
301 event
['timestamp']['microseconds'] = 'USECS'
304 def filter_qmp(qmsg
, filter_fn
):
305 '''Given a string filter, filter a QMP object's values.
306 filter_fn takes a (key, value) pair.'''
307 # Iterate through either lists or dicts;
308 if isinstance(qmsg
, list):
309 items
= enumerate(qmsg
)
314 if isinstance(v
, (dict, list)):
315 qmsg
[k
] = filter_qmp(v
, filter_fn
)
317 qmsg
[k
] = filter_fn(k
, v
)
320 def filter_testfiles(msg
):
321 prefix
= os
.path
.join(test_dir
, "%s-" % (os
.getpid()))
322 return msg
.replace(prefix
, 'TEST_DIR/PID-')
324 def filter_qmp_testfiles(qmsg
):
325 def _filter(_key
, value
):
327 return filter_testfiles(value
)
329 return filter_qmp(qmsg
, _filter
)
331 def filter_generated_node_ids(msg
):
332 return re
.sub("#block[0-9]+", "NODE_NAME", msg
)
334 def filter_img_info(output
, filename
):
336 for line
in output
.split('\n'):
337 if 'disk size' in line
or 'actual-size' in line
:
339 line
= line
.replace(filename
, 'TEST_IMG')
340 line
= filter_testfiles(line
)
341 line
= line
.replace(imgfmt
, 'IMGFMT')
342 line
= re
.sub('iters: [0-9]+', 'iters: XXX', line
)
343 line
= re
.sub('uuid: [-a-f0-9]+', 'uuid: XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', line
)
344 line
= re
.sub('cid: [0-9]+', 'cid: XXXXXXXXXX', line
)
346 return '\n'.join(lines
)
348 def filter_imgfmt(msg
):
349 return msg
.replace(imgfmt
, 'IMGFMT')
351 def filter_qmp_imgfmt(qmsg
):
352 def _filter(_key
, value
):
354 return filter_imgfmt(value
)
356 return filter_qmp(qmsg
, _filter
)
359 Msg
= TypeVar('Msg', Dict
[str, Any
], List
[Any
], str)
362 filters
: Iterable
[Callable
[[Msg
], Msg
]] = (),
363 indent
: Optional
[int] = None) -> None:
365 Logs either a string message or a JSON serializable message (like QMP).
366 If indent is provided, JSON serializable messages are pretty-printed.
370 if isinstance(msg
, (dict, list)):
371 # Don't sort if it's already sorted
372 do_sort
= not isinstance(msg
, OrderedDict
)
373 print(json
.dumps(msg
, sort_keys
=do_sort
, indent
=indent
))
378 def __init__(self
, seconds
, errmsg
="Timeout"):
379 self
.seconds
= seconds
382 signal
.signal(signal
.SIGALRM
, self
.timeout
)
383 signal
.setitimer(signal
.ITIMER_REAL
, self
.seconds
)
385 def __exit__(self
, exc_type
, value
, traceback
):
386 signal
.setitimer(signal
.ITIMER_REAL
, 0)
388 def timeout(self
, signum
, frame
):
389 raise Exception(self
.errmsg
)
391 def file_pattern(name
):
392 return "{0}-{1}".format(os
.getpid(), name
)
396 FilePaths is an auto-generated filename that cleans itself up.
398 Use this context manager to generate filenames and ensure that the file
401 with FilePaths(['test.img']) as img_path:
402 qemu_img('create', img_path, '1G')
403 # migration_sock_path is automatically deleted
405 def __init__(self
, names
, base_dir
=test_dir
):
408 self
.paths
.append(os
.path
.join(base_dir
, file_pattern(name
)))
413 def __exit__(self
, exc_type
, exc_val
, exc_tb
):
415 for path
in self
.paths
:
421 class FilePath(FilePaths
):
423 FilePath is a specialization of FilePaths that takes a single filename.
425 def __init__(self
, name
, base_dir
=test_dir
):
426 super(FilePath
, self
).__init
__([name
], base_dir
)
431 def file_path_remover():
432 for path
in reversed(file_path_remover
.paths
):
439 def file_path(*names
, base_dir
=test_dir
):
440 ''' Another way to get auto-generated filename that cleans itself up.
444 img_a, img_b = file_path('a.img', 'b.img')
445 sock = file_path('socket')
448 if not hasattr(file_path_remover
, 'paths'):
449 file_path_remover
.paths
= []
450 atexit
.register(file_path_remover
)
454 filename
= file_pattern(name
)
455 path
= os
.path
.join(base_dir
, filename
)
456 file_path_remover
.paths
.append(path
)
459 return paths
[0] if len(paths
) == 1 else paths
461 def remote_filename(path
):
462 if imgproto
== 'file':
464 elif imgproto
== 'ssh':
465 return "ssh://%s@127.0.0.1:22%s" % (os
.environ
.get('USER'), path
)
467 raise Exception("Protocol %s not supported" % (imgproto
))
469 class VM(qtest
.QEMUQtestMachine
):
472 def __init__(self
, path_suffix
=''):
473 name
= "qemu%s-%d" % (path_suffix
, os
.getpid())
474 super(VM
, self
).__init
__(qemu_prog
, qemu_opts
, name
=name
,
476 socket_scm_helper
=socket_scm_helper
,
480 def add_object(self
, opts
):
481 self
._args
.append('-object')
482 self
._args
.append(opts
)
485 def add_device(self
, opts
):
486 self
._args
.append('-device')
487 self
._args
.append(opts
)
490 def add_drive_raw(self
, opts
):
491 self
._args
.append('-drive')
492 self
._args
.append(opts
)
495 def add_drive(self
, path
, opts
='', interface
='virtio', img_format
=imgfmt
):
496 '''Add a virtio-blk drive to the VM'''
497 options
= ['if=%s' % interface
,
498 'id=drive%d' % self
._num
_drives
]
501 options
.append('file=%s' % path
)
502 options
.append('format=%s' % img_format
)
503 options
.append('cache=%s' % cachemode
)
504 options
.append('aio=%s' % aiomode
)
509 if img_format
== 'luks' and 'key-secret' not in opts
:
510 # default luks support
511 if luks_default_secret_object
not in self
._args
:
512 self
.add_object(luks_default_secret_object
)
514 options
.append(luks_default_key_secret_opt
)
516 self
._args
.append('-drive')
517 self
._args
.append(','.join(options
))
518 self
._num
_drives
+= 1
521 def add_blockdev(self
, opts
):
522 self
._args
.append('-blockdev')
523 if isinstance(opts
, str):
524 self
._args
.append(opts
)
526 self
._args
.append(','.join(opts
))
529 def add_incoming(self
, addr
):
530 self
._args
.append('-incoming')
531 self
._args
.append(addr
)
534 def pause_drive(self
, drive
, event
=None):
535 '''Pause drive r/w operations'''
537 self
.pause_drive(drive
, "read_aio")
538 self
.pause_drive(drive
, "write_aio")
540 self
.qmp('human-monitor-command',
541 command_line
='qemu-io %s "break %s bp_%s"' % (drive
, event
, drive
))
543 def resume_drive(self
, drive
):
544 self
.qmp('human-monitor-command',
545 command_line
='qemu-io %s "remove_break bp_%s"' % (drive
, drive
))
547 def hmp_qemu_io(self
, drive
, cmd
):
548 '''Write to a given drive using an HMP command'''
549 return self
.qmp('human-monitor-command',
550 command_line
='qemu-io %s "%s"' % (drive
, cmd
))
552 def flatten_qmp_object(self
, obj
, output
=None, basestr
=''):
555 if isinstance(obj
, list):
556 for i
, item
in enumerate(obj
):
557 self
.flatten_qmp_object(item
, output
, basestr
+ str(i
) + '.')
558 elif isinstance(obj
, dict):
560 self
.flatten_qmp_object(obj
[key
], output
, basestr
+ key
+ '.')
562 output
[basestr
[:-1]] = obj
# Strip trailing '.'
565 def qmp_to_opts(self
, obj
):
566 obj
= self
.flatten_qmp_object(obj
)
569 output_list
+= [key
+ '=' + obj
[key
]]
570 return ','.join(output_list
)
572 def get_qmp_events_filtered(self
, wait
=60.0):
574 for ev
in self
.get_qmp_events(wait
=wait
):
575 result
.append(filter_qmp_event(ev
))
578 def qmp_log(self
, cmd
, filters
=(), indent
=None, **kwargs
):
579 full_cmd
= OrderedDict((
581 ("arguments", ordered_qmp(kwargs
))
583 log(full_cmd
, filters
, indent
=indent
)
584 result
= self
.qmp(cmd
, **kwargs
)
585 log(result
, filters
, indent
=indent
)
588 # Returns None on success, and an error string on failure
589 def run_job(self
, job
, auto_finalize
=True, auto_dismiss
=False,
590 pre_finalize
=None, cancel
=False, use_log
=True, wait
=60.0):
592 run_job moves a job from creation through to dismissal.
594 :param job: String. ID of recently-launched job
595 :param auto_finalize: Bool. True if the job was launched with
596 auto_finalize. Defaults to True.
597 :param auto_dismiss: Bool. True if the job was launched with
598 auto_dismiss=True. Defaults to False.
599 :param pre_finalize: Callback. A callable that takes no arguments to be
600 invoked prior to issuing job-finalize, if any.
601 :param cancel: Bool. When true, cancels the job after the pre_finalize
603 :param use_log: Bool. When false, does not log QMP messages.
604 :param wait: Float. Timeout value specifying how long to wait for any
605 event, in seconds. Defaults to 60.0.
607 match_device
= {'data': {'device': job
}}
608 match_id
= {'data': {'id': job
}}
610 ('BLOCK_JOB_COMPLETED', match_device
),
611 ('BLOCK_JOB_CANCELLED', match_device
),
612 ('BLOCK_JOB_ERROR', match_device
),
613 ('BLOCK_JOB_READY', match_device
),
614 ('BLOCK_JOB_PENDING', match_id
),
615 ('JOB_STATUS_CHANGE', match_id
)
619 ev
= filter_qmp_event(self
.events_wait(events
, timeout
=wait
))
620 if ev
['event'] != 'JOB_STATUS_CHANGE':
624 status
= ev
['data']['status']
625 if status
== 'aborting':
626 result
= self
.qmp('query-jobs')
627 for j
in result
['return']:
631 log('Job failed: %s' % (j
['error']))
632 elif status
== 'ready':
634 self
.qmp_log('job-complete', id=job
)
636 self
.qmp('job-complete', id=job
)
637 elif status
== 'pending' and not auto_finalize
:
640 if cancel
and use_log
:
641 self
.qmp_log('job-cancel', id=job
)
643 self
.qmp('job-cancel', id=job
)
645 self
.qmp_log('job-finalize', id=job
)
647 self
.qmp('job-finalize', id=job
)
648 elif status
== 'concluded' and not auto_dismiss
:
650 self
.qmp_log('job-dismiss', id=job
)
652 self
.qmp('job-dismiss', id=job
)
653 elif status
== 'null':
656 # Returns None on success, and an error string on failure
657 def blockdev_create(self
, options
, job_id
='job0', filters
=None):
659 filters
= [filter_qmp_testfiles
]
660 result
= self
.qmp_log('blockdev-create', filters
=filters
,
661 job_id
=job_id
, options
=options
)
663 if 'return' in result
:
664 assert result
['return'] == {}
665 job_result
= self
.run_job(job_id
)
667 job_result
= result
['error']
672 def enable_migration_events(self
, name
):
673 log('Enabling migration QMP events on %s...' % name
)
674 log(self
.qmp('migrate-set-capabilities', capabilities
=[
676 'capability': 'events',
681 def wait_migration(self
, expect_runstate
):
683 event
= self
.event_wait('MIGRATION')
684 log(event
, filters
=[filter_qmp_event
])
685 if event
['data']['status'] == 'completed':
687 # The event may occur in finish-migrate, so wait for the expected
688 # post-migration runstate
689 while self
.qmp('query-status')['return']['status'] != expect_runstate
:
692 def node_info(self
, node_name
):
693 nodes
= self
.qmp('query-named-block-nodes')
694 for x
in nodes
['return']:
695 if x
['node-name'] == node_name
:
699 def query_bitmaps(self
):
700 res
= self
.qmp("query-named-block-nodes")
701 return {device
['node-name']: device
['dirty-bitmaps']
702 for device
in res
['return'] if 'dirty-bitmaps' in device
}
704 def get_bitmap(self
, node_name
, bitmap_name
, recording
=None, bitmaps
=None):
706 get a specific bitmap from the object returned by query_bitmaps.
707 :param recording: If specified, filter results by the specified value.
708 :param bitmaps: If specified, use it instead of call query_bitmaps()
711 bitmaps
= self
.query_bitmaps()
713 for bitmap
in bitmaps
[node_name
]:
714 if bitmap
.get('name', '') == bitmap_name
:
715 if recording
is None or bitmap
.get('recording') == recording
:
719 def check_bitmap_status(self
, node_name
, bitmap_name
, fields
):
720 ret
= self
.get_bitmap(node_name
, bitmap_name
)
722 return fields
.items() <= ret
.items()
724 def assert_block_path(self
, root
, path
, expected_node
, graph
=None):
726 Check whether the node under the given path in the block graph
729 @root is the node name of the node where the @path is rooted.
731 @path is a string that consists of child names separated by
732 slashes. It must begin with a slash.
734 Examples for @root + @path:
735 - root="qcow2-node", path="/backing/file"
736 - root="quorum-node", path="/children.2/file"
738 Hypothetically, @path could be empty, in which case it would
739 point to @root. However, in practice this case is not useful
740 and hence not allowed.
742 @expected_node may be None. (All elements of the path but the
743 leaf must still exist.)
745 @graph may be None or the result of an x-debug-query-block-graph
746 call that has already been performed.
749 graph
= self
.qmp('x-debug-query-block-graph')['return']
751 iter_path
= iter(path
.split('/'))
753 # Must start with a /
754 assert next(iter_path
) == ''
756 node
= next((node
for node
in graph
['nodes'] if node
['name'] == root
),
759 # An empty @path is not allowed, so the root node must be present
760 assert node
is not None, 'Root node %s not found' % root
762 for child_name
in iter_path
:
763 assert node
is not None, 'Cannot follow path %s%s' % (root
, path
)
766 node_id
= next(edge
['child'] for edge
in graph
['edges']
767 if (edge
['parent'] == node
['id'] and
768 edge
['name'] == child_name
))
770 node
= next(node
for node
in graph
['nodes']
771 if node
['id'] == node_id
)
773 except StopIteration:
777 assert expected_node
is None, \
778 'No node found under %s (but expected %s)' % \
779 (path
, expected_node
)
781 assert node
['name'] == expected_node
, \
782 'Found node %s under %s (but expected %s)' % \
783 (node
['name'], path
, expected_node
)
785 index_re
= re
.compile(r
'([^\[]+)\[([^\]]+)\]')
787 class QMPTestCase(unittest
.TestCase
):
788 '''Abstract base class for QMP test cases'''
790 def __init__(self
, *args
, **kwargs
):
791 super().__init
__(*args
, **kwargs
)
792 # Many users of this class set a VM property we rely on heavily
793 # in the methods below.
796 def dictpath(self
, d
, path
):
797 '''Traverse a path in a nested dict'''
798 for component
in path
.split('/'):
799 m
= index_re
.match(component
)
801 component
, idx
= m
.groups()
804 if not isinstance(d
, dict) or component
not in d
:
805 self
.fail('failed path traversal for "%s" in "%s"' % (path
, str(d
)))
809 if not isinstance(d
, list):
810 self
.fail('path component "%s" in "%s" is not a list in "%s"' % (component
, path
, str(d
)))
814 self
.fail('invalid index "%s" in path "%s" in "%s"' % (idx
, path
, str(d
)))
817 def assert_qmp_absent(self
, d
, path
):
819 result
= self
.dictpath(d
, path
)
820 except AssertionError:
822 self
.fail('path "%s" has value "%s"' % (path
, str(result
)))
824 def assert_qmp(self
, d
, path
, value
):
825 '''Assert that the value for a specific path in a QMP dict
826 matches. When given a list of values, assert that any of
829 result
= self
.dictpath(d
, path
)
831 # [] makes no sense as a list of valid values, so treat it as
832 # an actual single value.
833 if isinstance(value
, list) and value
!= []:
837 self
.fail('no match for "%s" in %s' % (str(result
), str(value
)))
839 self
.assertEqual(result
, value
,
840 '"%s" is "%s", expected "%s"'
841 % (path
, str(result
), str(value
)))
843 def assert_no_active_block_jobs(self
):
844 result
= self
.vm
.qmp('query-block-jobs')
845 self
.assert_qmp(result
, 'return', [])
847 def assert_has_block_node(self
, node_name
=None, file_name
=None):
848 """Issue a query-named-block-nodes and assert node_name and/or
849 file_name is present in the result"""
850 def check_equal_or_none(a
, b
):
851 return a
is None or b
is None or a
== b
852 assert node_name
or file_name
853 result
= self
.vm
.qmp('query-named-block-nodes')
854 for x
in result
["return"]:
855 if check_equal_or_none(x
.get("node-name"), node_name
) and \
856 check_equal_or_none(x
.get("file"), file_name
):
858 self
.fail("Cannot find %s %s in result:\n%s" %
859 (node_name
, file_name
, result
))
861 def assert_json_filename_equal(self
, json_filename
, reference
):
862 '''Asserts that the given filename is a json: filename and that its
863 content is equal to the given reference object'''
864 self
.assertEqual(json_filename
[:5], 'json:')
865 self
.assertEqual(self
.vm
.flatten_qmp_object(json
.loads(json_filename
[5:])),
866 self
.vm
.flatten_qmp_object(reference
))
868 def cancel_and_wait(self
, drive
='drive0', force
=False, resume
=False, wait
=60.0):
869 '''Cancel a block job and wait for it to finish, returning the event'''
870 result
= self
.vm
.qmp('block-job-cancel', device
=drive
, force
=force
)
871 self
.assert_qmp(result
, 'return', {})
874 self
.vm
.resume_drive(drive
)
879 for event
in self
.vm
.get_qmp_events(wait
=wait
):
880 if event
['event'] == 'BLOCK_JOB_COMPLETED' or \
881 event
['event'] == 'BLOCK_JOB_CANCELLED':
882 self
.assert_qmp(event
, 'data/device', drive
)
885 elif event
['event'] == 'JOB_STATUS_CHANGE':
886 self
.assert_qmp(event
, 'data/id', drive
)
889 self
.assert_no_active_block_jobs()
892 def wait_until_completed(self
, drive
='drive0', check_offset
=True, wait
=60.0,
894 '''Wait for a block job to finish, returning the event'''
896 for event
in self
.vm
.get_qmp_events(wait
=wait
):
897 if event
['event'] == 'BLOCK_JOB_COMPLETED':
898 self
.assert_qmp(event
, 'data/device', drive
)
900 self
.assert_qmp_absent(event
, 'data/error')
902 self
.assert_qmp(event
, 'data/offset',
903 event
['data']['len'])
905 self
.assert_qmp(event
, 'data/error', error
)
906 self
.assert_no_active_block_jobs()
908 if event
['event'] == 'JOB_STATUS_CHANGE':
909 self
.assert_qmp(event
, 'data/id', drive
)
911 def wait_ready(self
, drive
='drive0'):
912 """Wait until a BLOCK_JOB_READY event, and return the event."""
913 f
= {'data': {'type': 'mirror', 'device': drive
}}
914 return self
.vm
.event_wait(name
='BLOCK_JOB_READY', match
=f
)
916 def wait_ready_and_cancel(self
, drive
='drive0'):
917 self
.wait_ready(drive
=drive
)
918 event
= self
.cancel_and_wait(drive
=drive
)
919 self
.assertEqual(event
['event'], 'BLOCK_JOB_COMPLETED')
920 self
.assert_qmp(event
, 'data/type', 'mirror')
921 self
.assert_qmp(event
, 'data/offset', event
['data']['len'])
923 def complete_and_wait(self
, drive
='drive0', wait_ready
=True,
924 completion_error
=None):
925 '''Complete a block job and wait for it to finish'''
927 self
.wait_ready(drive
=drive
)
929 result
= self
.vm
.qmp('block-job-complete', device
=drive
)
930 self
.assert_qmp(result
, 'return', {})
932 event
= self
.wait_until_completed(drive
=drive
, error
=completion_error
)
933 self
.assert_qmp(event
, 'data/type', 'mirror')
935 def pause_wait(self
, job_id
='job0'):
936 with
Timeout(3, "Timeout waiting for job to pause"):
938 result
= self
.vm
.qmp('query-block-jobs')
940 for job
in result
['return']:
941 if job
['device'] == job_id
:
943 if job
['paused'] and not job
['busy']:
948 def pause_job(self
, job_id
='job0', wait
=True):
949 result
= self
.vm
.qmp('block-job-pause', device
=job_id
)
950 self
.assert_qmp(result
, 'return', {})
952 return self
.pause_wait(job_id
)
955 def case_skip(self
, reason
):
956 '''Skip this test case'''
958 self
.skipTest(reason
)
962 '''Skip this test suite'''
963 # Each test in qemu-iotests has a number ("seq")
964 seq
= os
.path
.basename(sys
.argv
[0])
966 open('%s/%s.notrun' % (output_dir
, seq
), 'w').write(reason
+ '\n')
967 print('%s not run: %s' % (seq
, reason
))
970 def case_notrun(reason
):
971 '''Mark this test case as not having been run (without actually
972 skipping it, that is left to the caller). See
973 QMPTestCase.case_skip() for a variant that actually skips the
974 current test case.'''
976 # Each test in qemu-iotests has a number ("seq")
977 seq
= os
.path
.basename(sys
.argv
[0])
979 open('%s/%s.casenotrun' % (output_dir
, seq
), 'a').write(
980 ' [case not run] ' + reason
+ '\n')
982 def verify_image_format(supported_fmts
=(), unsupported_fmts
=()):
983 assert not (supported_fmts
and unsupported_fmts
)
985 if 'generic' in supported_fmts
and \
986 os
.environ
.get('IMGFMT_GENERIC', 'true') == 'true':
988 # _supported_fmt generic
992 not_sup
= supported_fmts
and (imgfmt
not in supported_fmts
)
993 if not_sup
or (imgfmt
in unsupported_fmts
):
994 notrun('not suitable for this image format: %s' % imgfmt
)
996 def verify_protocol(supported
=(), unsupported
=()):
997 assert not (supported
and unsupported
)
999 if 'generic' in supported
:
1002 not_sup
= supported
and (imgproto
not in supported
)
1003 if not_sup
or (imgproto
in unsupported
):
1004 notrun('not suitable for this protocol: %s' % imgproto
)
1006 def verify_platform(supported
=None, unsupported
=None):
1007 if unsupported
is not None:
1008 if any((sys
.platform
.startswith(x
) for x
in unsupported
)):
1009 notrun('not suitable for this OS: %s' % sys
.platform
)
1011 if supported
is not None:
1012 if not any((sys
.platform
.startswith(x
) for x
in supported
)):
1013 notrun('not suitable for this OS: %s' % sys
.platform
)
1015 def verify_cache_mode(supported_cache_modes
=()):
1016 if supported_cache_modes
and (cachemode
not in supported_cache_modes
):
1017 notrun('not suitable for this cache mode: %s' % cachemode
)
1019 def verify_aio_mode(supported_aio_modes
=()):
1020 if supported_aio_modes
and (aiomode
not in supported_aio_modes
):
1021 notrun('not suitable for this aio mode: %s' % aiomode
)
1023 def supports_quorum():
1024 return 'quorum' in qemu_img_pipe('--help')
1026 def verify_quorum():
1027 '''Skip test suite if quorum support is not available'''
1028 if not supports_quorum():
1029 notrun('quorum support missing')
1031 def qemu_pipe(*args
):
1032 '''Run qemu with an option to print something and exit (e.g. a help option),
1033 and return its output'''
1034 args
= [qemu_prog
] + qemu_opts
+ list(args
)
1035 subp
= subprocess
.Popen(args
, stdout
=subprocess
.PIPE
,
1036 stderr
=subprocess
.STDOUT
,
1037 universal_newlines
=True)
1038 exitcode
= subp
.wait()
1040 sys
.stderr
.write('qemu received signal %i: %s\n' %
1041 (-exitcode
, ' '.join(args
)))
1042 return subp
.communicate()[0]
1044 def supported_formats(read_only
=False):
1045 '''Set 'read_only' to True to check ro-whitelist
1046 Otherwise, rw-whitelist is checked'''
1048 if not hasattr(supported_formats
, "formats"):
1049 supported_formats
.formats
= {}
1051 if read_only
not in supported_formats
.formats
:
1052 format_message
= qemu_pipe("-drive", "format=help")
1053 line
= 1 if read_only
else 0
1054 supported_formats
.formats
[read_only
] = \
1055 format_message
.splitlines()[line
].split(":")[1].split()
1057 return supported_formats
.formats
[read_only
]
1059 def skip_if_unsupported(required_formats
=(), read_only
=False):
1060 '''Skip Test Decorator
1061 Runs the test if all the required formats are whitelisted'''
1062 def skip_test_decorator(func
):
1063 def func_wrapper(test_case
: QMPTestCase
, *args
, **kwargs
):
1064 if callable(required_formats
):
1065 fmts
= required_formats(test_case
)
1067 fmts
= required_formats
1069 usf_list
= list(set(fmts
) - set(supported_formats(read_only
)))
1071 test_case
.case_skip('{}: formats {} are not whitelisted'.format(
1072 test_case
, usf_list
))
1075 return func(test_case
, *args
, **kwargs
)
1077 return skip_test_decorator
1079 def skip_if_user_is_root(func
):
1080 '''Skip Test Decorator
1081 Runs the test only without root permissions'''
1082 def func_wrapper(*args
, **kwargs
):
1083 if os
.getuid() == 0:
1084 case_notrun('{}: cannot be run as root'.format(args
[0]))
1087 return func(*args
, **kwargs
)
1090 def execute_unittest(output
, verbosity
, debug
):
1091 runner
= unittest
.TextTestRunner(stream
=output
, descriptions
=True,
1092 verbosity
=verbosity
)
1094 # unittest.main() will use sys.exit(); so expect a SystemExit
1096 unittest
.main(testRunner
=runner
)
1099 out
= output
.getvalue()
1100 out
= re
.sub(r
'Ran (\d+) tests? in [\d.]+s', r
'Ran \1 tests', out
)
1102 # Hide skipped tests from the reference output
1103 out
= re
.sub(r
'OK \(skipped=\d+\)', 'OK', out
)
1104 out_first_line
, out_rest
= out
.split('\n', 1)
1105 out
= out_first_line
.replace('s', '.') + '\n' + out_rest
1107 sys
.stderr
.write(out
)
1109 def execute_test(test_function
=None,
1111 supported_platforms
=None,
1112 supported_cache_modes
=(), supported_aio_modes
=(),
1113 unsupported_fmts
=(), supported_protocols
=(),
1114 unsupported_protocols
=()):
1115 """Run either unittest or script-style tests."""
1117 # We are using TEST_DIR and QEMU_DEFAULT_MACHINE as proxies to
1118 # indicate that we're not being run via "check". There may be
1119 # other things set up by "check" that individual test cases rely
1121 if test_dir
is None or qemu_default_machine
is None:
1122 sys
.stderr
.write('Please run this test via the "check" script\n')
1123 sys
.exit(os
.EX_USAGE
)
1125 debug
= '-d' in sys
.argv
1127 verify_image_format(supported_fmts
, unsupported_fmts
)
1128 verify_protocol(supported_protocols
, unsupported_protocols
)
1129 verify_platform(supported
=supported_platforms
)
1130 verify_cache_mode(supported_cache_modes
)
1131 verify_aio_mode(supported_aio_modes
)
1136 sys
.argv
.remove('-d')
1138 # We need to filter out the time taken from the output so that
1139 # qemu-iotest can reliably diff the results against master output.
1140 output
= io
.StringIO()
1142 logging
.basicConfig(level
=(logging
.DEBUG
if debug
else logging
.WARN
))
1144 if not test_function
:
1145 execute_unittest(output
, verbosity
, debug
)
1149 def script_main(test_function
, *args
, **kwargs
):
1150 """Run script-style tests outside of the unittest framework"""
1151 execute_test(test_function
, *args
, **kwargs
)
1153 def main(*args
, **kwargs
):
1154 """Run tests using the unittest framework"""
1155 execute_test(None, *args
, **kwargs
)