#!/usr/bin/env python3
+# group: rw
#
# Tests for drive-backup and blockdev-backup
#
def setUp(self):
qemu_img('create', '-f', iotests.imgfmt, blockdev_target_img, str(image_len))
- self.vm = iotests.VM().add_drive('blkdebug::' + test_img)
- self.vm.add_drive(blockdev_target_img, interface="none")
+ self.vm = iotests.VM()
+ self.vm.add_drive('blkdebug::' + test_img, 'node-name=source')
+ self.vm.add_drive(blockdev_target_img, 'node-name=target',
+ interface="none")
if iotests.qemu_default_machine == 'pc':
self.vm.add_drive(None, 'media=cdrom', 'ide')
self.vm.launch()
def test_pause_blockdev_backup(self):
self.do_test_pause('blockdev-backup', 'drive1', blockdev_target_img)
+ def do_test_resize_blockdev_backup(self, device, node):
+ def pre_finalize():
+ result = self.vm.qmp('block_resize', device=device, size=65536)
+ self.assert_qmp(result, 'error/class', 'GenericError')
+
+ result = self.vm.qmp('block_resize', node_name=node, size=65536)
+ self.assert_qmp(result, 'error/class', 'GenericError')
+
+ result = self.vm.qmp('blockdev-backup', job_id='job0', device='drive0',
+ target='drive1', sync='full', auto_finalize=False,
+ auto_dismiss=False)
+ self.assert_qmp(result, 'return', {})
+
+ self.vm.run_job('job0', auto_finalize=False, pre_finalize=pre_finalize)
+
+ def test_source_resize_blockdev_backup(self):
+ self.do_test_resize_blockdev_backup('drive0', 'source')
+
+ def test_target_resize_blockdev_backup(self):
+ self.do_test_resize_blockdev_backup('drive1', 'target')
+
+ def do_test_target_size(self, size):
+ result = self.vm.qmp('block_resize', device='drive1', size=size)
+ self.assert_qmp(result, 'return', {})
+
+ result = self.vm.qmp('blockdev-backup', job_id='job0', device='drive0',
+ target='drive1', sync='full')
+ self.assert_qmp(result, 'error/class', 'GenericError')
+
+ def test_small_target(self):
+ self.do_test_target_size(image_len // 2)
+
+ def test_large_target(self):
+ self.do_test_target_size(image_len * 2)
+
def test_medium_not_found(self):
if iotests.qemu_default_machine != 'pc':
return
self.assert_no_active_block_jobs()
-class TestDriveCompression(iotests.QMPTestCase):
+class TestCompressedToQcow2(iotests.QMPTestCase):
image_len = 64 * 1024 * 1024 # MB
- fmt_supports_compression = [{'type': 'qcow2', 'args': ()},
- {'type': 'vmdk', 'args': ('-o', 'subformat=streamOptimized')}]
+ target_fmt = {'type': 'qcow2', 'args': (), 'drive-opts': ''}
def tearDown(self):
self.vm.shutdown()
except OSError:
pass
- def do_prepare_drives(self, fmt, args, attach_target):
- self.vm = iotests.VM().add_drive('blkdebug::' + test_img)
+ def do_prepare_drives(self, attach_target):
+ self.vm = iotests.VM().add_drive('blkdebug::' + test_img,
+ opts=self.target_fmt['drive-opts'])
- qemu_img('create', '-f', fmt, blockdev_target_img,
- str(TestDriveCompression.image_len), *args)
+ qemu_img('create', '-f', self.target_fmt['type'], blockdev_target_img,
+ str(self.image_len), *self.target_fmt['args'])
if attach_target:
self.vm.add_drive(blockdev_target_img,
- img_format=fmt, interface="none")
+ img_format=self.target_fmt['type'],
+ interface="none",
+ opts=self.target_fmt['drive-opts'])
self.vm.launch()
- def do_test_compress_complete(self, cmd, format, attach_target, **args):
- self.do_prepare_drives(format['type'], format['args'], attach_target)
+ def do_test_compress_complete(self, cmd, attach_target, **args):
+ self.do_prepare_drives(attach_target)
self.assert_no_active_block_jobs()
self.vm.shutdown()
self.assertTrue(iotests.compare_images(test_img, blockdev_target_img,
- iotests.imgfmt, format['type']),
+ iotests.imgfmt,
+ self.target_fmt['type']),
'target image does not match source after backup')
def test_complete_compress_drive_backup(self):
- for format in TestDriveCompression.fmt_supports_compression:
- self.do_test_compress_complete('drive-backup', format, False,
- target=blockdev_target_img, mode='existing')
+ self.do_test_compress_complete('drive-backup', False,
+ target=blockdev_target_img,
+ mode='existing')
def test_complete_compress_blockdev_backup(self):
- for format in TestDriveCompression.fmt_supports_compression:
- self.do_test_compress_complete('blockdev-backup', format, True,
- target='drive1')
+ self.do_test_compress_complete('blockdev-backup',
+ True, target='drive1')
- def do_test_compress_cancel(self, cmd, format, attach_target, **args):
- self.do_prepare_drives(format['type'], format['args'], attach_target)
+ def do_test_compress_cancel(self, cmd, attach_target, **args):
+ self.do_prepare_drives(attach_target)
self.assert_no_active_block_jobs()
self.vm.shutdown()
def test_compress_cancel_drive_backup(self):
- for format in TestDriveCompression.fmt_supports_compression:
- self.do_test_compress_cancel('drive-backup', format, False,
- target=blockdev_target_img, mode='existing')
+ self.do_test_compress_cancel('drive-backup', False,
+ target=blockdev_target_img,
+ mode='existing')
def test_compress_cancel_blockdev_backup(self):
- for format in TestDriveCompression.fmt_supports_compression:
- self.do_test_compress_cancel('blockdev-backup', format, True,
- target='drive1')
+ self.do_test_compress_cancel('blockdev-backup', True,
+ target='drive1')
- def do_test_compress_pause(self, cmd, format, attach_target, **args):
- self.do_prepare_drives(format['type'], format['args'], attach_target)
+ def do_test_compress_pause(self, cmd, attach_target, **args):
+ self.do_prepare_drives(attach_target)
self.assert_no_active_block_jobs()
self.vm.shutdown()
self.assertTrue(iotests.compare_images(test_img, blockdev_target_img,
- iotests.imgfmt, format['type']),
+ iotests.imgfmt,
+ self.target_fmt['type']),
'target image does not match source after backup')
def test_compress_pause_drive_backup(self):
- for format in TestDriveCompression.fmt_supports_compression:
- self.do_test_compress_pause('drive-backup', format, False,
- target=blockdev_target_img, mode='existing')
+ self.do_test_compress_pause('drive-backup', False,
+ target=blockdev_target_img,
+ mode='existing')
def test_compress_pause_blockdev_backup(self):
- for format in TestDriveCompression.fmt_supports_compression:
- self.do_test_compress_pause('blockdev-backup', format, True,
- target='drive1')
+ self.do_test_compress_pause('blockdev-backup', True,
+ target='drive1')
+
+
+class TestCompressedToVmdk(TestCompressedToQcow2):
+ target_fmt = {'type': 'vmdk', 'args': ('-o', 'subformat=streamOptimized'),
+ 'drive-opts': 'cache.no-flush=on'}
+
+ @iotests.skip_if_unsupported(['vmdk'])
+ def setUp(self):
+ pass
+
if __name__ == '__main__':
iotests.main(supported_fmts=['raw', 'qcow2'],
projects / mirror_qemu.git / blobdiff