| # Copyright 2017 NEC Corporation. |
| # 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 io |
| |
| from tempest.common import compute |
| from tempest.common import waiters |
| from tempest import config |
| from tempest.lib.common import api_version_utils |
| from tempest.lib.common.utils import data_utils |
| from tempest.lib.common.utils import test_utils |
| from tempest import test |
| |
| CONF = config.CONF |
| |
| |
| class BaseVolumeTest(api_version_utils.BaseMicroversionTest, |
| test.BaseTestCase): |
| """Base test case class for all Cinder API tests.""" |
| |
| credentials = ['primary'] |
| |
| @classmethod |
| def skip_checks(cls): |
| super(BaseVolumeTest, cls).skip_checks() |
| |
| if not CONF.service_available.cinder: |
| skip_msg = ("%s skipped as Cinder is not available" % cls.__name__) |
| raise cls.skipException(skip_msg) |
| |
| api_version_utils.check_skip_with_microversion( |
| cls.min_microversion, cls.max_microversion, |
| CONF.volume.min_microversion, CONF.volume.max_microversion) |
| |
| @classmethod |
| def setup_clients(cls): |
| super(BaseVolumeTest, cls).setup_clients() |
| cls.backups_client = cls.os_primary.backups_client_latest |
| cls.volumes_client = cls.os_primary.volumes_client_latest |
| cls.snapshots_client = cls.os_primary.snapshots_client_latest |
| cls.volume_revert_client = ( |
| cls.os_primary.volume_revert_v3.VolumeRevertClient() |
| ) |
| |
| @classmethod |
| def setup_credentials(cls): |
| cls.set_network_resources() |
| super(BaseVolumeTest, cls).setup_credentials() |
| |
| def setUp(self): |
| super(BaseVolumeTest, self).setUp() |
| |
| @classmethod |
| def resource_setup(cls): |
| super(BaseVolumeTest, cls).resource_setup() |
| cls.request_microversion = ( |
| api_version_utils.select_request_microversion( |
| cls.min_microversion, |
| CONF.volume.min_microversion)) |
| cls.setup_api_microversion_fixture( |
| volume_microversion=cls.request_microversion) |
| |
| @classmethod |
| def create_volume(cls, wait_until='available', **kwargs): |
| """Wrapper utility that returns a test volume. |
| |
| :param wait_until: wait till volume status, None means no wait. |
| """ |
| if 'size' not in kwargs: |
| kwargs['size'] = CONF.volume.volume_size |
| |
| if 'imageRef' in kwargs: |
| image = cls.os_primary.image_client_v2.show_image( |
| kwargs['imageRef']) |
| min_disk = image['min_disk'] |
| kwargs['size'] = max(kwargs['size'], min_disk) |
| |
| if 'name' not in kwargs: |
| name = data_utils.rand_name(cls.__name__ + '-Volume') |
| kwargs['name'] = name |
| |
| if CONF.compute.compute_volume_common_az: |
| kwargs.setdefault('availability_zone', |
| CONF.compute.compute_volume_common_az) |
| |
| volume = cls.volumes_client.create_volume(**kwargs)['volume'] |
| cls.addClassResourceCleanup( |
| cls.volumes_client.wait_for_resource_deletion, volume['id']) |
| cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc, |
| cls.volumes_client.delete_volume, |
| volume['id']) |
| if wait_until: |
| waiters.wait_for_volume_resource_status(cls.volumes_client, |
| volume['id'], wait_until) |
| return volume |
| |
| @classmethod |
| def create_snapshot(cls, volume_id=1, **kwargs): |
| """Wrapper utility that returns a test snapshot.""" |
| if 'name' not in kwargs: |
| name = data_utils.rand_name(cls.__name__ + '-Snapshot') |
| kwargs['name'] = name |
| |
| snapshot = cls.snapshots_client.create_snapshot( |
| volume_id=volume_id, **kwargs)['snapshot'] |
| cls.addClassResourceCleanup( |
| cls.snapshots_client.wait_for_resource_deletion, snapshot['id']) |
| cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc, |
| cls.snapshots_client.delete_snapshot, |
| snapshot['id']) |
| waiters.wait_for_volume_resource_status(cls.snapshots_client, |
| snapshot['id'], 'available') |
| return snapshot |
| |
| def create_backup(self, volume_id, backup_client=None, **kwargs): |
| """Wrapper utility that returns a test backup.""" |
| if backup_client is None: |
| backup_client = self.backups_client |
| if 'name' not in kwargs: |
| name = data_utils.rand_name(self.__class__.__name__ + '-Backup') |
| kwargs['name'] = name |
| |
| backup = backup_client.create_backup( |
| volume_id=volume_id, **kwargs)['backup'] |
| self.addCleanup(backup_client.wait_for_resource_deletion, backup['id']) |
| self.addCleanup(test_utils.call_and_ignore_notfound_exc, |
| backup_client.delete_backup, backup['id']) |
| waiters.wait_for_volume_resource_status(backup_client, backup['id'], |
| 'available') |
| return backup |
| |
| def create_server(self, wait_until='ACTIVE', **kwargs): |
| name = kwargs.pop( |
| 'name', |
| data_utils.rand_name(self.__class__.__name__ + '-instance')) |
| |
| if wait_until == 'SSHABLE' and not kwargs.get('validation_resources'): |
| kwargs['validation_resources'] = ( |
| self.get_test_validation_resources(self.os_primary)) |
| kwargs['validatable'] = True |
| |
| tenant_network = self.get_tenant_network() |
| body, _ = compute.create_test_server( |
| self.os_primary, |
| tenant_network=tenant_network, |
| name=name, |
| wait_until=wait_until, |
| **kwargs) |
| |
| self.addCleanup(test_utils.call_and_ignore_notfound_exc, |
| waiters.wait_for_server_termination, |
| self.os_primary.servers_client, body['id']) |
| self.addCleanup(test_utils.call_and_ignore_notfound_exc, |
| self.os_primary.servers_client.delete_server, |
| body['id']) |
| return body |
| |
| @classmethod |
| def create_image_with_data(cls, **kwargs): |
| # we do this as a class method so we can use the |
| # addClassResourceCleanup functionality of tempest.test.BaseTestCase |
| images_client = cls.os_primary.image_client_v2 |
| if 'min_disk' not in kwargs: |
| kwargs['min_disk'] = 1 |
| response = images_client.create_image(**kwargs) |
| image_id = response['id'] |
| cls.addClassResourceCleanup( |
| images_client.wait_for_resource_deletion, image_id) |
| cls.addClassResourceCleanup( |
| test_utils.call_and_ignore_notfound_exc, |
| images_client.delete_image, image_id) |
| |
| # upload "data" to image |
| image_file = io.BytesIO(data_utils.random_bytes(size=1024)) |
| images_client.store_image_file(image_id, image_file) |
| |
| waiters.wait_for_image_status(images_client, image_id, 'active') |
| image = images_client.show_image(image_id) |
| return image |
| |
| |
| class BaseVolumeAdminTest(BaseVolumeTest): |
| """Base test case class for all Volume Admin API tests.""" |
| |
| credentials = ['primary', 'admin'] |
| |
| @classmethod |
| def setup_clients(cls): |
| super(BaseVolumeAdminTest, cls).setup_clients() |
| |
| cls.admin_volume_types_client = cls.os_admin.volume_types_client_latest |
| cls.admin_backups_client = cls.os_admin.backups_client_latest |
| cls.admin_volume_client = cls.os_admin.volumes_client_latest |
| cls.admin_consistencygroups_client = ( |
| cls.os_admin.consistencygroups_v3.ConsistencyGroupsClient() |
| ) |
| |
| @classmethod |
| def create_volume_type(cls, name=None, **kwargs): |
| """Create a test volume-type""" |
| |
| name = name or data_utils.rand_name(cls.__name__ + '-volume-type') |
| volume_type = cls.admin_volume_types_client.create_volume_type( |
| name=name, **kwargs)['volume_type'] |
| cls.addClassResourceCleanup(cls._clear_volume_type, volume_type) |
| return volume_type |
| |
| @classmethod |
| def _clear_volume_type(cls, volume_type): |
| # If image caching is enabled, we must delete the cached volume |
| # before cinder will allow us to delete the volume_type. This function |
| # solves that problem by taking the brute-force approach of deleting |
| # any volumes of this volume_type that exist *no matter what project |
| # they are in*. Since this won't happen until the teardown of the |
| # test class, that should be OK. |
| type_id = volume_type['id'] |
| type_name = volume_type['name'] |
| |
| volumes = cls.admin_volume_client.list_volumes( |
| detail=True, params={'all_tenants': 1})['volumes'] |
| for volume in [v for v in volumes if v['volume_type'] == type_name]: |
| test_utils.call_and_ignore_notfound_exc( |
| cls.admin_volume_client.delete_volume, volume['id']) |
| cls.admin_volume_client.wait_for_resource_deletion(volume['id']) |
| |
| test_utils.call_and_ignore_notfound_exc( |
| cls.admin_volume_types_client.delete_volume_type, type_id) |
| test_utils.call_and_ignore_notfound_exc( |
| cls.admin_volume_types_client.wait_for_resource_deletion, type_id) |
| |
| |
| class CreateMultipleResourceTest(BaseVolumeTest): |
| |
| def _create_multiple_resource(self, callback, repeat_count=5, |
| **kwargs): |
| |
| res = [] |
| for _ in range(repeat_count): |
| res.append(callback(**kwargs)['id']) |
| return res |
| |
| def _wait_for_multiple_resources(self, callback, wait_list, **kwargs): |
| |
| for r in wait_list: |
| callback(resource_id=r, **kwargs) |