cirrus: fix off-by-one in cirrus_bitblt_rop_bkwd_transp_*_16
[qemu.git] / tests / qemu-iotests / 139
1 #!/usr/bin/env python
2 #
3 # Test cases for the QMP 'x-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
22 import os
23 import iotests
24 import time
25
26 base_img = os.path.join(iotests.test_dir, 'base.img')
27 new_img = os.path.join(iotests.test_dir, 'new.img')
28
29 class 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('x-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 x-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 x-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
352 if __name__ == '__main__':
353     iotests.main(supported_fmts=["qcow2"])