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