xref: /openbmc/qemu/tests/qemu-iotests/057 (revision fef80ea073c4862bc9eaddb6ddb0ed970b8ad7c4)

#!/usr/bin/env python3
# group: rw
#
# Tests for internal snapshot.
#
# Copyright (C) 2013 IBM, Inc.
#
# Based on 055.
#
# 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 time
import os
import iotests
from iotests import qemu_img, qemu_io

test_drv_base_name = 'drive'

class ImageSnapshotTestCase(iotests.QMPTestCase):
    image_len = 120 * 1024 * 1024 # MB

    def __init__(self, *args):
        self.expect = []
        super(ImageSnapshotTestCase, self).__init__(*args)

    def _setUp(self, test_img_base_name, image_num):
        self.vm = iotests.VM()
        for i in range(0, image_num):
            filename = '%s%d' % (test_img_base_name, i)
            img = os.path.join(iotests.test_dir, filename)
            device = '%s%d' % (test_drv_base_name, i)
            qemu_img('create', '-f', iotests.imgfmt, img, str(self.image_len))
            self.vm.add_drive(img)
            self.expect.append({'image': img, 'device': device,
                                'snapshots': [],
                                'snapshots_name_counter': 0})
        self.vm.launch()

    def tearDown(self):
        self.vm.shutdown()
        for dev_expect in self.expect:
            os.remove(dev_expect['image'])

    def createSnapshotInTransaction(self, snapshot_num, abort = False):
        actions = []
        for dev_expect in self.expect:
            num = dev_expect['snapshots_name_counter']
            for j in range(0, snapshot_num):
                name = '%s_sn%d' % (dev_expect['device'], num)
                num = num + 1
                if abort == False:
                    dev_expect['snapshots'].append({'name': name})
                    dev_expect['snapshots_name_counter'] = num
                actions.append({
                    'type': 'blockdev-snapshot-internal-sync',
                    'data': { 'device': dev_expect['device'],
                              'name': name },
                })

        if abort == True:
            actions.append({
                'type': 'abort',
                'data': {},
            })

        result = self.vm.qmp('transaction', actions = actions)

        if abort == True:
            self.assert_qmp(result, 'error/class', 'GenericError')
        else:
            self.assert_qmp(result, 'return', {})

    def verifySnapshotInfo(self):
        result = self.vm.qmp('query-block')

        # Verify each expected result
        for dev_expect in self.expect:
            # 1. Find the returned image value and snapshot info
            image_result = None
            for device in result['return']:
                if device['device'] == dev_expect['device']:
                    image_result = device['inserted']['image']
                    break
            self.assertTrue(image_result != None)
            # Do not consider zero snapshot case now
            sn_list_result = image_result['snapshots']
            sn_list_expect = dev_expect['snapshots']

            # 2. Verify it with expect
            self.assertTrue(len(sn_list_result) == len(sn_list_expect))

            for sn_expect in sn_list_expect:
                sn_result = None
                for sn in sn_list_result:
                    if sn_expect['name'] == sn['name']:
                        sn_result = sn
                        break
                self.assertTrue(sn_result != None)
                # Fill in the detail info
                sn_expect.update(sn_result)

    def deleteSnapshot(self, device, id = None, name = None):
        sn_list_expect = None
        sn_expect = None

        self.assertTrue(id != None or name != None)

        # Fill in the detail info include ID
        self.verifySnapshotInfo()

        #find the expected snapshot list
        for dev_expect in self.expect:
            if dev_expect['device'] == device:
                sn_list_expect = dev_expect['snapshots']
                break
        self.assertTrue(sn_list_expect != None)

        if id != None and name != None:
            for sn in sn_list_expect:
                if sn['id'] == id and sn['name'] == name:
                    sn_expect = sn
                    result = \
                          self.vm.qmp('blockdev-snapshot-delete-internal-sync',
                                      device = device,
                                      id = id,
                                      name = name)
                    break
        elif id != None:
            for sn in sn_list_expect:
                if sn['id'] == id:
                    sn_expect = sn
                    result = \
                          self.vm.qmp('blockdev-snapshot-delete-internal-sync',
                                      device = device,
                                      id = id)
                    break
        else:
            for sn in sn_list_expect:
                if sn['name'] == name:
                    sn_expect = sn
                    result = \
                          self.vm.qmp('blockdev-snapshot-delete-internal-sync',
                                      device = device,
                                      name = name)
                    break

        self.assertTrue(sn_expect != None)

        self.assert_qmp(result, 'return', sn_expect)
        sn_list_expect.remove(sn_expect)

