mirrors/qemu
0
0
Fork 0
mirror of https://github.com/qemu/qemu.git synced 2025-01-22 05:23:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

qemu-iotests: extract wait_until_completed() into iotests.py

Browse Source 
The 'drive-mirror' tests often issue 'block-job-complete' and wait for
the QMP completion event.  Other types of block jobs also want to wait
for completion but they may not need to issue 'block-job-complete'.

Extract wait_until_completed() from 041 and put it into iotests.py.
Return the QMP event object so the caller can make additional
assertions, if necessary.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
Signed-off-by: Kevin Wolf <kwolf@redhat.com>
This commit is contained in:
Stefan Hajnoczi 2013-06-24 17:13:19 +02:00 committed by Kevin Wolf
parent 78b18b78aa
commit 0dbe8a1b04
2 changed files with 17 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
tests/qemu-iotests/041
View File

@ -57,18 +57,8 @@ class ImageMirroringTestCase(iotests.QMPTestCase):
result = self.vm.qmp('block-job-complete', device=drive)
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', 'mirror')
self.assert_qmp(event, 'data/device', drive)
self.assert_qmp_absent(event, 'data/error')
self.assert_qmp(event, 'data/offset', self.image_len)
self.assert_qmp(event, 'data/len', self.image_len)
completed = True
self.assert_no_active_block_jobs()
event = self.wait_until_completed()
self.assert_qmp(event, 'data/type', 'mirror')
class TestSingleDrive(ImageMirroringTestCase):
image_len = 1 * 1024 * 1024 # MB

15
tests/qemu-iotests/iotests.py
View File

@ -208,6 +208,21 @@ class QMPTestCase(unittest.TestCase):
self.assert_no_active_block_jobs()
return result
def wait_until_completed(self, drive='drive0'):
'''Wait for a block job to finish, returning the event'''
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/device', drive)
self.assert_qmp_absent(event, 'data/error')
self.assert_qmp(event, 'data/offset', self.image_len)
self.assert_qmp(event, 'data/len', self.image_len)
completed = True
self.assert_no_active_block_jobs()
return event
def notrun(reason):
'''Skip this test suite'''
# Each test in qemu-iotests has a number ("seq")