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