xref: /openbmc/qemu/tests/qemu-iotests/139 (revision e9e51b71) 
1#!/usr/bin/env python3
2# group: rw quick
3#
4# Test cases for the QMP 'blockdev-del' command
5#
6# Copyright (C) 2015 Igalia, S.L.
7# Author: Alberto Garcia <berto@igalia.com>
8#
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.
13#
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.
18#
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/>.
21#
22
23import os
24import iotests
25import time
26
27base_img = os.path.join(iotests.test_dir, 'base.img')
28new_img = os.path.join(iotests.test_dir, 'new.img')
29if iotests.qemu_default_machine == 's390-ccw-virtio':
30    default_virtio_blk = 'virtio-blk-ccw'
31else:
32    default_virtio_blk = 'virtio-blk-pci'
33
34class TestBlockdevDel(iotests.QMPTestCase):
35
36    def setUp(self):
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()))
41        self.vm.launch()
42
43    def tearDown(self):
44        self.vm.shutdown()
45        os.remove(base_img)
46        if os.path.isfile(new_img):
47            os.remove(new_img)
48
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)
55
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,
62                'node-name': node,
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)
70
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,
77                'node-name': node,
78                'backing': None,
79                'file': {'driver': 'file',
80                         'filename': new_img}}
81        result = self.vm.qmp('blockdev-add', conv_keys = False, **opts)
82        self.assert_qmp(result, 'return', {})
83        self.checkBlockDriverState(node)
84
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)
89        if expect_error:
90            self.assert_qmp(result, 'error/class', 'GenericError')
91        else:
92            self.assert_qmp(result, 'return', {})
93        self.checkBlockDriverState(node, expect_error)
94
95    # Add a device model
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', {})
100
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', {})
105
106        result = self.vm.qmp('system_reset')
107        self.assert_qmp(result, 'return', {})
108
109        if is_virtio_blk:
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)
114
115        event = self.vm.event_wait(name="DEVICE_DELETED",
116                                   match={'data': {'device': device}})
117        self.assertNotEqual(event, None)
118
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)
124        if expect_error:
125            self.assert_qmp(result, 'error/class', 'GenericError')
126            self.checkBlockDriverState(node)
127        else:
128            self.assert_qmp(result, 'return', {})
129            self.checkBlockDriverState(node, not destroys_media)
130
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)
138
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)
151
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)
161
162    # Create a mirror
163    def createMirror(self, node, new_node):
164        self.checkBlockDriverState(new_node, False)
165        opts = {'device': node,
166                'job-id': node,
167                'target': new_img,
168                'node-name': new_node,
169                'sync': 'top',
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)
174
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)
180
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,
188                 'node-name': node,
189                 'file': {'driver': 'file',
190                          'filename': base_img}}
191        opts = {'driver': 'blkdebug',
192                'node-name': debug,
193                'image': image}
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)
198
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,
208                  'node-name': test,
209                  'file': {'driver': 'file',
210                           'filename': base_img}}
211        node_1 = {'driver': iotests.imgfmt,
212                  'node-name': raw,
213                  'file': {'driver': 'file',
214                           'filename': new_img}}
215        opts = {'driver': 'blkverify',
216                'node-name': blkverify,
217                'test': node_0,
218                'raw': node_1}
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)
224
225    # Add a Quorum node
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,
232                   'node-name': child0,
233                   'file': {'driver': 'file',
234                            'filename': base_img}}
235        child_1 = {'driver': iotests.imgfmt,
236                   'node-name': child1,
237                   'file': {'driver': 'file',
238                            'filename': new_img}}
239        opts = {'driver': 'quorum',
240                'node-name': quorum,
241                'vote-threshold': 1,
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)
248
249    ########################
250    # The tests start here #
251    ########################
252
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')
258
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')
266
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')
273
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)
283
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')
296
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')
308
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')
325
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)
334
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)
345
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)
356
357
358if __name__ == '__main__':
359    iotests.main(supported_fmts=["qcow2"],
360                 supported_protocols=["file"])
361