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