xref: /openbmc/qemu/tests/qemu-iotests/155 (revision a37eaa53) 
1#!/usr/bin/env python
2#
3# Test whether the backing BDSs are correct after completion of a
4# mirror block job; in "existing" modes (drive-mirror with
5# mode=existing and blockdev-mirror) the backing chain should not be
6# overridden.
7#
8# Copyright (C) 2016 Red Hat, Inc.
9#
10# This program is free software; you can redistribute it and/or modify
11# it under the terms of the GNU General Public License as published by
12# the Free Software Foundation; either version 2 of the License, or
13# (at your option) any later version.
14#
15# This program is distributed in the hope that it will be useful,
16# but WITHOUT ANY WARRANTY; without even the implied warranty of
17# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
18# GNU General Public License for more details.
19#
20# You should have received a copy of the GNU General Public License
21# along with this program.  If not, see <http://www.gnu.org/licenses/>.
22#
23
24import os
25import iotests
26from iotests import qemu_img
27
28back0_img = os.path.join(iotests.test_dir, 'back0.' + iotests.imgfmt)
29back1_img = os.path.join(iotests.test_dir, 'back1.' + iotests.imgfmt)
30back2_img = os.path.join(iotests.test_dir, 'back2.' + iotests.imgfmt)
31source_img = os.path.join(iotests.test_dir, 'source.' + iotests.imgfmt)
32target_img = os.path.join(iotests.test_dir, 'target.' + iotests.imgfmt)
33
34
35# Class variables for controlling its behavior:
36#
37# existing: If True, explicitly create the target image and blockdev-add it
38# target_backing: If existing is True: Use this filename as the backing file
39#                 of the target image
40#                 (None: no backing file)
41# target_blockdev_backing: If existing is True: Pass this dict as "backing"
42#                          for the blockdev-add command
43#                          (None: do not pass "backing")
44# target_real_backing: If existing is True: The real filename of the backing
45#                      image during runtime, only makes sense if
46#                      target_blockdev_backing is not None
47#                      (None: same as target_backing)
48
49class BaseClass(iotests.QMPTestCase):
50    target_blockdev_backing = None
51    target_real_backing = None
52
53    def setUp(self):
54        qemu_img('create', '-f', iotests.imgfmt, back0_img, '1440K')
55        qemu_img('create', '-f', iotests.imgfmt, '-b', back0_img, back1_img)
56        qemu_img('create', '-f', iotests.imgfmt, '-b', back1_img, back2_img)
57        qemu_img('create', '-f', iotests.imgfmt, '-b', back2_img, source_img)
58
59        self.vm = iotests.VM()
60        # Add the BDS via blockdev-add so it stays around after the mirror block
61        # job has been completed
62        blockdev = {'node-name': 'source',
63                    'driver': iotests.imgfmt,
64                    'file': {'driver': 'file',
65                             'filename': source_img}}
66        self.vm.add_blockdev(self.qmp_to_opts(blockdev))
67        self.vm.add_device('floppy,id=qdev0,drive=source')
68        self.vm.launch()
69
70        self.assertIntactSourceBackingChain()
71
72        if self.existing:
73            if self.target_backing:
74                qemu_img('create', '-f', iotests.imgfmt,
75                         '-b', self.target_backing, target_img, '1440K')
76            else:
77                qemu_img('create', '-f', iotests.imgfmt, target_img, '1440K')
78
79            if self.cmd == 'blockdev-mirror':
80                options = { 'node-name': 'target',
81                            'driver': iotests.imgfmt,
82                            'file': { 'driver': 'file',
83                                      'filename': target_img } }
84                if self.target_blockdev_backing:
85                    options['backing'] = self.target_blockdev_backing
86
87                result = self.vm.qmp('blockdev-add', **options)
88                self.assert_qmp(result, 'return', {})
89
90    def tearDown(self):
91        self.vm.shutdown()
92        os.remove(source_img)
93        os.remove(back2_img)
94        os.remove(back1_img)
95        os.remove(back0_img)
96        try:
97            os.remove(target_img)
98        except OSError:
99            pass
100
101    def findBlockNode(self, node_name, qdev=None):
102        if qdev:
103            result = self.vm.qmp('query-block')
104            for device in result['return']:
105                if device['qdev'] == qdev:
106                    if node_name:
107                        self.assert_qmp(device, 'inserted/node-name', node_name)
108                    return device['inserted']
109        else:
110            result = self.vm.qmp('query-named-block-nodes')
111            for node in result['return']:
112                if node['node-name'] == node_name:
113                    return node
114
115        self.fail('Cannot find node %s/%s' % (qdev, node_name))
116
117    def assertIntactSourceBackingChain(self):
118        node = self.findBlockNode('source')
119
120        self.assert_qmp(node, 'image' + '/backing-image' * 0 + '/filename',
121                        source_img)
122        self.assert_qmp(node, 'image' + '/backing-image' * 1 + '/filename',
123                        back2_img)
124        self.assert_qmp(node, 'image' + '/backing-image' * 2 + '/filename',
125                        back1_img)
126        self.assert_qmp(node, 'image' + '/backing-image' * 3 + '/filename',
127                        back0_img)
128        self.assert_qmp_absent(node, 'image' + '/backing-image' * 4)
129
130    def assertCorrectBackingImage(self, node, default_image):
131        if self.existing:
132            if self.target_real_backing:
133                image = self.target_real_backing
134            else:
135                image = self.target_backing
136        else:
137            image = default_image
138
139        if image:
140            self.assert_qmp(node, 'image/backing-image/filename', image)
141        else:
142            self.assert_qmp_absent(node, 'image/backing-image')
143
144
145# Class variables for controlling its behavior:
146#
147# cmd: Mirroring command to execute, either drive-mirror or blockdev-mirror
148
149class MirrorBaseClass(BaseClass):
150    def runMirror(self, sync):
151        if self.cmd == 'blockdev-mirror':
152            result = self.vm.qmp(self.cmd, job_id='mirror-job', device='source',
153                                 sync=sync, target='target')
154        else:
155            if self.existing:
156                mode = 'existing'
157            else:
158                mode = 'absolute-paths'
159            result = self.vm.qmp(self.cmd, job_id='mirror-job', device='source',
160                                 sync=sync, target=target_img,
161                                 format=iotests.imgfmt, mode=mode,
162                                 node_name='target')
163
164        self.assert_qmp(result, 'return', {})
165
166        self.vm.event_wait('BLOCK_JOB_READY')
167
168        result = self.vm.qmp('block-job-complete', device='mirror-job')
169        self.assert_qmp(result, 'return', {})
170
171        self.vm.event_wait('BLOCK_JOB_COMPLETED')
172
173    def testFull(self):
174        self.runMirror('full')
175
176        node = self.findBlockNode('target', 'qdev0')
177        self.assertCorrectBackingImage(node, None)
178        self.assertIntactSourceBackingChain()
179
180    def testTop(self):
181        self.runMirror('top')
182
183        node = self.findBlockNode('target', 'qdev0')
184        self.assertCorrectBackingImage(node, back2_img)
185        self.assertIntactSourceBackingChain()
186
187    def testNone(self):
188        self.runMirror('none')
189
190        node = self.findBlockNode('target', 'qdev0')
191        self.assertCorrectBackingImage(node, source_img)
192        self.assertIntactSourceBackingChain()
193
194
195class TestDriveMirrorAbsolutePaths(MirrorBaseClass):
196    cmd = 'drive-mirror'
197    existing = False
198
199class TestDriveMirrorExistingNoBacking(MirrorBaseClass):
200    cmd = 'drive-mirror'
201    existing = True
202    target_backing = None
203
204class TestDriveMirrorExistingBacking(MirrorBaseClass):
205    cmd = 'drive-mirror'
206    existing = True
207    target_backing = 'null-co://'
208
209class TestBlockdevMirrorNoBacking(MirrorBaseClass):
210    cmd = 'blockdev-mirror'
211    existing = True
212    target_backing = None
213
214class TestBlockdevMirrorBacking(MirrorBaseClass):
215    cmd = 'blockdev-mirror'
216    existing = True
217    target_backing = 'null-co://'
218
219class TestBlockdevMirrorForcedBacking(MirrorBaseClass):
220    cmd = 'blockdev-mirror'
221    existing = True
222    target_backing = None
223    target_blockdev_backing = { 'driver': 'null-co' }
224    target_real_backing = 'null-co://'
225
226
227class TestCommit(BaseClass):
228    existing = False
229
230    def testCommit(self):
231        result = self.vm.qmp('block-commit', job_id='commit-job',
232                             device='source', base=back1_img)
233        self.assert_qmp(result, 'return', {})
234
235        self.vm.event_wait('BLOCK_JOB_READY')
236
237        result = self.vm.qmp('block-job-complete', device='commit-job')
238        self.assert_qmp(result, 'return', {})
239
240        self.vm.event_wait('BLOCK_JOB_COMPLETED')
241
242        node = self.findBlockNode(None, 'qdev0')
243        self.assert_qmp(node, 'image' + '/backing-image' * 0 + '/filename',
244                        back1_img)
245        self.assert_qmp(node, 'image' + '/backing-image' * 1 + '/filename',
246                        back0_img)
247        self.assert_qmp_absent(node, 'image' + '/backing-image' * 2 +
248                               '/filename')
249
250        self.assertIntactSourceBackingChain()
251
252
253BaseClass = None
254MirrorBaseClass = None
255
256if __name__ == '__main__':
257    iotests.main(supported_fmts=['qcow2'])
258