xemu-project/xemu
1
0
Fork 0
mirror of https://github.com/xemu-project/xemu.git synced 2024-11-29 06:20:26 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity
d6f9c82c62
xemu/tests/qemu-iotests/041
Max Reitz 171d64319c qemu-iotests: Fix test 041 
Performing multiple drive-mirror blockjobs on the same qemu instance
results in the image file used for the block device being replaced by
the newly mirrored file, which is not what we want.

Fix this by performing one dedicated test per sync mode.

Signed-off-by: Max Reitz <mreitz@redhat.com>
Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>
Reviewed-by: Kevin Wolf <kwolf@redhat.com>
Reviewed-by: Eric Blake <eblake@redhat.com>
Message-id: 1385407736-13941-3-git-send-email-mreitz@redhat.com
Signed-off-by: Anthony Liguori <aliguori@amazon.com>
2013-11-27 07:53:32 -08:00

723 lines
28 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python
#
# Tests for image mirroring.
#
# Copyright (C) 2012 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import time
import os
import iotests
from iotests import qemu_img, qemu_io
backing_img = os.path.join(iotests.test_dir, 'backing.img')
target_backing_img = os.path.join(iotests.test_dir, 'target-backing.img')
test_img = os.path.join(iotests.test_dir, 'test.img')
target_img = os.path.join(iotests.test_dir, 'target.img')
class ImageMirroringTestCase(iotests.QMPTestCase):
'''Abstract base class for image mirroring test cases'''
def wait_ready(self, drive='drive0'):
'''Wait until a block job BLOCK_JOB_READY event'''
ready = False
while not ready:
for event in self.vm.get_qmp_events(wait=True):
if event['event'] == 'BLOCK_JOB_READY':
self.assert_qmp(event, 'data/type', 'mirror')
self.assert_qmp(event, 'data/device', drive)
ready = True
def wait_ready_and_cancel(self, drive='drive0'):
self.wait_ready(drive)
event = self.cancel_and_wait()
self.assertEquals(event['event'], 'BLOCK_JOB_COMPLETED')
self.assert_qmp(event, 'data/type', 'mirror')
self.assert_qmp(event, 'data/offset', self.image_len)
self.assert_qmp(event, 'data/len', self.image_len)
def complete_and_wait(self, drive='drive0', wait_ready=True):
'''Complete a block job and wait for it to finish'''
if wait_ready:
self.wait_ready()
result = self.vm.qmp('block-job-complete', device=drive)
self.assert_qmp(result, 'return', {})
event = self.wait_until_completed()
self.assert_qmp(event, 'data/type', 'mirror')
class TestSingleDrive(ImageMirroringTestCase):
image_len = 1 * 1024 * 1024 # MB
def setUp(self):
iotests.create_image(backing_img, TestSingleDrive.image_len)
qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=%s' % backing_img, test_img)
self.vm = iotests.VM().add_drive(test_img)
self.vm.launch()
def tearDown(self):
self.vm.shutdown()
os.remove(test_img)
os.remove(backing_img)
try:
os.remove(target_img)
except OSError:
pass
def test_complete(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
target=target_img)
self.assert_qmp(result, 'return', {})
self.complete_and_wait()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/file', target_img)
self.vm.shutdown()
self.assertTrue(iotests.compare_images(test_img, target_img),
'target image does not match source after mirroring')
def test_cancel(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
target=target_img)
self.assert_qmp(result, 'return', {})
self.cancel_and_wait(force=True)
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/file', test_img)
self.vm.shutdown()
def test_cancel_after_ready(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
target=target_img)
self.assert_qmp(result, 'return', {})
self.wait_ready_and_cancel()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/file', test_img)
self.vm.shutdown()
self.assertTrue(iotests.compare_images(test_img, target_img),
'target image does not match source after mirroring')
def test_pause(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
target=target_img)
self.assert_qmp(result, 'return', {})
result = self.vm.qmp('block-job-pause', device='drive0')
self.assert_qmp(result, 'return', {})
time.sleep(1)
result = self.vm.qmp('query-block-jobs')
offset = self.dictpath(result, 'return[0]/offset')
time.sleep(1)
result = self.vm.qmp('query-block-jobs')
self.assert_qmp(result, 'return[0]/offset', offset)
result = self.vm.qmp('block-job-resume', device='drive0')
self.assert_qmp(result, 'return', {})
self.complete_and_wait()
self.vm.shutdown()
self.assertTrue(iotests.compare_images(test_img, target_img),
'target image does not match source after mirroring')
def test_small_buffer(self):
self.assert_no_active_block_jobs()
# A small buffer is rounded up automatically
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
buf_size=4096, target=target_img)
self.assert_qmp(result, 'return', {})
self.complete_and_wait()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/file', target_img)
self.vm.shutdown()
self.assertTrue(iotests.compare_images(test_img, target_img),
'target image does not match source after mirroring')
def test_small_buffer2(self):
self.assert_no_active_block_jobs()
qemu_img('create', '-f', iotests.imgfmt, '-o', 'cluster_size=%d,size=%d'
% (TestSingleDrive.image_len, TestSingleDrive.image_len), target_img)
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
buf_size=65536, mode='existing', target=target_img)
self.assert_qmp(result, 'return', {})
self.complete_and_wait()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/file', target_img)
self.vm.shutdown()
self.assertTrue(iotests.compare_images(test_img, target_img),
'target image does not match source after mirroring')
def test_large_cluster(self):
self.assert_no_active_block_jobs()
qemu_img('create', '-f', iotests.imgfmt, '-o', 'cluster_size=%d,backing_file=%s'
% (TestSingleDrive.image_len, backing_img), target_img)
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
mode='existing', target=target_img)
self.assert_qmp(result, 'return', {})
self.complete_and_wait()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/file', target_img)
self.vm.shutdown()
self.assertTrue(iotests.compare_images(test_img, target_img),
'target image does not match source after mirroring')
def test_medium_not_found(self):
result = self.vm.qmp('drive-mirror', device='ide1-cd0', sync='full',
target=target_img)
self.assert_qmp(result, 'error/class', 'GenericError')
def test_image_not_found(self):
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
mode='existing', target=target_img)
self.assert_qmp(result, 'error/class', 'GenericError')
def test_device_not_found(self):
result = self.vm.qmp('drive-mirror', device='nonexistent', sync='full',
target=target_img)
self.assert_qmp(result, 'error/class', 'DeviceNotFound')
class TestMirrorNoBacking(ImageMirroringTestCase):
image_len = 2 * 1024 * 1024 # MB
def complete_and_wait(self, drive='drive0', wait_ready=True):
iotests.create_image(target_backing_img, TestMirrorNoBacking.image_len)
return ImageMirroringTestCase.complete_and_wait(self, drive, wait_ready)
def compare_images(self, img1, img2):
iotests.create_image(target_backing_img, TestMirrorNoBacking.image_len)
return iotests.compare_images(img1, img2)
def setUp(self):
iotests.create_image(backing_img, TestMirrorNoBacking.image_len)
qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=%s' % backing_img, test_img)
self.vm = iotests.VM().add_drive(test_img)
self.vm.launch()
def tearDown(self):
self.vm.shutdown()
os.remove(test_img)
os.remove(backing_img)
os.remove(target_backing_img)
os.remove(target_img)
def test_complete(self):
self.assert_no_active_block_jobs()
qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=%s' % backing_img, target_img)
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
mode='existing', target=target_img)
self.assert_qmp(result, 'return', {})
self.complete_and_wait()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/file', target_img)
self.vm.shutdown()
self.assertTrue(self.compare_images(test_img, target_img),
'target image does not match source after mirroring')
def test_cancel(self):
self.assert_no_active_block_jobs()
qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=%s' % backing_img, target_img)
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
mode='existing', target=target_img)
self.assert_qmp(result, 'return', {})
self.wait_ready_and_cancel()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/file', test_img)
self.vm.shutdown()
self.assertTrue(self.compare_images(test_img, target_img),
'target image does not match source after mirroring')
def test_large_cluster(self):
self.assert_no_active_block_jobs()
# qemu-img create fails if the image is not there
qemu_img('create', '-f', iotests.imgfmt, '-o', 'size=%d'
%(TestMirrorNoBacking.image_len), target_backing_img)
qemu_img('create', '-f', iotests.imgfmt, '-o', 'cluster_size=%d,backing_file=%s'
% (TestMirrorNoBacking.image_len, target_backing_img), target_img)
os.remove(target_backing_img)
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
mode='existing', target=target_img)
self.assert_qmp(result, 'return', {})
self.complete_and_wait()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/file', target_img)
self.vm.shutdown()
self.assertTrue(self.compare_images(test_img, target_img),
'target image does not match source after mirroring')
class TestMirrorResized(ImageMirroringTestCase):
backing_len = 1 * 1024 * 1024 # MB
image_len = 2 * 1024 * 1024 # MB
def setUp(self):
iotests.create_image(backing_img, TestMirrorResized.backing_len)
qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=%s' % backing_img, test_img)
qemu_img('resize', test_img, '2M')
self.vm = iotests.VM().add_drive(test_img)
self.vm.launch()
def tearDown(self):
self.vm.shutdown()
os.remove(test_img)
os.remove(backing_img)
try:
os.remove(target_img)
except OSError:
pass
def test_complete_top(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='top',
target=target_img)
self.assert_qmp(result, 'return', {})
self.complete_and_wait()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/file', target_img)
self.vm.shutdown()
self.assertTrue(iotests.compare_images(test_img, target_img),
'target image does not match source after mirroring')
def test_complete_full(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
target=target_img)
self.assert_qmp(result, 'return', {})
self.complete_and_wait()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/file', target_img)
self.vm.shutdown()
self.assertTrue(iotests.compare_images(test_img, target_img),
'target image does not match source after mirroring')
class TestReadErrors(ImageMirroringTestCase):
image_len = 2 * 1024 * 1024 # MB
# this should be a multiple of twice the default granularity
# so that we hit this offset first in state 1
MIRROR_GRANULARITY = 1024 * 1024
def create_blkdebug_file(self, name, event, errno):
file = open(name, 'w')
file.write('''
[inject-error]
state = "1"
event = "%s"
errno = "%d"
immediately = "off"
once = "on"
sector = "%d"
[set-state]
state = "1"
event = "%s"
new_state = "2"
[set-state]
state = "2"
event = "%s"
new_state = "1"
''' % (event, errno, self.MIRROR_GRANULARITY / 512, event, event))
file.close()
def setUp(self):
self.blkdebug_file = backing_img + ".blkdebug"
iotests.create_image(backing_img, TestReadErrors.image_len)
self.create_blkdebug_file(self.blkdebug_file, "read_aio", 5)
qemu_img('create', '-f', iotests.imgfmt,
'-o', 'backing_file=blkdebug:%s:%s,backing_fmt=raw'
% (self.blkdebug_file, backing_img),
test_img)
# Write something for tests that use sync='top'
qemu_io('-c', 'write %d 512' % (self.MIRROR_GRANULARITY + 65536),
test_img)
self.vm = iotests.VM().add_drive(test_img)
self.vm.launch()
def tearDown(self):
self.vm.shutdown()
os.remove(test_img)
os.remove(backing_img)
os.remove(self.blkdebug_file)
def test_report_read(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
target=target_img)
self.assert_qmp(result, 'return', {})
completed = False
error = False
while not completed:
for event in self.vm.get_qmp_events(wait=True):
if event['event'] == 'BLOCK_JOB_ERROR':
self.assert_qmp(event, 'data/device', 'drive0')
self.assert_qmp(event, 'data/operation', 'read')
error = True
elif event['event'] == 'BLOCK_JOB_READY':
self.assertTrue(False, 'job completed unexpectedly')
elif event['event'] == 'BLOCK_JOB_COMPLETED':
self.assertTrue(error, 'job completed unexpectedly')
self.assert_qmp(event, 'data/type', 'mirror')
self.assert_qmp(event, 'data/device', 'drive0')
self.assert_qmp(event, 'data/error', 'Input/output error')
self.assert_qmp(event, 'data/len', self.image_len)
completed = True
self.assert_no_active_block_jobs()
self.vm.shutdown()
def test_ignore_read(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
target=target_img, on_source_error='ignore')
self.assert_qmp(result, 'return', {})
event = self.vm.get_qmp_event(wait=True)
self.assertEquals(event['event'], 'BLOCK_JOB_ERROR')
self.assert_qmp(event, 'data/device', 'drive0')
self.assert_qmp(event, 'data/operation', 'read')
result = self.vm.qmp('query-block-jobs')
self.assert_qmp(result, 'return[0]/paused', False)
self.complete_and_wait()
self.vm.shutdown()
def test_large_cluster(self):
self.assert_no_active_block_jobs()
# Test COW into the target image. The first half of the
# cluster at MIRROR_GRANULARITY has to be copied from
# backing_img, even though sync='top'.
qemu_img('create', '-f', iotests.imgfmt, '-ocluster_size=131072,backing_file=%s' %(backing_img), target_img)
result = self.vm.qmp('drive-mirror', device='drive0', sync='top',
on_source_error='ignore',
mode='existing', target=target_img)
self.assert_qmp(result, 'return', {})
event = self.vm.get_qmp_event(wait=True)
self.assertEquals(event['event'], 'BLOCK_JOB_ERROR')
self.assert_qmp(event, 'data/device', 'drive0')
self.assert_qmp(event, 'data/operation', 'read')
result = self.vm.qmp('query-block-jobs')
self.assert_qmp(result, 'return[0]/paused', False)
self.complete_and_wait()
self.vm.shutdown()
# Detach blkdebug to compare images successfully
qemu_img('rebase', '-f', iotests.imgfmt, '-u', '-b', backing_img, test_img)
self.assertTrue(iotests.compare_images(test_img, target_img),
'target image does not match source after mirroring')
def test_stop_read(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
target=target_img, on_source_error='stop')
self.assert_qmp(result, 'return', {})
error = False
ready = False
while not ready:
for event in self.vm.get_qmp_events(wait=True):
if event['event'] == 'BLOCK_JOB_ERROR':
self.assert_qmp(event, 'data/device', 'drive0')
self.assert_qmp(event, 'data/operation', 'read')
result = self.vm.qmp('query-block-jobs')
self.assert_qmp(result, 'return[0]/paused', True)
self.assert_qmp(result, 'return[0]/io-status', 'failed')
result = self.vm.qmp('block-job-resume', device='drive0')
self.assert_qmp(result, 'return', {})
error = True
elif event['event'] == 'BLOCK_JOB_READY':
self.assertTrue(error, 'job completed unexpectedly')
self.assert_qmp(event, 'data/device', 'drive0')
ready = True
result = self.vm.qmp('query-block-jobs')
self.assert_qmp(result, 'return[0]/paused', False)
self.assert_qmp(result, 'return[0]/io-status', 'ok')
self.complete_and_wait(wait_ready=False)
self.assert_no_active_block_jobs()
self.vm.shutdown()
class TestWriteErrors(ImageMirroringTestCase):
image_len = 2 * 1024 * 1024 # MB
# this should be a multiple of twice the default granularity
# so that we hit this offset first in state 1
MIRROR_GRANULARITY = 1024 * 1024
def create_blkdebug_file(self, name, event, errno):
file = open(name, 'w')
file.write('''
[inject-error]
state = "1"
event = "%s"
errno = "%d"
immediately = "off"
once = "on"
sector = "%d"
[set-state]
state = "1"
event = "%s"
new_state = "2"
[set-state]
state = "2"
event = "%s"
new_state = "1"
''' % (event, errno, self.MIRROR_GRANULARITY / 512, event, event))
file.close()
def setUp(self):
self.blkdebug_file = target_img + ".blkdebug"
iotests.create_image(backing_img, TestWriteErrors.image_len)
self.create_blkdebug_file(self.blkdebug_file, "write_aio", 5)
qemu_img('create', '-f', iotests.imgfmt, '-obacking_file=%s' %(backing_img), test_img)
self.vm = iotests.VM().add_drive(test_img)
self.target_img = 'blkdebug:%s:%s' % (self.blkdebug_file, target_img)
qemu_img('create', '-f', iotests.imgfmt, '-osize=%d' %(TestWriteErrors.image_len), target_img)
self.vm.launch()
def tearDown(self):
self.vm.shutdown()
os.remove(test_img)
os.remove(backing_img)
os.remove(self.blkdebug_file)
def test_report_write(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
mode='existing', target=self.target_img)
self.assert_qmp(result, 'return', {})
completed = False
error = False
while not completed:
for event in self.vm.get_qmp_events(wait=True):
if event['event'] == 'BLOCK_JOB_ERROR':
self.assert_qmp(event, 'data/device', 'drive0')
self.assert_qmp(event, 'data/operation', 'write')
error = True
elif event['event'] == 'BLOCK_JOB_READY':
self.assertTrue(False, 'job completed unexpectedly')
elif event['event'] == 'BLOCK_JOB_COMPLETED':
self.assertTrue(error, 'job completed unexpectedly')
self.assert_qmp(event, 'data/type', 'mirror')
self.assert_qmp(event, 'data/device', 'drive0')
self.assert_qmp(event, 'data/error', 'Input/output error')
self.assert_qmp(event, 'data/len', self.image_len)
completed = True
self.assert_no_active_block_jobs()
self.vm.shutdown()
def test_ignore_write(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
mode='existing', target=self.target_img,
on_target_error='ignore')
self.assert_qmp(result, 'return', {})
event = self.vm.get_qmp_event(wait=True)
self.assertEquals(event['event'], 'BLOCK_JOB_ERROR')
self.assert_qmp(event, 'data/device', 'drive0')
self.assert_qmp(event, 'data/operation', 'write')
result = self.vm.qmp('query-block-jobs')
self.assert_qmp(result, 'return[0]/paused', False)
self.complete_and_wait()
self.vm.shutdown()
def test_stop_write(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
mode='existing', target=self.target_img,
on_target_error='stop')
self.assert_qmp(result, 'return', {})
error = False
ready = False
while not ready:
for event in self.vm.get_qmp_events(wait=True):
if event['event'] == 'BLOCK_JOB_ERROR':
self.assert_qmp(event, 'data/device', 'drive0')
self.assert_qmp(event, 'data/operation', 'write')
result = self.vm.qmp('query-block-jobs')
self.assert_qmp(result, 'return[0]/paused', True)
self.assert_qmp(result, 'return[0]/io-status', 'failed')
result = self.vm.qmp('block-job-resume', device='drive0')
self.assert_qmp(result, 'return', {})
result = self.vm.qmp('query-block-jobs')
self.assert_qmp(result, 'return[0]/paused', False)
self.assert_qmp(result, 'return[0]/io-status', 'ok')
error = True
elif event['event'] == 'BLOCK_JOB_READY':
self.assertTrue(error, 'job completed unexpectedly')
self.assert_qmp(event, 'data/device', 'drive0')
ready = True
self.complete_and_wait(wait_ready=False)
self.assert_no_active_block_jobs()
self.vm.shutdown()
class TestSetSpeed(ImageMirroringTestCase):
image_len = 80 * 1024 * 1024 # MB
def setUp(self):
qemu_img('create', backing_img, str(TestSetSpeed.image_len))
qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=%s' % backing_img, test_img)
self.vm = iotests.VM().add_drive(test_img)
self.vm.launch()
def tearDown(self):
self.vm.shutdown()
os.remove(test_img)
os.remove(backing_img)
os.remove(target_img)
def test_set_speed(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
target=target_img)
self.assert_qmp(result, 'return', {})
# Default speed is 0
result = self.vm.qmp('query-block-jobs')
self.assert_qmp(result, 'return[0]/device', 'drive0')
self.assert_qmp(result, 'return[0]/speed', 0)
result = self.vm.qmp('block-job-set-speed', device='drive0', speed=8 * 1024 * 1024)
self.assert_qmp(result, 'return', {})
# Ensure the speed we set was accepted
result = self.vm.qmp('query-block-jobs')
self.assert_qmp(result, 'return[0]/device', 'drive0')
self.assert_qmp(result, 'return[0]/speed', 8 * 1024 * 1024)
self.wait_ready_and_cancel()
# Check setting speed in drive-mirror works
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
target=target_img, speed=4*1024*1024)
self.assert_qmp(result, 'return', {})
result = self.vm.qmp('query-block-jobs')
self.assert_qmp(result, 'return[0]/device', 'drive0')
self.assert_qmp(result, 'return[0]/speed', 4 * 1024 * 1024)
self.wait_ready_and_cancel()
def test_set_speed_invalid(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
target=target_img, speed=-1)
self.assert_qmp(result, 'error/class', 'GenericError')
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0', sync='full',
target=target_img)
self.assert_qmp(result, 'return', {})
result = self.vm.qmp('block-job-set-speed', device='drive0', speed=-1)
self.assert_qmp(result, 'error/class', 'GenericError')
self.wait_ready_and_cancel()
class TestUnbackedSource(ImageMirroringTestCase):
image_len = 2 * 1024 * 1024 # MB
def setUp(self):
qemu_img('create', '-f', iotests.imgfmt, test_img,
str(TestUnbackedSource.image_len))
self.vm = iotests.VM().add_drive(test_img)
self.vm.launch()
def tearDown(self):
self.vm.shutdown()
os.remove(test_img)
os.remove(target_img)
def test_absolute_paths_full(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0',
sync='full', target=target_img,
mode='absolute-paths')
self.assert_qmp(result, 'return', {})
self.complete_and_wait()
self.assert_no_active_block_jobs()
def test_absolute_paths_top(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0',
sync='top', target=target_img,
mode='absolute-paths')
self.assert_qmp(result, 'return', {})
self.complete_and_wait()
self.assert_no_active_block_jobs()
def test_absolute_paths_none(self):
self.assert_no_active_block_jobs()
result = self.vm.qmp('drive-mirror', device='drive0',
sync='none', target=target_img,
mode='absolute-paths')
self.assert_qmp(result, 'return', {})
self.complete_and_wait()
self.assert_no_active_block_jobs()
if __name__ == '__main__':
iotests.main(supported_fmts=['qcow2', 'qed'])
Reference in New Issue View Git Blame Copy Permalink