| # Copyright 2012 OpenStack Foundation |
| # 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.utils import data_utils |
| from tempest.common import waiters |
| from tempest import config |
| from tempest.lib import decorators |
| from tempest import test |
| |
| CONF = config.CONF |
| |
| |
| class VolumesV2GetTest(base.BaseVolumeTest): |
| |
| def _volume_create_get_update_delete(self, **kwargs): |
| name_field = self.special_fields['name_field'] |
| descrip_field = self.special_fields['descrip_field'] |
| |
| # Create a volume, Get it's details and Delete the volume |
| v_name = data_utils.rand_name(self.__class__.__name__ + '-Volume') |
| metadata = {'Type': 'Test'} |
| # Create a volume |
| kwargs[name_field] = v_name |
| kwargs['metadata'] = metadata |
| volume = self.volumes_client.create_volume(**kwargs)['volume'] |
| self.assertIn('id', volume) |
| self.addCleanup(self.delete_volume, self.volumes_client, volume['id']) |
| waiters.wait_for_volume_resource_status(self.volumes_client, |
| volume['id'], 'available') |
| self.assertIn(name_field, volume) |
| self.assertEqual(volume[name_field], v_name, |
| "The created volume name is not equal " |
| "to the requested name") |
| |
| # Get Volume information |
| fetched_volume = self.volumes_client.show_volume( |
| volume['id'])['volume'] |
| self.assertEqual(v_name, |
| fetched_volume[name_field], |
| 'The fetched Volume name is different ' |
| 'from the created Volume') |
| self.assertEqual(volume['id'], |
| fetched_volume['id'], |
| 'The fetched Volume id is different ' |
| 'from the created Volume') |
| self.assertThat(fetched_volume['metadata'].items(), |
| matchers.ContainsAll(metadata.items()), |
| 'The fetched Volume metadata misses data ' |
| 'from the created Volume') |
| |
| if 'imageRef' in kwargs: |
| self.assertEqual('true', fetched_volume['bootable']) |
| else: |
| self.assertEqual('false', fetched_volume['bootable']) |
| |
| # Update Volume |
| # Test volume update when display_name is same with original value |
| params = {name_field: v_name} |
| self.volumes_client.update_volume(volume['id'], **params) |
| # Test volume update when display_name is new |
| new_v_name = data_utils.rand_name( |
| self.__class__.__name__ + '-new-Volume') |
| new_desc = 'This is the new description of volume' |
| params = {name_field: new_v_name, |
| descrip_field: new_desc} |
| update_volume = self.volumes_client.update_volume( |
| volume['id'], **params)['volume'] |
| # Assert response body for update_volume method |
| self.assertEqual(new_v_name, update_volume[name_field]) |
| self.assertEqual(new_desc, update_volume[descrip_field]) |
| # Assert response body for show_volume method |
| updated_volume = self.volumes_client.show_volume( |
| volume['id'])['volume'] |
| self.assertEqual(volume['id'], updated_volume['id']) |
| self.assertEqual(new_v_name, updated_volume[name_field]) |
| self.assertEqual(new_desc, updated_volume[descrip_field]) |
| self.assertThat(updated_volume['metadata'].items(), |
| matchers.ContainsAll(metadata.items()), |
| 'The fetched Volume metadata misses data ' |
| 'from the created Volume') |
| |
| # Test volume create when display_name is none and display_description |
| # contains specific characters, |
| # then test volume update if display_name is duplicated |
| new_v_desc = data_utils.rand_name('@#$%^* description') |
| params = {descrip_field: new_v_desc, |
| 'availability_zone': volume['availability_zone'], |
| 'size': CONF.volume.volume_size} |
| new_volume = self.volumes_client.create_volume(**params)['volume'] |
| self.assertIn('id', new_volume) |
| self.addCleanup(self.delete_volume, self.volumes_client, |
| new_volume['id']) |
| waiters.wait_for_volume_resource_status(self.volumes_client, |
| new_volume['id'], 'available') |
| |
| params = {name_field: volume[name_field], |
| descrip_field: volume[descrip_field]} |
| self.volumes_client.update_volume(new_volume['id'], **params) |
| |
| if 'imageRef' in kwargs: |
| self.assertEqual('true', updated_volume['bootable']) |
| else: |
| self.assertEqual('false', updated_volume['bootable']) |
| |
| @test.attr(type='smoke') |
| @decorators.idempotent_id('27fb0e9f-fb64-41dd-8bdb-1ffa762f0d51') |
| def test_volume_create_get_update_delete(self): |
| self._volume_create_get_update_delete(size=CONF.volume.volume_size) |
| |
| @test.attr(type='smoke') |
| @decorators.idempotent_id('54a01030-c7fc-447c-86ee-c1182beae638') |
| @test.services('image') |
| def test_volume_create_get_update_delete_from_image(self): |
| image = self.compute_images_client.show_image( |
| CONF.compute.image_ref)['image'] |
| min_disk = image.get('minDisk') |
| disk_size = max(min_disk, CONF.volume.volume_size) |
| self._volume_create_get_update_delete( |
| imageRef=CONF.compute.image_ref, size=disk_size) |
| |
| @decorators.idempotent_id('3f591b4a-7dc6-444c-bd51-77469506b3a1') |
| @testtools.skipUnless(CONF.volume_feature_enabled.clone, |
| 'Cinder volume clones are disabled') |
| def test_volume_create_get_update_delete_as_clone(self): |
| origin = self.create_volume() |
| self._volume_create_get_update_delete(source_volid=origin['id'], |
| size=CONF.volume.volume_size) |
| |
| |
| class VolumesV1GetTest(VolumesV2GetTest): |
| _api_version = 1 |