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