xref: /openbmc/qemu/tests/qemu-iotests/155 (revision 0153d2f50bc2ad3f41810d838fcf66acbf10f07a)

#!/usr/bin/env python
#
# Test whether the backing BDSs are correct after completion of a
# mirror block job; in "existing" modes (drive-mirror with
# mode=existing and blockdev-mirror) the backing chain should not be
# overridden.
#
# Copyright (C) 2016 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

import os
import iotests
from iotests import qemu_img

back0_img = os.path.join(iotests.test_dir, 'back0.' + iotests.imgfmt)
back1_img = os.path.join(iotests.test_dir, 'back1.' + iotests.imgfmt)
back2_img = os.path.join(iotests.test_dir, 'back2.' + iotests.imgfmt)
source_img = os.path.join(iotests.test_dir, 'source.' + iotests.imgfmt)
target_img = os.path.join(iotests.test_dir, 'target.' + iotests.imgfmt)


# Class variables for controlling its behavior:
#
# existing: If True, explicitly create the target image and blockdev-add it
# target_backing: If existing is True: Use this filename as the backing file
#                 of the target image
#                 (None: no backing file)
# target_blockdev_backing: If existing is True: Pass this dict as "backing"
#                          for the blockdev-add command
#                          (None: do not pass "backing")
# target_real_backing: If existing is True: The real filename of the backing
#                      image during runtime, only makes sense if
#                      target_blockdev_backing is not None
#                      (None: same as target_backing)

class BaseClass(iotests.QMPTestCase):
    target_blockdev_backing = None
    target_real_backing = None

    def setUp(self):
        qemu_img('create', '-f', iotests.imgfmt, back0_img, '1M')
        qemu_img('create', '-f', iotests.imgfmt, '-b', back0_img, back1_img)
        qemu_img('create', '-f', iotests.imgfmt, '-b', back1_img, back2_img)
        qemu_img('create', '-f', iotests.imgfmt, '-b', back2_img, source_img)

        self.vm = iotests.VM()
        self.vm.add_drive(None, '', 'none')
        self.vm.launch()

        # Add the BDS via blockdev-add so it stays around after the mirror block
        # job has been completed
        result = self.vm.qmp('blockdev-add',
                             node_name='source',
                             driver=iotests.imgfmt,
                             file={'driver': 'file',
                                   'filename': source_img})
        self.assert_qmp(result, 'return', {})

        result = self.vm.qmp('x-blockdev-insert-medium',
                             device='drive0', node_name='source')
        self.assert_qmp(result, 'return', {})

        self.assertIntactSourceBackingChain()

        if self.existing:
            if self.target_backing:
                qemu_img('create', '-f', iotests.imgfmt,
                         '-b', self.target_backing, target_img, '1M')
            else:
                qemu_img('create', '-f', iotests.imgfmt, target_img, '1M')

            if self.cmd == 'blockdev-mirror':
                options = { 'node-name': 'target',
                            'driver': iotests.imgfmt,
                            'file': { 'driver': 'file',
                                      'filename': target_img } }
                if self.target_blockdev_backing:
                    options['backing'] = self.target_blockdev_backing

                result = self.vm.qmp('blockdev-add', **options)
                self.assert_qmp(result, 'return', {})

    def tearDown(self):
        self.vm.shutdown()
        os.remove(source_img)
        os.remove(back2_img)
        os.remove(back1_img)
        os.remove(back0_img)
        try:
            os.remove(target_img)
        except OSError:
            pass

    def findBlockNode(self, node_name, id=None):
        if id:
            result = self.vm.qmp('query-block')
            for device in result['return']:
                if device['device'] == id:
                    if node_name:
                        self.assert_qmp(device, 'inserted/node-name', node_name)
                    return device['inserted']
        else:
            result = self.vm.qmp('query-named-block-nodes')
            for node in result['return']:
                if node['node-name'] == node_name:
                    return node

        self.fail('Cannot find node %s/%s' % (id, node_name))

    def assertIntactSourceBackingChain(self):
        node = self.findBlockNode('source')

        self.assert_qmp(node, 'image' + '/backing-image' * 0 + '/filename',
                        source_img)
        self.assert_qmp(node, 'image' + '/backing-image' * 1 + '/filename',
                        back2_img)
        self.assert_qmp(node, 'image' + '/backing-image' * 2 + '/filename',
                        back1_img)
        self.assert_qmp(node, 'image' + '/backing-image' * 3 + '/filename',
                        back0_img)
        self.assert_qmp_absent(node, 'image' + '/backing-image' * 4)

    def assertCorrectBackingImage(self, node, default_image):
        if self.existing:
            if self.target_real_backing:
                image = self.target_real_backing
            else:
                image = self.target_backing
        else:
            image = default_image

        if image:
            self.assert_qmp(node, 'image/backing-image/filename', image)
        else:
            self.assert_qmp_absent(node, 'image/backing-image')


