3 # Test cases for the QMP 'blockdev-del' command
5 # Copyright (C) 2015 Igalia, S.L.
6 # Author: Alberto Garcia <berto@igalia.com>
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.
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.
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/>.
26 base_img = os.path.join(iotests.test_dir, 'base.img')
27 new_img = os.path.join(iotests.test_dir, 'new.img')
28 if iotests.qemu_default_machine == 's390-ccw-virtio':
29 default_virtio_blk = 'virtio-blk-ccw'
31 default_virtio_blk = 'virtio-blk-pci'
33 class TestBlockdevDel(iotests.QMPTestCase):
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()))
45 if os.path.isfile(new_img):
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)
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,
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)
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,
78 'file': {'driver': 'file',
80 result = self.vm.qmp('blockdev-add', conv_keys = False, **opts)
81 self.assert_qmp(result, 'return', {})
82 self.checkBlockDriverState(node)
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)
89 self.assert_qmp(result, 'error/class', 'GenericError')
91 self.assert_qmp(result, 'return', {})
92 self.checkBlockDriverState(node, expect_error)
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', {})
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', {})
105 result = self.vm.qmp('system_reset')
106 self.assert_qmp(result, 'return', {})
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)
114 event = self.vm.event_wait(name="DEVICE_DELETED",
115 match={'data': {'device': device}})
116 self.assertNotEqual(event, None)
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)
124 self.assert_qmp(result, 'error/class', 'GenericError')
125 self.checkBlockDriverState(node)
127 self.assert_qmp(result, 'return', {})
128 self.checkBlockDriverState(node, not destroys_media)
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)
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)
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)
162 def createMirror(self, node, new_node):
163 self.checkBlockDriverState(new_node, False)
164 opts = {'device': node,
167 'node-name': new_node,
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)
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)
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,
188 'file': {'driver': 'file',
189 'filename': base_img}}
190 opts = {'driver': 'blkdebug',
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)
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,
208 'file': {'driver': 'file',
209 'filename': base_img}}
210 node_1 = {'driver': iotests.imgfmt,
212 'file': {'driver': 'file',
213 'filename': new_img}}
214 opts = {'driver': 'blkverify',
215 'node-name': blkverify,
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)
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,
232 'file': {'driver': 'file',
233 'filename': base_img}}
234 child_1 = {'driver': iotests.imgfmt,
236 'file': {'driver': 'file',
237 'filename': new_img}}
238 opts = {'driver': 'quorum',
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)
248 ########################
249 # The tests start here #
250 ########################
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')
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')
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')
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)
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')
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')
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')
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)
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)
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)
357 if __name__ == '__main__':
358 iotests.main(supported_fmts=["qcow2"],
359 supported_protocols=["file"])