diff options
Diffstat (limited to 'tests/qemu-iotests/139')
| -rwxr-xr-x | tests/qemu-iotests/139 | 355 |
1 files changed, 355 insertions, 0 deletions
diff --git a/tests/qemu-iotests/139 b/tests/qemu-iotests/139 new file mode 100755 index 000000000..178b1ee23 --- /dev/null +++ b/tests/qemu-iotests/139 @@ -0,0 +1,355 @@ +#!/usr/bin/env python3 +# group: rw quick +# +# Test cases for the QMP 'blockdev-del' command +# +# Copyright (C) 2015 Igalia, S.L. +# Author: Alberto Garcia <berto@igalia.com> +# +# 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 +import time + +base_img = os.path.join(iotests.test_dir, 'base.img') +new_img = os.path.join(iotests.test_dir, 'new.img') + +class TestBlockdevDel(iotests.QMPTestCase): + + def setUp(self): + iotests.qemu_img('create', '-f', iotests.imgfmt, base_img, '1M') + self.vm = iotests.VM() + self.vm.add_device("{},id=virtio-scsi".format('virtio-scsi')) + self.vm.launch() + + def tearDown(self): + self.vm.shutdown() + os.remove(base_img) + if os.path.isfile(new_img): + os.remove(new_img) + + # Check whether a BlockDriverState exists + def checkBlockDriverState(self, node, must_exist = True): + result = self.vm.qmp('query-named-block-nodes') + nodes = [x for x in result['return'] if x['node-name'] == node] + self.assertLessEqual(len(nodes), 1) + self.assertEqual(must_exist, len(nodes) == 1) + + # Add a BlockDriverState without a BlockBackend + def addBlockDriverState(self, node): + file_node = '%s_file' % node + self.checkBlockDriverState(node, False) + self.checkBlockDriverState(file_node, False) + opts = {'driver': iotests.imgfmt, + 'node-name': node, + 'file': {'driver': 'file', + 'node-name': file_node, + 'filename': base_img}} + result = self.vm.qmp('blockdev-add', conv_keys = False, **opts) + self.assert_qmp(result, 'return', {}) + self.checkBlockDriverState(node) + self.checkBlockDriverState(file_node) + + # Add a BlockDriverState that will be used as overlay for the base_img BDS + def addBlockDriverStateOverlay(self, node): + self.checkBlockDriverState(node, False) + iotests.qemu_img('create', '-u', '-f', iotests.imgfmt, + '-b', base_img, '-F', iotests.imgfmt, new_img, '1M') + opts = {'driver': iotests.imgfmt, + 'node-name': node, + 'backing': None, + 'file': {'driver': 'file', + 'filename': new_img}} + result = self.vm.qmp('blockdev-add', conv_keys = False, **opts) + self.assert_qmp(result, 'return', {}) + self.checkBlockDriverState(node) + + # Delete a BlockDriverState + def delBlockDriverState(self, node, expect_error = False): + self.checkBlockDriverState(node) + result = self.vm.qmp('blockdev-del', node_name = node) + if expect_error: + self.assert_qmp(result, 'error/class', 'GenericError') + else: + self.assert_qmp(result, 'return', {}) + self.checkBlockDriverState(node, expect_error) + + # Add a device model + def addDeviceModel(self, device, backend, driver = 'virtio-blk'): + result = self.vm.qmp('device_add', id = device, + driver = driver, drive = backend) + self.assert_qmp(result, 'return', {}) + + # Delete a device model + def delDeviceModel(self, device, is_virtio_blk = True): + result = self.vm.qmp('device_del', id = device) + self.assert_qmp(result, 'return', {}) + + result = self.vm.qmp('system_reset') + self.assert_qmp(result, 'return', {}) + + if is_virtio_blk: + device_path = '/machine/peripheral/%s/virtio-backend' % device + event = self.vm.event_wait(name="DEVICE_DELETED", + match={'data': {'path': device_path}}) + self.assertNotEqual(event, None) + + event = self.vm.event_wait(name="DEVICE_DELETED", + match={'data': {'device': device}}) + self.assertNotEqual(event, None) + + # Remove a BlockDriverState + def ejectDrive(self, device, node, expect_error = False, + destroys_media = True): + self.checkBlockDriverState(node) + result = self.vm.qmp('eject', id = device) + if expect_error: + self.assert_qmp(result, 'error/class', 'GenericError') + self.checkBlockDriverState(node) + else: + self.assert_qmp(result, 'return', {}) + self.checkBlockDriverState(node, not destroys_media) + + # Insert a BlockDriverState + def insertDrive(self, device, node): + self.checkBlockDriverState(node) + result = self.vm.qmp('blockdev-insert-medium', + id = device, node_name = node) + self.assert_qmp(result, 'return', {}) + self.checkBlockDriverState(node) + + # Create a snapshot using 'blockdev-snapshot-sync' + def createSnapshotSync(self, node, overlay): + self.checkBlockDriverState(node) + self.checkBlockDriverState(overlay, False) + opts = {'node-name': node, + 'snapshot-file': new_img, + 'snapshot-node-name': overlay, + 'format': iotests.imgfmt} + result = self.vm.qmp('blockdev-snapshot-sync', conv_keys=False, **opts) + self.assert_qmp(result, 'return', {}) + self.checkBlockDriverState(node) + self.checkBlockDriverState(overlay) + + # Create a snapshot using 'blockdev-snapshot' + def createSnapshot(self, node, overlay): + self.checkBlockDriverState(node) + self.checkBlockDriverState(overlay) + result = self.vm.qmp('blockdev-snapshot', + node = node, overlay = overlay) + self.assert_qmp(result, 'return', {}) + self.checkBlockDriverState(node) + self.checkBlockDriverState(overlay) + + # Create a mirror + def createMirror(self, node, new_node): + self.checkBlockDriverState(new_node, False) + opts = {'device': node, + 'job-id': node, + 'target': new_img, + 'node-name': new_node, + 'sync': 'top', + 'format': iotests.imgfmt} + result = self.vm.qmp('drive-mirror', conv_keys=False, **opts) + self.assert_qmp(result, 'return', {}) + self.checkBlockDriverState(new_node) + + # Complete an existing block job + def completeBlockJob(self, id, node_before, node_after): + result = self.vm.qmp('block-job-complete', device=id) + self.assert_qmp(result, 'return', {}) + self.wait_until_completed(id) + + # Add a BlkDebug node + # Note that the purpose of this is to test the blockdev-del + # sanity checks, not to create a usable blkdebug drive + def addBlkDebug(self, debug, node): + self.checkBlockDriverState(node, False) + self.checkBlockDriverState(debug, False) + image = {'driver': iotests.imgfmt, + 'node-name': node, + 'file': {'driver': 'file', + 'filename': base_img}} + opts = {'driver': 'blkdebug', + 'node-name': debug, + 'image': image} + result = self.vm.qmp('blockdev-add', conv_keys = False, **opts) + self.assert_qmp(result, 'return', {}) + self.checkBlockDriverState(node) + self.checkBlockDriverState(debug) + + # Add a BlkVerify node + # Note that the purpose of this is to test the blockdev-del + # sanity checks, not to create a usable blkverify drive + def addBlkVerify(self, blkverify, test, raw): + self.checkBlockDriverState(test, False) + self.checkBlockDriverState(raw, False) + self.checkBlockDriverState(blkverify, False) + iotests.qemu_img('create', '-f', iotests.imgfmt, new_img, '1M') + node_0 = {'driver': iotests.imgfmt, + 'node-name': test, + 'file': {'driver': 'file', + 'filename': base_img}} + node_1 = {'driver': iotests.imgfmt, + 'node-name': raw, + 'file': {'driver': 'file', + 'filename': new_img}} + opts = {'driver': 'blkverify', + 'node-name': blkverify, + 'test': node_0, + 'raw': node_1} + result = self.vm.qmp('blockdev-add', conv_keys = False, **opts) + self.assert_qmp(result, 'return', {}) + self.checkBlockDriverState(test) + self.checkBlockDriverState(raw) + self.checkBlockDriverState(blkverify) + + # Add a Quorum node + def addQuorum(self, quorum, child0, child1): + self.checkBlockDriverState(child0, False) + self.checkBlockDriverState(child1, False) + self.checkBlockDriverState(quorum, False) + iotests.qemu_img('create', '-f', iotests.imgfmt, new_img, '1M') + child_0 = {'driver': iotests.imgfmt, + 'node-name': child0, + 'file': {'driver': 'file', + 'filename': base_img}} + child_1 = {'driver': iotests.imgfmt, + 'node-name': child1, + 'file': {'driver': 'file', + 'filename': new_img}} + opts = {'driver': 'quorum', + 'node-name': quorum, + 'vote-threshold': 1, + 'children': [ child_0, child_1 ]} + result = self.vm.qmp('blockdev-add', conv_keys = False, **opts) + self.assert_qmp(result, 'return', {}) + self.checkBlockDriverState(child0) + self.checkBlockDriverState(child1) + self.checkBlockDriverState(quorum) + + ######################## + # The tests start here # + ######################## + + def testBlockDriverState(self): + self.addBlockDriverState('node0') + # You cannot delete a file BDS directly + self.delBlockDriverState('node0_file', expect_error = True) + self.delBlockDriverState('node0') + + def testDeviceModel(self): + self.addBlockDriverState('node0') + self.addDeviceModel('device0', 'node0') + self.ejectDrive('device0', 'node0', expect_error = True) + self.delBlockDriverState('node0', expect_error = True) + self.delDeviceModel('device0') + self.delBlockDriverState('node0') + + def testAttachMedia(self): + # This creates a BlockBackend and removes its media + self.addBlockDriverState('node0') + self.addDeviceModel('device0', 'node0', 'scsi-cd') + self.ejectDrive('device0', 'node0', destroys_media = False) + self.delBlockDriverState('node0') + + # This creates a new BlockDriverState and inserts it into the device + self.addBlockDriverState('node1') + self.insertDrive('device0', 'node1') + # The node can't be removed: the new device has an extra reference + self.delBlockDriverState('node1', expect_error = True) + # The BDS still exists after being ejected, but now it can be removed + self.ejectDrive('device0', 'node1', destroys_media = False) + self.delBlockDriverState('node1') + self.delDeviceModel('device0', False) + + def testSnapshotSync(self): + self.addBlockDriverState('node0') + self.addDeviceModel('device0', 'node0') + self.createSnapshotSync('node0', 'overlay0') + # This fails because node0 is now being used as a backing image + self.delBlockDriverState('node0', expect_error = True) + self.delBlockDriverState('overlay0', expect_error = True) + # This succeeds because device0 only has the backend reference + self.delDeviceModel('device0') + # FIXME Would still be there if blockdev-snapshot-sync took a ref + self.checkBlockDriverState('overlay0', False) + self.delBlockDriverState('node0') + + def testSnapshot(self): + self.addBlockDriverState('node0') + self.addDeviceModel('device0', 'node0', 'scsi-cd') + self.addBlockDriverStateOverlay('overlay0') + self.createSnapshot('node0', 'overlay0') + self.delBlockDriverState('node0', expect_error = True) + self.delBlockDriverState('overlay0', expect_error = True) + self.ejectDrive('device0', 'overlay0', destroys_media = False) + self.delBlockDriverState('node0', expect_error = True) + self.delBlockDriverState('overlay0') + self.delBlockDriverState('node0') + + def testMirror(self): + self.addBlockDriverState('node0') + self.addDeviceModel('device0', 'node0', 'scsi-cd') + self.createMirror('node0', 'mirror0') + # The block job prevents removing the device + self.delBlockDriverState('node0', expect_error = True) + self.delBlockDriverState('mirror0', expect_error = True) + self.wait_ready('node0') + self.completeBlockJob('node0', 'node0', 'mirror0') + self.assert_no_active_block_jobs() + # This succeeds because the device now points to mirror0 + self.delBlockDriverState('node0') + self.delBlockDriverState('mirror0', expect_error = True) + self.delDeviceModel('device0', False) + # FIXME mirror0 disappears, drive-mirror doesn't take a reference + #self.delBlockDriverState('mirror0') + + @iotests.skip_if_unsupported(['blkdebug']) + def testBlkDebug(self): + self.addBlkDebug('debug0', 'node0') + # 'node0' is used by the blkdebug node + self.delBlockDriverState('node0', expect_error = True) + # But we can remove the blkdebug node directly + self.delBlockDriverState('debug0') + self.checkBlockDriverState('node0', False) + + @iotests.skip_if_unsupported(['blkverify']) + def testBlkVerify(self): + self.addBlkVerify('verify0', 'node0', 'node1') + # We cannot remove the children of a blkverify device + self.delBlockDriverState('node0', expect_error = True) + self.delBlockDriverState('node1', expect_error = True) + # But we can remove the blkverify node directly + self.delBlockDriverState('verify0') + self.checkBlockDriverState('node0', False) + self.checkBlockDriverState('node1', False) + + @iotests.skip_if_unsupported(['quorum']) + def testQuorum(self): + self.addQuorum('quorum0', 'node0', 'node1') + # We cannot remove the children of a Quorum device + self.delBlockDriverState('node0', expect_error = True) + self.delBlockDriverState('node1', expect_error = True) + # But we can remove the Quorum node directly + self.delBlockDriverState('quorum0') + self.checkBlockDriverState('node0', False) + self.checkBlockDriverState('node1', False) + + +if __name__ == '__main__': + iotests.main(supported_fmts=["qcow2"], + supported_protocols=["file"]) |