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