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