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