xref: /openbmc/qemu/tests/qemu-iotests/139 (revision 2bfe11c8) 
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
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')
28
29class 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, options = 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, options = 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, options = 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, options = 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, options = 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 'quorum' in iotests.qemu_img_pipe('--help'):
340            return
341        self.addQuorum('quorum0', 'node0', 'node1')
342        # We cannot remove the children of a Quorum device
343        self.delBlockDriverState('node0', expect_error = True)
344        self.delBlockDriverState('node1', expect_error = True)
345        # But we can remove the Quorum node directly
346        self.delBlockDriverState('quorum0')
347        self.checkBlockDriverState('node0', False)
348        self.checkBlockDriverState('node1', False)
349
350
351if __name__ == '__main__':
352    iotests.main(supported_fmts=["qcow2"])
353