tests/qemu-iotests/139

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