haraldwolff/qemu-patch-raspberry4
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity
qemu-patch-raspberry4/tests/qemu-iotests/030
Paolo Bonzini 0fd05e8dd1 qemu-iotests: start vms in qtest mode 
This way, they will not execute any VM code at all.  However, right now
the cancellation test is "relying" on being slowed down by TCG executing
BIOS code.  So, change the timeouts.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Kevin Wolf <kwolf@redhat.com>
2012-06-15 14:03:43 +02:00

244 lines
8.7 KiB
Python
Executable file
Raw Blame History

#!/usr/bin/env python
#
# Tests for image streaming.
#
# Copyright (C) 2012 IBM Corp.
#
# 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 os
import iotests
from iotests import qemu_img, qemu_io
import struct
backing_img = os.path.join(iotests.test_dir, 'backing.img')
mid_img = os.path.join(iotests.test_dir, 'mid.img')
test_img = os.path.join(iotests.test_dir, 'test.img')
class ImageStreamingTestCase(iotests.QMPTestCase):
'''Abstract base class for image streaming test cases'''
def assert_no_active_streams(self):
result = self.vm.qmp('query-block-jobs')
self.assert_qmp(result, 'return', [])
def cancel_and_wait(self, drive='drive0'):
'''Cancel a block job and wait for it to finish'''
result = self.vm.qmp('block-job-cancel', device=drive)
self.assert_qmp(result, 'return', {})
cancelled = False
while not cancelled:
for event in self.vm.get_qmp_events(wait=True):
if event['event'] == 'BLOCK_JOB_CANCELLED':
self.assert_qmp(event, 'data/type', 'stream')
self.assert_qmp(event, 'data/device', drive)
cancelled = True
self.assert_no_active_streams()
def create_image(self, name, size):
file = open(name, 'w')
i = 0
while i < size:
sector = struct.pack('>l504xl', i / 512, i / 512)
file.write(sector)
i = i + 512
file.close()
class TestSingleDrive(ImageStreamingTestCase):
image_len = 1 * 1024 * 1024 # MB
def setUp(self):
self.create_image(backing_img, TestSingleDrive.image_len)
qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=%s' % backing_img, mid_img)
qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=%s' % mid_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(mid_img)
os.remove(backing_img)
def test_stream(self):
self.assert_no_active_streams()
result = self.vm.qmp('block-stream', device='drive0')
self.assert_qmp(result, 'return', {})
completed = False
while not completed:
for event in self.vm.get_qmp_events(wait=True):
if event['event'] == 'BLOCK_JOB_COMPLETED':
self.assert_qmp(event, 'data/type', 'stream')
self.assert_qmp(event, 'data/device', 'drive0')
self.assert_qmp(event, 'data/offset', self.image_len)
self.assert_qmp(event, 'data/len', self.image_len)
completed = True
self.assert_no_active_streams()
self.vm.shutdown()
self.assertEqual(qemu_io('-c', 'map', backing_img),
qemu_io('-c', 'map', test_img),
'image file map does not match backing file after streaming')
def test_stream_partial(self):
self.assert_no_active_streams()
result = self.vm.qmp('block-stream', device='drive0', base=mid_img)
self.assert_qmp(result, 'return', {})
completed = False
while not completed:
for event in self.vm.get_qmp_events(wait=True):
if event['event'] == 'BLOCK_JOB_COMPLETED':
self.assert_qmp(event, 'data/type', 'stream')
self.assert_qmp(event, 'data/device', 'drive0')
self.assert_qmp(event, 'data/offset', self.image_len)
self.assert_qmp(event, 'data/len', self.image_len)
completed = True
self.assert_no_active_streams()
self.vm.shutdown()
self.assertEqual(qemu_io('-c', 'map', mid_img),
qemu_io('-c', 'map', test_img),
'image file map does not match backing file after streaming')
def test_device_not_found(self):
result = self.vm.qmp('block-stream', device='nonexistent')
self.assert_qmp(result, 'error/class', 'DeviceNotFound')
class TestStreamStop(ImageStreamingTestCase):
image_len = 8 * 1024 * 1024 * 1024 # GB
def setUp(self):
qemu_img('create', backing_img, str(TestStreamStop.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)
def test_stream_stop(self):
import time
self.assert_no_active_streams()
result = self.vm.qmp('block-stream', device='drive0')
self.assert_qmp(result, 'return', {})
time.sleep(0.1)
events = self.vm.get_qmp_events(wait=False)
self.assertEqual(events, [], 'unexpected QMP event: %s' % events)
self.cancel_and_wait()
class TestSetSpeed(ImageStreamingTestCase):
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)
# This is a short performance test which is not run by default.
# Invoke "IMGFMT=qed ./030 TestSetSpeed.perf_test_throughput"
def perf_test_throughput(self):
self.assert_no_active_streams()
result = self.vm.qmp('block-stream', device='drive0')
self.assert_qmp(result, 'return', {})
result = self.vm.qmp('block-job-set-speed', device='drive0', speed=8 * 1024 * 1024)
self.assert_qmp(result, 'return', {})
completed = False
while not completed:
for event in self.vm.get_qmp_events(wait=True):
if event['event'] == 'BLOCK_JOB_COMPLETED':
self.assert_qmp(event, 'data/type', 'stream')
self.assert_qmp(event, 'data/device', 'drive0')
self.assert_qmp(event, 'data/offset', self.image_len)
self.assert_qmp(event, 'data/len', self.image_len)
completed = True
self.assert_no_active_streams()
def test_set_speed(self):
self.assert_no_active_streams()
result = self.vm.qmp('block-stream', device='drive0')
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.cancel_and_wait()
# Check setting speed in block-stream works
result = self.vm.qmp('block-stream', device='drive0', 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.cancel_and_wait()
def test_set_speed_invalid(self):
self.assert_no_active_streams()
result = self.vm.qmp('block-stream', device='drive0', speed=-1)
self.assert_qmp(result, 'error/class', 'InvalidParameter')
self.assert_qmp(result, 'error/data/name', 'speed')
self.assert_no_active_streams()
result = self.vm.qmp('block-stream', device='drive0')
self.assert_qmp(result, 'return', {})
result = self.vm.qmp('block-job-set-speed', device='drive0', speed=-1)
self.assert_qmp(result, 'error/class', 'InvalidParameter')
self.assert_qmp(result, 'error/data/name', 'speed')
self.cancel_and_wait()
if __name__ == '__main__':
iotests.main(supported_fmts=['qcow2', 'qed'])
Reference in a new issue View git blame Copy permalink