# Class variables for controlling its behavior:
#
# cmd: Mirroring command to execute, either drive-mirror or blockdev-mirror

class MirrorBaseClass(BaseClass):
    def runMirror(self, sync):
        if self.cmd == 'blockdev-mirror':
            result = self.vm.qmp(self.cmd, device='drive0', sync=sync,
                                 target='target')
        else:
            if self.existing:
                mode = 'existing'
            else:
                mode = 'absolute-paths'
            result = self.vm.qmp(self.cmd, device='drive0', sync=sync,
                                 target=target_img, format=iotests.imgfmt,
                                 mode=mode, node_name='target')

        self.assert_qmp(result, 'return', {})

        self.vm.event_wait('BLOCK_JOB_READY')

        result = self.vm.qmp('block-job-complete', device='drive0')
        self.assert_qmp(result, 'return', {})

        self.vm.event_wait('BLOCK_JOB_COMPLETED')

    def testFull(self):
        self.runMirror('full')

        node = self.findBlockNode('target', 'drive0')
        self.assertCorrectBackingImage(node, None)
        self.assertIntactSourceBackingChain()

    def testTop(self):
        self.runMirror('top')

        node = self.findBlockNode('target', 'drive0')
        self.assertCorrectBackingImage(node, back2_img)
        self.assertIntactSourceBackingChain()

    def testNone(self):
        self.runMirror('none')

        node = self.findBlockNode('target', 'drive0')
        self.assertCorrectBackingImage(node, source_img)
        self.assertIntactSourceBackingChain()


class TestDriveMirrorAbsolutePaths(MirrorBaseClass):
    cmd = 'drive-mirror'
    existing = False

class TestDriveMirrorExistingNoBacking(MirrorBaseClass):
    cmd = 'drive-mirror'
    existing = True
    target_backing = None

class TestDriveMirrorExistingBacking(MirrorBaseClass):
    cmd = 'drive-mirror'
    existing = True
    target_backing = 'null-co://'

class TestBlockdevMirrorNoBacking(MirrorBaseClass):
    cmd = 'blockdev-mirror'
    existing = True
    target_backing = None

class TestBlockdevMirrorBacking(MirrorBaseClass):
    cmd = 'blockdev-mirror'
    existing = True
    target_backing = 'null-co://'

class TestBlockdevMirrorForcedBacking(MirrorBaseClass):
    cmd = 'blockdev-mirror'
    existing = True
    target_backing = None
    target_blockdev_backing = { 'driver': 'null-co' }
    target_real_backing = 'null-co://'


class TestCommit(BaseClass):
    existing = False

    def testCommit(self):
        result = self.vm.qmp('block-commit', device='drive0', base=back1_img)
        self.assert_qmp(result, 'return', {})

        self.vm.event_wait('BLOCK_JOB_READY')

        result = self.vm.qmp('block-job-complete', device='drive0')
        self.assert_qmp(result, 'return', {})

        self.vm.event_wait('BLOCK_JOB_COMPLETED')

        node = self.findBlockNode(None, 'drive0')
        self.assert_qmp(node, 'image' + '/backing-image' * 0 + '/filename',
                        back1_img)
        self.assert_qmp(node, 'image' + '/backing-image' * 1 + '/filename',
                        back0_img)
        self.assert_qmp_absent(node, 'image' + '/backing-image' * 2 +
                               '/filename')

        self.assertIntactSourceBackingChain()


BaseClass = None
MirrorBaseClass = None

if __name__ == '__main__':
    iotests.main(supported_fmts=['qcow2'])