qapi/error: Check format string argument in error_*prepend()
[qemu/armbru.git] / tests / qemu-iotests / 139
blob1452fd24b335ce045f5a7f48dc96eb9d50c66cd7
1 #!/usr/bin/env python3
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/>.
22 import os
23 import iotests
24 import time
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'
30 else:
31     default_virtio_blk = 'virtio-blk-pci'
33 class TestBlockdevDel(iotests.QMPTestCase):
35     def setUp(self):
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()))
40         self.vm.launch()
42     def tearDown(self):
43         self.vm.shutdown()
44         os.remove(base_img)
45         if os.path.isfile(new_img):
46             os.remove(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,
61                 'node-name': node,
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,
76                 'node-name': node,
77                 'backing': None,
78                 'file': {'driver': 'file',
79                          'filename': new_img}}
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)
88         if expect_error:
89             self.assert_qmp(result, 'error/class', 'GenericError')
90         else:
91             self.assert_qmp(result, 'return', {})
92         self.checkBlockDriverState(node, expect_error)
94     # Add a device model
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', {})
108         if is_virtio_blk:
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)
123         if expect_error:
124             self.assert_qmp(result, 'error/class', 'GenericError')
125             self.checkBlockDriverState(node)
126         else:
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)
161     # Create a mirror
162     def createMirror(self, node, new_node):
163         self.checkBlockDriverState(new_node, False)
164         opts = {'device': node,
165                 'job-id': node,
166                 'target': new_img,
167                 'node-name': new_node,
168                 'sync': 'top',
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,
187                  'node-name': node,
188                  'file': {'driver': 'file',
189                           'filename': base_img}}
190         opts = {'driver': 'blkdebug',
191                 'node-name': debug,
192                 'image': image}
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,
207                   'node-name': test,
208                   'file': {'driver': 'file',
209                            'filename': base_img}}
210         node_1 = {'driver': iotests.imgfmt,
211                   'node-name': raw,
212                   'file': {'driver': 'file',
213                            'filename': new_img}}
214         opts = {'driver': 'blkverify',
215                 'node-name': blkverify,
216                 'test': node_0,
217                 'raw': node_1}
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)
224     # Add a Quorum node
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,
231                    'node-name': child0,
232                    'file': {'driver': 'file',
233                             'filename': base_img}}
234         child_1 = {'driver': iotests.imgfmt,
235                    'node-name': child1,
236                    'file': {'driver': 'file',
237                             'filename': new_img}}
238         opts = {'driver': 'quorum',
239                 'node-name': quorum,
240                 'vote-threshold': 1,
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"])