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