class TestSingleTransaction(ImageSnapshotTestCase):
    def setUp(self):
        self._setUp('test_a.img', 1)

    def test_create(self):
        self.createSnapshotInTransaction(1)
        self.verifySnapshotInfo()

    def test_error_name_empty(self):
        actions = [{'type': 'blockdev-snapshot-internal-sync',
                    'data': { 'device': self.expect[0]['device'],
                              'name': '' },
                  }]
        result = self.vm.qmp('transaction', actions = actions)
        self.assert_qmp(result, 'error/class', 'GenericError')

    def test_error_device(self):
        actions = [{'type': 'blockdev-snapshot-internal-sync',
                    'data': { 'device': 'drive_error',
                              'name': 'a' },
                  }]
        result = self.vm.qmp('transaction', actions = actions)
        self.assert_qmp(result, 'error/class', 'GenericError')

    def test_error_exist(self):
        self.createSnapshotInTransaction(1)
        self.verifySnapshotInfo()
        actions = [{'type': 'blockdev-snapshot-internal-sync',
                    'data': { 'device': self.expect[0]['device'],
                              'name': self.expect[0]['snapshots'][0] },
                  }]
        result = self.vm.qmp('transaction', actions = actions)
        self.assert_qmp(result, 'error/class', 'GenericError')

class TestMultipleTransaction(ImageSnapshotTestCase):
    def setUp(self):
        self._setUp('test_b.img', 2)

    def test_create(self):
        self.createSnapshotInTransaction(3)
        self.verifySnapshotInfo()

    def test_abort(self):
        self.createSnapshotInTransaction(2)
        self.verifySnapshotInfo()
        self.createSnapshotInTransaction(3, abort = True)
        self.verifySnapshotInfo()

class TestSnapshotDelete(ImageSnapshotTestCase):
    def setUp(self):
        self._setUp('test_c.img', 1)

    def test_delete_with_id(self):
        self.createSnapshotInTransaction(2)
        self.verifySnapshotInfo()
        self.deleteSnapshot(self.expect[0]['device'],
                            id = self.expect[0]['snapshots'][0]['id'])
        self.verifySnapshotInfo()

    def test_delete_with_name(self):
        self.createSnapshotInTransaction(3)
        self.verifySnapshotInfo()
        self.deleteSnapshot(self.expect[0]['device'],
                            name = self.expect[0]['snapshots'][1]['name'])
        self.verifySnapshotInfo()

    def test_delete_with_id_and_name(self):
        self.createSnapshotInTransaction(4)
        self.verifySnapshotInfo()
        self.deleteSnapshot(self.expect[0]['device'],
                            id = self.expect[0]['snapshots'][2]['id'],
                            name = self.expect[0]['snapshots'][2]['name'])
        self.verifySnapshotInfo()


    def test_error_device(self):
        result = self.vm.qmp('blockdev-snapshot-delete-internal-sync',
                              device = 'drive_error',
                              id = '0')
        self.assert_qmp(result, 'error/class', 'GenericError')

    def test_error_no_id_and_name(self):
        result = self.vm.qmp('blockdev-snapshot-delete-internal-sync',
                              device = self.expect[0]['device'])
        self.assert_qmp(result, 'error/class', 'GenericError')

    def test_error_snapshot_not_exist(self):
        self.createSnapshotInTransaction(2)
        self.verifySnapshotInfo()
        result = self.vm.qmp('blockdev-snapshot-delete-internal-sync',
                              device = self.expect[0]['device'],
                              id = self.expect[0]['snapshots'][0]['id'],
                              name = self.expect[0]['snapshots'][1]['name'])
        self.assert_qmp(result, 'error/class', 'GenericError')

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