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 22import os 23import iotests 24import time 25 26base_img = os.path.join(iotests.test_dir, 'base.img') 27new_img = os.path.join(iotests.test_dir, 'new.img') 28 29class 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 415if __name__ == '__main__': 416 iotests.main(supported_fmts=["qcow2"]) 417