tests/qemu-iotests/139

   1 #!/usr/bin/env python3
   2 #
   3 # Test cases for the QMP 'blockdev-del' command
   4 #
   5 # Copyright (C) 2015 Igalia, S.L.
   6 # Author: Alberto Garcia <berto@igalia.com>
   7 #
   8 # This program is free software; you can redistribute it and/or modify
   9 # it under the terms of the GNU General Public License as published by
  10 # the Free Software Foundation; either version 2 of the License, or
  11 # (at your option) any later version.
  12 #
  13 # This program is distributed in the hope that it will be useful,
  14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16 # GNU General Public License for more details.
  17 #
  18 # You should have received a copy of the GNU General Public License
  19 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  20 #
  21
  22 import os
  23 import iotests
  24 import time
  25
  26 base_img = os.path.join(iotests.test_dir, 'base.img')
  27 new_img = os.path.join(iotests.test_dir, 'new.img')
  28 if iotests.qemu_default_machine == 's390-ccw-virtio':
  29     default_virtio_blk = 'virtio-blk-ccw'
  30 else:
  31     default_virtio_blk = 'virtio-blk-pci'
  32
  33 class TestBlockdevDel(iotests.QMPTestCase):
  34
  35     def setUp(self):
  36         iotests.qemu_img('create', '-f', iotests.imgfmt, base_img, '1M')
  37         self.vm = iotests.VM()
  38         self.vm.add_device("{},id=virtio-scsi".format(
  39             iotests.get_virtio_scsi_device()))
  40         self.vm.launch()
  41
  42     def tearDown(self):
  43         self.vm.shutdown()
  44         os.remove(base_img)
  45         if os.path.isfile(new_img):
  46             os.remove(new_img)
  47
  48     # Check whether a BlockDriverState exists
  49     def checkBlockDriverState(self, node, must_exist = True):
  50         result = self.vm.qmp('query-named-block-nodes')
  51         nodes = [x for x in result['return'] if x['node-name'] == node]
  52         self.assertLessEqual(len(nodes), 1)
  53         self.assertEqual(must_exist, len(nodes) == 1)
  54
  55     # Add a BlockDriverState without a BlockBackend
  56     def addBlockDriverState(self, node):
  57         file_node = '%s_file' % node
  58         self.checkBlockDriverState(node, False)
  59         self.checkBlockDriverState(file_node, False)
  60         opts = {'driver': iotests.imgfmt,
  61                 'node-name': node,
  62                 'file': {'driver': 'file',
  63                          'node-name': file_node,
  64                          'filename': base_img}}
  65         result = self.vm.qmp('blockdev-add', conv_keys = False, **opts)
  66         self.assert_qmp(result, 'return', {})
  67         self.checkBlockDriverState(node)
  68         self.checkBlockDriverState(file_node)
  69
  70     # Add a BlockDriverState that will be used as overlay for the base_img BDS
  71     def addBlockDriverStateOverlay(self, node):
  72         self.checkBlockDriverState(node, False)
  73         iotests.qemu_img('create', '-u', '-f', iotests.imgfmt,
  74                          '-b', base_img, '-F', iotests.imgfmt, new_img, '1M')
  75         opts = {'driver': iotests.imgfmt,
  76                 'node-name': node,
  77                 'backing': None,
  78                 'file': {'driver': 'file',
  79                          'filename': new_img}}
  80         result = self.vm.qmp('blockdev-add', conv_keys = False, **opts)
  81         self.assert_qmp(result, 'return', {})
  82         self.checkBlockDriverState(node)
  83
  84     # Delete a BlockDriverState
  85     def delBlockDriverState(self, node, expect_error = False):
  86         self.checkBlockDriverState(node)
  87         result = self.vm.qmp('blockdev-del', node_name = node)
  88         if expect_error:
  89             self.assert_qmp(result, 'error/class', 'GenericError')
  90         else:
  91             self.assert_qmp(result, 'return', {})
  92         self.checkBlockDriverState(node, expect_error)
  93
  94     # Add a device model
  95     def addDeviceModel(self, device, backend, driver = default_virtio_blk):
  96         result = self.vm.qmp('device_add', id = device,
  97                              driver = driver, drive = backend)
  98         self.assert_qmp(result, 'return', {})
  99
 100     # Delete a device model
 101     def delDeviceModel(self, device, is_virtio_blk = True):
 102         result = self.vm.qmp('device_del', id = device)
 103         self.assert_qmp(result, 'return', {})
 104
 105         result = self.vm.qmp('system_reset')
 106         self.assert_qmp(result, 'return', {})
 107
 108         if is_virtio_blk:
 109             device_path = '/machine/peripheral/%s/virtio-backend' % device
 110             event = self.vm.event_wait(name="DEVICE_DELETED",
 111                                        match={'data': {'path': device_path}})
 112             self.assertNotEqual(event, None)
 113
 114         event = self.vm.event_wait(name="DEVICE_DELETED",
 115                                    match={'data': {'device': device}})
 116         self.assertNotEqual(event, None)
 117
 118     # Remove a BlockDriverState
 119     def ejectDrive(self, device, node, expect_error = False,
 120                    destroys_media = True):
 121         self.checkBlockDriverState(node)
 122         result = self.vm.qmp('eject', id = device)
 123         if expect_error:
 124             self.assert_qmp(result, 'error/class', 'GenericError')
 125             self.checkBlockDriverState(node)
 126         else:
 127             self.assert_qmp(result, 'return', {})
 128             self.checkBlockDriverState(node, not destroys_media)
 129
 130     # Insert a BlockDriverState
 131     def insertDrive(self, device, node):
 132         self.checkBlockDriverState(node)
 133         result = self.vm.qmp('blockdev-insert-medium',
 134                              id = device, node_name = node)
 135         self.assert_qmp(result, 'return', {})
 136         self.checkBlockDriverState(node)
 137
 138     # Create a snapshot using 'blockdev-snapshot-sync'
 139     def createSnapshotSync(self, node, overlay):
 140         self.checkBlockDriverState(node)
 141         self.checkBlockDriverState(overlay, False)
 142         opts = {'node-name': node,
 143                 'snapshot-file': new_img,
 144                 'snapshot-node-name': overlay,
 145                 'format': iotests.imgfmt}
 146         result = self.vm.qmp('blockdev-snapshot-sync', conv_keys=False, **opts)
 147         self.assert_qmp(result, 'return', {})
 148         self.checkBlockDriverState(node)
 149         self.checkBlockDriverState(overlay)
 150
 151     # Create a snapshot using 'blockdev-snapshot'
 152     def createSnapshot(self, node, overlay):
 153         self.checkBlockDriverState(node)
 154         self.checkBlockDriverState(overlay)
 155         result = self.vm.qmp('blockdev-snapshot',
 156                              node = node, overlay = overlay)
 157         self.assert_qmp(result, 'return', {})
 158         self.checkBlockDriverState(node)
 159         self.checkBlockDriverState(overlay)
 160
 161     # Create a mirror
 162     def createMirror(self, node, new_node):
 163         self.checkBlockDriverState(new_node, False)
 164         opts = {'device': node,
 165                 'job-id': node,
 166                 'target': new_img,
 167                 'node-name': new_node,
 168                 'sync': 'top',
 169                 'format': iotests.imgfmt}
 170         result = self.vm.qmp('drive-mirror', conv_keys=False, **opts)
 171         self.assert_qmp(result, 'return', {})
 172         self.checkBlockDriverState(new_node)
 173
 174     # Complete an existing block job
 175     def completeBlockJob(self, id, node_before, node_after):
 176         result = self.vm.qmp('block-job-complete', device=id)
 177         self.assert_qmp(result, 'return', {})
 178         self.wait_until_completed(id)
 179
 180     # Add a BlkDebug node
 181     # Note that the purpose of this is to test the blockdev-del
 182     # sanity checks, not to create a usable blkdebug drive
 183     def addBlkDebug(self, debug, node):
 184         self.checkBlockDriverState(node, False)
 185         self.checkBlockDriverState(debug, False)
 186         image = {'driver': iotests.imgfmt,
 187                  'node-name': node,
 188                  'file': {'driver': 'file',
 189                           'filename': base_img}}
 190         opts = {'driver': 'blkdebug',
 191                 'node-name': debug,
 192                 'image': image}
 193         result = self.vm.qmp('blockdev-add', conv_keys = False, **opts)
 194         self.assert_qmp(result, 'return', {})
 195         self.checkBlockDriverState(node)
 196         self.checkBlockDriverState(debug)
 197
 198     # Add a BlkVerify node
 199     # Note that the purpose of this is to test the blockdev-del
 200     # sanity checks, not to create a usable blkverify drive
 201     def addBlkVerify(self, blkverify, test, raw):
 202         self.checkBlockDriverState(test, False)
 203         self.checkBlockDriverState(raw, False)
 204         self.checkBlockDriverState(blkverify, False)
 205         iotests.qemu_img('create', '-f', iotests.imgfmt, new_img, '1M')
 206         node_0 = {'driver': iotests.imgfmt,
 207                   'node-name': test,
 208                   'file': {'driver': 'file',
 209                            'filename': base_img}}
 210         node_1 = {'driver': iotests.imgfmt,
 211                   'node-name': raw,
 212                   'file': {'driver': 'file',
 213                            'filename': new_img}}
 214         opts = {'driver': 'blkverify',
 215                 'node-name': blkverify,
 216                 'test': node_0,
 217                 'raw': node_1}
 218         result = self.vm.qmp('blockdev-add', conv_keys = False, **opts)
 219         self.assert_qmp(result, 'return', {})
 220         self.checkBlockDriverState(test)
 221         self.checkBlockDriverState(raw)
 222         self.checkBlockDriverState(blkverify)
 223
 224     # Add a Quorum node
 225     def addQuorum(self, quorum, child0, child1):
 226         self.checkBlockDriverState(child0, False)
 227         self.checkBlockDriverState(child1, False)
 228         self.checkBlockDriverState(quorum, False)
 229         iotests.qemu_img('create', '-f', iotests.imgfmt, new_img, '1M')
 230         child_0 = {'driver': iotests.imgfmt,
 231                    'node-name': child0,
 232                    'file': {'driver': 'file',
 233                             'filename': base_img}}
 234         child_1 = {'driver': iotests.imgfmt,
 235                    'node-name': child1,
 236                    'file': {'driver': 'file',
 237                             'filename': new_img}}
 238         opts = {'driver': 'quorum',
 239                 'node-name': quorum,
 240                 'vote-threshold': 1,
 241                 'children': [ child_0, child_1 ]}
 242         result = self.vm.qmp('blockdev-add', conv_keys = False, **opts)
 243         self.assert_qmp(result, 'return', {})
 244         self.checkBlockDriverState(child0)
 245         self.checkBlockDriverState(child1)
 246         self.checkBlockDriverState(quorum)
 247
 248     ########################
 249     # The tests start here #
 250     ########################
 251
 252     def testBlockDriverState(self):
 253         self.addBlockDriverState('node0')
 254         # You cannot delete a file BDS directly
 255         self.delBlockDriverState('node0_file', expect_error = True)
 256         self.delBlockDriverState('node0')
 257
 258     def testDeviceModel(self):
 259         self.addBlockDriverState('node0')
 260         self.addDeviceModel('device0', 'node0')
 261         self.ejectDrive('device0', 'node0', expect_error = True)
 262         self.delBlockDriverState('node0', expect_error = True)
 263         self.delDeviceModel('device0')
 264         self.delBlockDriverState('node0')
 265
 266     def testAttachMedia(self):
 267         # This creates a BlockBackend and removes its media
 268         self.addBlockDriverState('node0')
 269         self.addDeviceModel('device0', 'node0', 'scsi-cd')
 270         self.ejectDrive('device0', 'node0', destroys_media = False)
 271         self.delBlockDriverState('node0')
 272
 273         # This creates a new BlockDriverState and inserts it into the device
 274         self.addBlockDriverState('node1')
 275         self.insertDrive('device0', 'node1')
 276         # The node can't be removed: the new device has an extra reference
 277         self.delBlockDriverState('node1', expect_error = True)
 278         # The BDS still exists after being ejected, but now it can be removed
 279         self.ejectDrive('device0', 'node1', destroys_media = False)
 280         self.delBlockDriverState('node1')
 281         self.delDeviceModel('device0', False)
 282
 283     def testSnapshotSync(self):
 284         self.addBlockDriverState('node0')
 285         self.addDeviceModel('device0', 'node0')
 286         self.createSnapshotSync('node0', 'overlay0')
 287         # This fails because node0 is now being used as a backing image
 288         self.delBlockDriverState('node0', expect_error = True)
 289         self.delBlockDriverState('overlay0', expect_error = True)
 290         # This succeeds because device0 only has the backend reference
 291         self.delDeviceModel('device0')
 292         # FIXME Would still be there if blockdev-snapshot-sync took a ref
 293         self.checkBlockDriverState('overlay0', False)
 294         self.delBlockDriverState('node0')
 295
 296     def testSnapshot(self):
 297         self.addBlockDriverState('node0')
 298         self.addDeviceModel('device0', 'node0', 'scsi-cd')
 299         self.addBlockDriverStateOverlay('overlay0')
 300         self.createSnapshot('node0', 'overlay0')
 301         self.delBlockDriverState('node0', expect_error = True)
 302         self.delBlockDriverState('overlay0', expect_error = True)
 303         self.ejectDrive('device0', 'overlay0', destroys_media = False)
 304         self.delBlockDriverState('node0', expect_error = True)
 305         self.delBlockDriverState('overlay0')
 306         self.delBlockDriverState('node0')
 307
 308     def testMirror(self):
 309         self.addBlockDriverState('node0')
 310         self.addDeviceModel('device0', 'node0', 'scsi-cd')
 311         self.createMirror('node0', 'mirror0')
 312         # The block job prevents removing the device
 313         self.delBlockDriverState('node0', expect_error = True)
 314         self.delBlockDriverState('mirror0', expect_error = True)
 315         self.wait_ready('node0')
 316         self.completeBlockJob('node0', 'node0', 'mirror0')
 317         self.assert_no_active_block_jobs()
 318         # This succeeds because the device now points to mirror0
 319         self.delBlockDriverState('node0')
 320         self.delBlockDriverState('mirror0', expect_error = True)
 321         self.delDeviceModel('device0', False)
 322         # FIXME mirror0 disappears, drive-mirror doesn't take a reference
 323         #self.delBlockDriverState('mirror0')
 324
 325     @iotests.skip_if_unsupported(['blkdebug'])
 326     def testBlkDebug(self):
 327         self.addBlkDebug('debug0', 'node0')
 328         # 'node0' is used by the blkdebug node
 329         self.delBlockDriverState('node0', expect_error = True)
 330         # But we can remove the blkdebug node directly
 331         self.delBlockDriverState('debug0')
 332         self.checkBlockDriverState('node0', False)
 333
 334     @iotests.skip_if_unsupported(['blkverify'])
 335     def testBlkVerify(self):
 336         self.addBlkVerify('verify0', 'node0', 'node1')
 337         # We cannot remove the children of a blkverify device
 338         self.delBlockDriverState('node0', expect_error = True)
 339         self.delBlockDriverState('node1', expect_error = True)
 340         # But we can remove the blkverify node directly
 341         self.delBlockDriverState('verify0')
 342         self.checkBlockDriverState('node0', False)
 343         self.checkBlockDriverState('node1', False)
 344
 345     @iotests.skip_if_unsupported(['quorum'])
 346     def testQuorum(self):
 347         self.addQuorum('quorum0', 'node0', 'node1')
 348         # We cannot remove the children of a Quorum device
 349         self.delBlockDriverState('node0', expect_error = True)
 350         self.delBlockDriverState('node1', expect_error = True)
 351         # But we can remove the Quorum node directly
 352         self.delBlockDriverState('quorum0')
 353         self.checkBlockDriverState('node0', False)
 354         self.checkBlockDriverState('node1', False)
 355
 356
 357 if __name__ == '__main__':
 358     iotests.main(supported_fmts=["qcow2"],
 359                  supported_protocols=["file"])