Source code for heat_integrationtests.scenario.test_volumes

#    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.


from cinderclient import exceptions as cinder_exceptions
from oslo_log import log as logging
import six

from heat_integrationtests.common import exceptions
from heat_integrationtests.scenario import scenario_base

LOG = logging.getLogger(__name__)


[docs]class VolumeBackupRestoreIntegrationTest(scenario_base.ScenarioTestsBase):
    """Class is responsible for testing of volume backup."""

[docs]    def setUp(self):
        super(VolumeBackupRestoreIntegrationTest, self).setUp()
        self.volume_description = 'A test volume description 123'
        self.volume_size = self.conf.volume_size

    def _cinder_verify(self, volume_id, expected_status='available'):
        self.assertIsNotNone(volume_id)
        volume = self.volume_client.volumes.get(volume_id)
        self.assertIsNotNone(volume)
        self.assertEqual(expected_status, volume.status)
        self.assertEqual(self.volume_size, volume.size)
        self.assertEqual(self.volume_description,
                         volume.display_description)

    def _outputs_verify(self, stack, expected_status='available'):
        self.assertEqual(expected_status,
                         self._stack_output(stack, 'status'))
        self.assertEqual(six.text_type(self.volume_size),
                         self._stack_output(stack, 'size'))
        self.assertEqual(self.volume_description,
                         self._stack_output(stack, 'display_description'))

[docs]    def check_stack(self, stack_id, parameters):
        stack = self.client.stacks.get(stack_id)

        # Verify with cinder that the volume exists, with matching details
        volume_id = self._stack_output(stack, 'volume_id')
        self._cinder_verify(volume_id, expected_status='in-use')

        # Verify the stack outputs are as expected
        self._outputs_verify(stack, expected_status='in-use')

        # Delete the stack and ensure a backup is created for volume_id
        # but the volume itself is gone
        self._stack_delete(stack_id)
        self.assertRaises(cinder_exceptions.NotFound,
                          self.volume_client.volumes.get,
                          volume_id)

        backups = self.volume_client.backups.list()
        self.assertIsNotNone(backups)
        backups_filtered = [b for b in backups if b.volume_id == volume_id]
        self.assertEqual(1, len(backups_filtered))
        backup = backups_filtered[0]
        self.addCleanup(self.volume_client.backups.delete, backup.id)

        # Now, we create another stack where the volume is created from the
        # backup created by the previous stack
        try:
            stack_identifier2 = self.launch_stack(
                template_name='test_volumes_create_from_backup.yaml',
                parameters=parameters,
                add_parameters={'backup_id': backup.id})
            stack2 = self.client.stacks.get(stack_identifier2)
        except exceptions.StackBuildErrorException:
            LOG.exception("Halting test due to bug: #1382300")
            return

        # Verify with cinder that the volume exists, with matching details
        volume_id2 = self._stack_output(stack2, 'volume_id')
        self._cinder_verify(volume_id2, expected_status='in-use')

        # Verify the stack outputs are as expected
        self._outputs_verify(stack2, expected_status='in-use')
        testfile_data = self._stack_output(stack2, 'testfile_data')
        self.assertEqual('{"instance1": "Volume Data:ateststring"}',
                         testfile_data)

        # Delete the stack and ensure the volume is gone
        self._stack_delete(stack_identifier2)
        self.assertRaises(cinder_exceptions.NotFound,
                          self.volume_client.volumes.get,
                          volume_id2)

[docs]    def test_cinder_volume_create_backup_restore(self):
        """Ensure the 'Snapshot' deletion policy works.

        This requires a more complex test, but it tests several aspects
        of the heat cinder resources:
           1. Create a volume, attach it to an instance, write some data to it
           2. Delete the stack, with 'Snapshot' specified, creates a backup
           3. Check the snapshot has created a volume backup
           4. Create a new stack, where the volume is created from the backup
           5. Verify the test data written in (1) is present in the new volume
        """
        parameters = {
            'key_name': self.keypair_name,
            'instance_type': self.conf.minimal_instance_type,
            'image_id': self.conf.minimal_image_ref,
            'volume_description': self.volume_description,
            'timeout': self.conf.build_timeout,
            'network': self.net['id']
        }

        # Launch stack
        stack_id = self.launch_stack(
            template_name='test_volumes_delete_snapshot.yaml',
            parameters=parameters,
            add_parameters={'volume_size': self.volume_size}
        )

        # Check stack
        self.check_stack(stack_id, parameters)