xref: /openbmc/qemu/tests/qemu-iotests/139 (revision 2d7ac0af) 
1#!/usr/bin/env python3
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        self.vm.add_device("{},id=virtio-scsi".format(
39            iotests.get_virtio_scsi_device()))
40        self.vm.launch()
41
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)
47
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)
54
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)
69
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)
83
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)
93
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', {})
99
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', {})
104
105        result = self.vm.qmp('system_reset')
106        self.assert_qmp(result, 'return', {})
107
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)
113
114        event = self.vm.event_wait(name="DEVICE_DELETED",
115                                   match={'data': {'device': device}})
116        self.assertNotEqual(event, None)
117
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)
129
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)
137
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)
150
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)
160
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)
173
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)
179
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)
197
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)
223
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)
247
248    ########################
249    # The tests start here #
250    ########################
251
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')
257
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')
265
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')
272
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)
282
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')
295
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')
307
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')
324
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)
333
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)
344
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)
355
356
357if __name__ == '__main__':
358    iotests.main(supported_fmts=["qcow2"],
359                 supported_protocols=["file"])
360