mirrors/qemu
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
c73272f52f
qemu/tests/qemu-iotests/118
Vladimir Sementsov-Ogievskiy b6aed193e5 python: use vm.cmd() instead of vm.qmp() where appropriate 
In many cases we just want an effect of qmp command and want to raise
on failure.  Use vm.cmd() method which does exactly this.

The commit is generated by command

  git grep -l '\.qmp(' | xargs ./scripts/python_qmp_updater.py

And then, fix self.assertRaises to expect ExecuteError exception in
tests/qemu-iotests/124

Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@yandex-team.ru>
Reviewed-by: Eric Blake <eblake@redhat.com>
Message-id: 20231006154125.1068348-16-vsementsov@yandex-team.ru
Signed-off-by: John Snow <jsnow@redhat.com>
2023-10-12 14:21:44 -04:00

670 lines
25 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
# group: rw
#
# Test case for media change monitor commands
#
# Copyright (C) 2015 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 os
import stat
import time
import iotests
from iotests import qemu_img
old_img = os.path.join(iotests.test_dir, 'test0.img')
new_img = os.path.join(iotests.test_dir, 'test1.img')
def interface_to_device_name(interface):
if interface == 'ide':
return 'ide-cd'
elif interface == 'floppy':
return 'floppy'
elif interface == 'scsi':
return 'scsi-cd'
else:
return None
class ChangeBaseClass(iotests.QMPTestCase):
has_opened = False
has_closed = False
device_name = 'qdev0'
use_drive = False
def process_events(self):
for event in self.vm.get_qmp_events(wait=False):
if (event['event'] == 'DEVICE_TRAY_MOVED' and
(event['data']['device'] == 'drive0' or
event['data']['id'] == self.device_name)):
if event['data']['tray-open'] == False:
self.has_closed = True
else:
self.has_opened = True
def wait_for_open(self):
if not self.has_real_tray:
return
with iotests.Timeout(3, 'Timeout while waiting for the tray to open'):
while not self.has_opened:
self.process_events()
def wait_for_close(self):
if not self.has_real_tray:
return
with iotests.Timeout(3, 'Timeout while waiting for the tray to close'):
while not self.has_closed:
self.process_events()
class GeneralChangeTestsBaseClass(ChangeBaseClass):
def test_blockdev_change_medium(self):
self.vm.cmd('blockdev-change-medium',
id=self.device_name, filename=new_img,
format=iotests.imgfmt)
self.wait_for_open()
self.wait_for_close()
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
def test_eject(self):
self.vm.cmd('eject', id=self.device_name, force=True)
self.wait_for_open()
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', True)
self.assert_qmp_absent(result, 'return[0]/inserted')
def test_tray_eject_change(self):
self.vm.cmd('eject', id=self.device_name, force=True)
self.wait_for_open()
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', True)
self.assert_qmp_absent(result, 'return[0]/inserted')
self.vm.cmd('blockdev-change-medium', id=self.device_name,
filename=new_img, format=iotests.imgfmt)
self.wait_for_close()
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
def test_tray_open_close(self):
self.vm.cmd('blockdev-open-tray',
id=self.device_name, force=True)
self.wait_for_open()
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', True)
if self.was_empty == True:
self.assert_qmp_absent(result, 'return[0]/inserted')
else:
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
self.vm.cmd('blockdev-close-tray', id=self.device_name)
if self.has_real_tray or not self.was_empty:
self.wait_for_close()
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', False)
if self.was_empty == True:
self.assert_qmp_absent(result, 'return[0]/inserted')
else:
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
def test_tray_eject_close(self):
self.vm.cmd('eject', id=self.device_name, force=True)
self.wait_for_open()
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', True)
self.assert_qmp_absent(result, 'return[0]/inserted')
self.vm.cmd('blockdev-close-tray', id=self.device_name)
self.wait_for_close()
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', False)
self.assert_qmp_absent(result, 'return[0]/inserted')
def test_tray_open_change(self):
self.vm.cmd('blockdev-open-tray', id=self.device_name,
force=True)
self.wait_for_open()
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', True)
if self.was_empty == True:
self.assert_qmp_absent(result, 'return[0]/inserted')
else:
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
self.vm.cmd('blockdev-change-medium', id=self.device_name,
filename=new_img,
format=iotests.imgfmt)
self.wait_for_close()
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
def test_cycle(self, read_only_node=False):
self.vm.cmd('blockdev-add',
node_name='new',
driver=iotests.imgfmt,
read_only=read_only_node,
file={'filename': new_img,
'driver': 'file'})
self.vm.cmd('blockdev-open-tray',
id=self.device_name, force=True)
self.wait_for_open()
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', True)
if self.was_empty == True:
self.assert_qmp_absent(result, 'return[0]/inserted')
else:
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
self.vm.cmd('blockdev-remove-medium',
id=self.device_name)
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', True)
self.assert_qmp_absent(result, 'return[0]/inserted')
self.vm.cmd('blockdev-insert-medium',
id=self.device_name, node_name='new')
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', True)
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
self.vm.cmd('blockdev-close-tray', id=self.device_name)
self.wait_for_close()
result = self.vm.qmp('query-block')
if self.has_real_tray:
self.assert_qmp(result, 'return[0]/tray_open', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
def test_cycle_read_only_media(self):
self.test_cycle(True)
def test_close_on_closed(self):
# Should be a no-op
self.vm.cmd('blockdev-close-tray', id=self.device_name)
self.assertEqual(self.vm.get_qmp_events(wait=False), [])
def test_remove_on_closed(self):
if not self.has_real_tray:
return
result = self.vm.qmp('blockdev-remove-medium', id=self.device_name)
self.assert_qmp(result, 'error/class', 'GenericError')
def test_insert_on_closed(self):
if not self.has_real_tray:
return
self.vm.cmd('blockdev-add',
node_name='new',
driver=iotests.imgfmt,
file={'filename': new_img,
'driver': 'file'})
result = self.vm.qmp('blockdev-insert-medium', id=self.device_name,
node_name='new')
self.assert_qmp(result, 'error/class', 'GenericError')
class TestInitiallyFilled(GeneralChangeTestsBaseClass):
was_empty = False
def setUp(self):
qemu_img('create', '-f', iotests.imgfmt, old_img, '1440k')
qemu_img('create', '-f', iotests.imgfmt, new_img, '1440k')
self.vm = iotests.VM()
if self.use_drive:
self.vm.add_drive(old_img, 'media=%s' % self.media, 'none')
else:
self.vm.add_blockdev([ 'node-name=drive0',
'driver=%s' % iotests.imgfmt,
'file.driver=file',
'file.filename=%s' % old_img ])
if self.interface == 'scsi':
self.vm.add_device('virtio-scsi-pci')
self.vm.add_device('%s,drive=drive0,id=%s' %
(interface_to_device_name(self.interface),
self.device_name))
self.vm.launch()
def tearDown(self):
self.vm.shutdown()
os.remove(old_img)
os.remove(new_img)
def test_insert_on_filled(self):
self.vm.cmd('blockdev-add',
node_name='new',
driver=iotests.imgfmt,
file={'filename': new_img,
'driver': 'file'})
self.vm.cmd('blockdev-open-tray', id=self.device_name)
self.wait_for_open()
result = self.vm.qmp('blockdev-insert-medium', id=self.device_name,
node_name='new')
self.assert_qmp(result, 'error/class', 'GenericError')
class TestInitiallyEmpty(GeneralChangeTestsBaseClass):
was_empty = True
def setUp(self):
qemu_img('create', '-f', iotests.imgfmt, new_img, '1440k')
self.vm = iotests.VM()
if self.use_drive:
self.vm.add_drive(None, 'media=%s' % self.media, 'none')
if self.interface == 'scsi':
self.vm.add_device('virtio-scsi-pci')
self.vm.add_device('%s,%sid=%s' %
(interface_to_device_name(self.interface),
'drive=drive0,' if self.use_drive else '',
self.device_name))
self.vm.launch()
def tearDown(self):
self.vm.shutdown()
os.remove(new_img)
def test_remove_on_empty(self):
self.vm.cmd('blockdev-open-tray', id=self.device_name)
self.wait_for_open()
# Should be a no-op
self.vm.cmd('blockdev-remove-medium', id=self.device_name)
# Do this in a function to avoid leaking variables like case into the global
# name space (otherwise tests would be run for the abstract base classes)
def create_basic_test_classes():
for (media, interface, has_real_tray) in [ ('cdrom', 'ide', True),
('cdrom', 'scsi', True),
('disk', 'floppy', False) ]:
for case in [ TestInitiallyFilled, TestInitiallyEmpty ]:
for use_drive in [ True, False ]:
attr = { 'media': media,
'interface': interface,
'has_real_tray': has_real_tray,
'use_drive': use_drive }
name = '%s_%s_%s_%s' % (case.__name__, media, interface,
'drive' if use_drive else 'blockdev')
globals()[name] = type(name, (case, ), attr)
create_basic_test_classes()
class TestChangeReadOnly(ChangeBaseClass):
device_name = 'qdev0'
def setUp(self):
qemu_img('create', '-f', iotests.imgfmt, old_img, '1440k')
qemu_img('create', '-f', iotests.imgfmt, new_img, '1440k')
self.vm = iotests.VM()
def tearDown(self):
self.vm.shutdown()
os.chmod(old_img, 0o666)
os.chmod(new_img, 0o666)
os.remove(old_img)
os.remove(new_img)
def test_ro_ro_retain(self):
os.chmod(old_img, 0o444)
os.chmod(new_img, 0o444)
self.vm.add_drive(old_img, 'media=disk,read-only=on', 'none')
self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
self.vm.launch()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', True)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
self.vm.cmd('blockdev-change-medium', id=self.device_name,
filename=new_img,
format=iotests.imgfmt,
read_only_mode='retain')
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', True)
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
def test_ro_rw_retain(self):
os.chmod(old_img, 0o444)
self.vm.add_drive(old_img, 'media=disk,read-only=on', 'none')
self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
self.vm.launch()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', True)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
self.vm.cmd('blockdev-change-medium', id=self.device_name,
filename=new_img,
format=iotests.imgfmt,
read_only_mode='retain')
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', True)
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
@iotests.skip_if_user_is_root
def test_rw_ro_retain(self):
os.chmod(new_img, 0o444)
self.vm.add_drive(old_img, 'media=disk', 'none')
self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
self.vm.launch()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
result = self.vm.qmp('blockdev-change-medium', id=self.device_name,
filename=new_img,
format=iotests.imgfmt,
read_only_mode='retain')
self.assert_qmp(result, 'error/class', 'GenericError')
self.assertEqual(self.vm.get_qmp_events(wait=False), [])
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
def test_ro_rw(self):
os.chmod(old_img, 0o444)
self.vm.add_drive(old_img, 'media=disk,read-only=on', 'none')
self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
self.vm.launch()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', True)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
self.vm.cmd('blockdev-change-medium',
id=self.device_name,
filename=new_img,
format=iotests.imgfmt,
read_only_mode='read-write')
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
def test_rw_ro(self):
os.chmod(new_img, 0o444)
self.vm.add_drive(old_img, 'media=disk', 'none')
self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
self.vm.launch()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
self.vm.cmd('blockdev-change-medium',
id=self.device_name,
filename=new_img,
format=iotests.imgfmt,
read_only_mode='read-only')
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', True)
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
def test_make_rw_ro(self):
self.vm.add_drive(old_img, 'media=disk', 'none')
self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
self.vm.launch()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
self.vm.cmd('blockdev-change-medium',
id=self.device_name,
filename=new_img,
format=iotests.imgfmt,
read_only_mode='read-only')
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', True)
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
@iotests.skip_if_user_is_root
def test_make_ro_rw(self):
os.chmod(new_img, 0o444)
self.vm.add_drive(old_img, 'media=disk', 'none')
self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
self.vm.launch()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
result = self.vm.qmp('blockdev-change-medium',
id=self.device_name,
filename=new_img,
format=iotests.imgfmt,
read_only_mode='read-write')
self.assert_qmp(result, 'error/class', 'GenericError')
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
def test_make_rw_ro_by_retain(self):
os.chmod(old_img, 0o444)
self.vm.add_drive(old_img, 'media=disk,read-only=on', 'none')
self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
self.vm.launch()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', True)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
self.vm.cmd('blockdev-change-medium', id=self.device_name,
filename=new_img,
format=iotests.imgfmt,
read_only_mode='retain')
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', True)
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
@iotests.skip_if_user_is_root
def test_make_ro_rw_by_retain(self):
os.chmod(new_img, 0o444)
self.vm.add_drive(old_img, 'media=disk', 'none')
self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
self.vm.launch()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
result = self.vm.qmp('blockdev-change-medium', id=self.device_name,
filename=new_img,
format=iotests.imgfmt,
read_only_mode='retain')
self.assert_qmp(result, 'error/class', 'GenericError')
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
def test_rw_ro_cycle(self):
os.chmod(new_img, 0o444)
self.vm.add_drive(old_img, 'media=disk', 'none')
self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
self.vm.launch()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
self.vm.cmd('blockdev-add',
node_name='new',
driver=iotests.imgfmt,
read_only=True,
file={'filename': new_img,
'driver': 'file'})
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', False)
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
self.vm.cmd('blockdev-remove-medium', id=self.device_name)
result = self.vm.qmp('query-block')
self.assert_qmp_absent(result, 'return[0]/inserted')
self.vm.cmd('blockdev-insert-medium', id=self.device_name,
node_name='new')
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', True)
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/ro', True)
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
GeneralChangeTestsBaseClass = None
TestInitiallyFilled = None
TestInitiallyEmpty = None
class TestBlockJobsAfterCycle(ChangeBaseClass):
device_name = 'qdev0'
def setUp(self):
qemu_img('create', '-f', iotests.imgfmt, old_img, '1440K')
self.vm = iotests.VM()
self.vm.add_drive_raw("id=drive0,driver=null-co,if=none")
self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
self.vm.launch()
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/image/format', 'null-co')
# For device-less BBs, calling blockdev-open-tray or blockdev-close-tray
# is not necessary
self.vm.cmd('blockdev-remove-medium', id=self.device_name)
result = self.vm.qmp('query-block')
self.assert_qmp_absent(result, 'return[0]/inserted')
self.vm.cmd('blockdev-add',
node_name='node0',
driver=iotests.imgfmt,
file={'filename': old_img,
'driver': 'file'})
self.vm.cmd('blockdev-insert-medium', id=self.device_name,
node_name='node0')
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)
def tearDown(self):
self.vm.shutdown()
os.remove(old_img)
try:
os.remove(new_img)
except OSError:
pass
# We need backing file support
@iotests.skip_for_formats(('vpc', 'parallels', 'qcow', 'vdi', 'vmdk', 'raw',
'vhdx'))
def test_snapshot_and_commit(self):
self.vm.cmd('blockdev-snapshot-sync', device='drive0',
snapshot_file=new_img,
format=iotests.imgfmt)
result = self.vm.qmp('query-block')
self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
self.assert_qmp(result,
'return[0]/inserted/image/backing-image/filename',
old_img)
self.vm.cmd('block-commit', device='drive0')
self.vm.event_wait(name='BLOCK_JOB_READY')
result = self.vm.qmp('query-block-jobs')
self.assert_qmp(result, 'return[0]/device', 'drive0')
self.vm.cmd('block-job-complete', device='drive0')
self.vm.event_wait(name='BLOCK_JOB_COMPLETED')
if __name__ == '__main__':
if iotests.qemu_default_machine != 'pc':
# We need floppy and IDE CD-ROM
iotests.notrun('not suitable for this machine type: %s' %
iotests.qemu_default_machine)
# Need to support image creation
iotests.main(supported_fmts=['vpc', 'parallels', 'qcow', 'vdi', 'qcow2',
'vmdk', 'raw', 'vhdx', 'qed'],
supported_protocols=['file'])
Reference in New Issue View Git Blame Copy Permalink