tests/qemu-iotests/139

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