tempest/api/volume/test_volumes_backup.py

# Copyright 2016 Red Hat, Inc.
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import testtools
from testtools import matchers

from tempest.api.volume import base
from tempest.common import waiters
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest import test

CONF = config.CONF


class VolumesBackupsTest(base.BaseVolumeTest):

    @classmethod
    def skip_checks(cls):
        super(VolumesBackupsTest, cls).skip_checks()
        if not CONF.volume_feature_enabled.backup:
            raise cls.skipException("Cinder backup feature disabled")

    def restore_backup(self, backup_id):
        # Restore a backup
        restored_volume = self.backups_client.restore_backup(
            backup_id)['restore']

        # Delete backup
        self.addCleanup(self.volumes_client.delete_volume,
                        restored_volume['volume_id'])
        self.assertEqual(backup_id, restored_volume['backup_id'])
        waiters.wait_for_volume_resource_status(self.backups_client,
                                                backup_id, 'available')
        waiters.wait_for_volume_resource_status(self.volumes_client,
                                                restored_volume['volume_id'],
                                                'available')
        return restored_volume

    @testtools.skipIf(CONF.volume.storage_protocol == 'ceph',
                      'ceph does not support arbitrary container names')
    @decorators.idempotent_id('a66eb488-8ee1-47d4-8e9f-575a095728c6')
    def test_volume_backup_create_get_detailed_list_restore_delete(self):
        # Create a volume with metadata
        metadata = {"vol-meta1": "value1",
                    "vol-meta2": "value2",
                    "vol-meta3": "value3"}
        volume = self.create_volume(metadata=metadata)
        self.addCleanup(self.volumes_client.delete_volume,
                        volume['id'])

        # Create a backup
        backup_name = data_utils.rand_name(
            self.__class__.__name__ + '-Backup')
        description = data_utils.rand_name("volume-backup-description")
        backup = self.create_backup(volume_id=volume['id'],
                                    name=backup_name,
                                    description=description,
                                    container='container')
        self.assertEqual(backup_name, backup['name'])
        waiters.wait_for_volume_resource_status(self.volumes_client,
                                                volume['id'], 'available')

        # Get a given backup
        backup = self.backups_client.show_backup(backup['id'])['backup']
        self.assertEqual(backup_name, backup['name'])
        self.assertEqual(description, backup['description'])
        self.assertEqual('container', backup['container'])

        # Get all backups with detail
        backups = self.backups_client.list_backups(
            detail=True)['backups']
        self.assertIn((backup['name'], backup['id']),
                      [(m['name'], m['id']) for m in backups])

        restored_volume = self.restore_backup(backup['id'])

        restored_volume_metadata = self.volumes_client.show_volume(
            restored_volume['volume_id'])['volume']['metadata']

        # Verify the backups has been restored successfully
        # with the metadata of the source volume.
        self.assertThat(restored_volume_metadata.items(),
                        matchers.ContainsAll(metadata.items()))

    @decorators.idempotent_id('07af8f6d-80af-44c9-a5dc-c8427b1b62e6')
    @test.services('compute')
    def test_backup_create_attached_volume(self):
        """Test backup create using force flag.

        Cinder allows to create a volume backup, whether the volume status
        is "available" or "in-use".
        """
        # Create a server
        volume = self.create_volume()
        self.addCleanup(self.volumes_client.delete_volume,
                        volume['id'])
        server = self.create_server()
        # Attach volume to instance
        self.attach_volume(server['id'], volume['id'])
        # Create backup using force flag
        backup_name = data_utils.rand_name(
            self.__class__.__name__ + '-Backup')
        backup = self.create_backup(volume_id=volume['id'],
                                    name=backup_name, force=True)
        self.assertEqual(backup_name, backup['name'])

    @decorators.idempotent_id('2a8ba340-dff2-4511-9db7-646f07156b15')
    @test.services('image')
    def test_bootable_volume_backup_and_restore(self):
        # Create volume from image
        img_uuid = CONF.compute.image_ref
        volume = self.create_volume(imageRef=img_uuid)

        volume_details = self.volumes_client.show_volume(
            volume['id'])['volume']
        self.assertEqual('true', volume_details['bootable'])

        # Create a backup
        backup = self.create_backup(volume_id=volume['id'])

        # Restore the backup
        restored_volume_id = self.restore_backup(backup['id'])['volume_id']

        # Verify the restored backup volume is bootable
        restored_volume_info = self.volumes_client.show_volume(
            restored_volume_id)['volume']

        self.assertEqual('true', restored_volume_info['bootable'])


class VolumesBackupsV39Test(base.BaseVolumeTest):

    _api_version = 3
    min_microversion = '3.9'
    max_microversion = 'latest'

    @classmethod
    def skip_checks(cls):
        super(VolumesBackupsV39Test, cls).skip_checks()
        if not CONF.volume_feature_enabled.backup:
            raise cls.skipException("Cinder backup feature disabled")

    @decorators.idempotent_id('9b374cbc-be5f-4d37-8848-7efb8a873dcc')
    def test_update_backup(self):
        # Create volume and backup
        volume = self.create_volume()
        backup = self.create_backup(volume_id=volume['id'])

        # Update backup and assert response body for update_backup method
        update_kwargs = {
            'name': data_utils.rand_name(self.__class__.__name__ + '-Backup'),
            'description': data_utils.rand_name("volume-backup-description")
        }
        update_backup = self.backups_client.update_backup(
            backup['id'], **update_kwargs)['backup']
        self.assertEqual(backup['id'], update_backup['id'])
        self.assertEqual(update_kwargs['name'], update_backup['name'])
        self.assertIn('links', update_backup)

        # Assert response body for show_backup method
        retrieved_backup = self.backups_client.show_backup(
            backup['id'])['backup']
        for key in update_kwargs:
            self.assertEqual(update_kwargs[key], retrieved_backup[key])