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 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') 28if iotests.qemu_default_machine == 's390-ccw-virtio': 29 default_virtio_blk = 'virtio-blk-ccw' 30else: 31 default_virtio_blk = 'virtio-blk-pci' 32 33class 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 363if __name__ == '__main__': 364 iotests.main(supported_fmts=["qcow2"]) 365