# 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 pytest import allure import time from tests.api.volume import base from configs import config from lib.common.utils import data_utils from lib import exceptions as lib_exc from common import waiters CONF = config.CONF class TestVolumesSnapshotJSON(base.BaseVolumeTest): """Test volume snapshots""" create_default_network = True @classmethod def skip_checks(cls): super(TestVolumesSnapshotJSON, cls).skip_checks() if not CONF.volume_feature_enabled.snapshot: raise cls.skipException("Cinder volume snapshots are disabled") @classmethod def resource_setup(cls): super(TestVolumesSnapshotJSON, cls).resource_setup() @pytest.mark.positive @pytest.mark.skipif( not CONF.common.availability_zone_list, reason='there is no availability_zone_list', ) @pytest.mark.skipif( not CONF.volume.volume_type_ssd and not CONF.volume.volume_type_hdd, reason='SSD and HDD volume type not available.', ) def test_snapshot_create_get_list_update_delete(self): """Test create/get/list/update/delete snapshot :ref: `test_snapshot_create_get_list_update_delete `__ """ if CONF.volume.volume_type_ssd: vol_type = CONF.volume.volume_type_ssd else: vol_type = CONF.volume.volume_type_hdd for vol_az in CONF.common.availability_zone_list: # Create a test volume with allure.step(f"Volume create with AZ:{vol_az} VolumeType:{vol_type}"): volume = self.create_volume( availability_zone=vol_az, size=CONF.volume.volume_size, volume_type=vol_type ) assert volume['id'], f"Failed to create volume: {volume['id']}" # Create a snapshot with metadata with allure.step(f"Verify create snapshot with metadata AZ:{vol_az} VolumeType:{vol_type}"): metadata = {"snap-meta1": "value1", "snap-meta2": "value2", "snap-meta3": "value3"} snapshot = self.create_snapshot(volume['id'], metadata=metadata) assert snapshot['id'], f"Failed to create snapshot: {snapshot['id']}" # Get the snap and check for some of its details with allure.step(f"Verify get snapshot AZ:{vol_az} VolumeType:{vol_type}"): snap_get = self.snapshots_client.show_snapshot( snapshot['id'])['snapshot'] assert volume['id'] == snap_get['volume_id'], \ f"Not matched volume id({volume['id']}) with snapshot get volume id({snap_get['volume_id']})" assert volume['size'] == snap_get['size'], \ f"Not matched volume size({volume['size']}) with snapshot get volume size({snap_get['size']})" # Verify snapshot metadata with allure.step(f"Verify snapshot metadata AZ:{vol_az} VolumeType:{vol_type}"): assert set(snap_get['metadata'].items()).issubset(metadata.items()), f"Failed to verify snapshot metadata({snap_get['metadata']})" # Compare also with the output from the list action with allure.step(f"Verify list snapshot AZ:{vol_az} VolumeType:{vol_type}"): tracking_data = (snapshot['id'], snapshot['name'], 'available') snaps_list = self.snapshots_client.list_snapshots()['snapshots'] snaps_data = [(f['id'], f['name'], f['status']) for f in snaps_list] assert tracking_data in snaps_data, f"Failed to verify list snapshot ({snaps_data})" # Updates snapshot with new values with allure.step(f"Update snapshot with new values AZ:{vol_az} VolumeType:{vol_type}"): new_s_name = data_utils.rand_name('new-snap') new_desc = 'This is the new description of snapshot.' params = {'name': new_s_name, 'description': new_desc} update_snapshot = self.snapshots_client.update_snapshot( snapshot['id'], **params)['snapshot'] # Assert response body for update_snapshot method with allure.step(f"Verify Updated snapshot by response body values AZ:{vol_az} VolumeType:{vol_type}"): assert new_s_name == update_snapshot['name'], f"Failed to verify updated name {new_s_name}" assert new_desc == update_snapshot['description'], f"Failed to verify updated description {new_desc}" # Assert response body for show_snapshot method with allure.step(f"Verify Updated snapshot by show snapshot values AZ:{vol_az} VolumeType:{vol_type}"): updated_snapshot = self.snapshots_client.show_snapshot( snapshot['id'])['snapshot'] assert new_s_name == updated_snapshot['name'], f"Failed to verify updated name {new_s_name}" assert new_desc == updated_snapshot['description'], f"Failed to verify updated description {new_desc}" # Delete the snapshot with allure.step(f"Verify Delete snapshot AZ:{vol_az} VolumeType:{vol_type}"): try: self.delete_snapshot(snapshot['id']) except lib_exc.TimeoutException: assert False, f"Failed to delete snapshot : {snapshot['id']}" def _create_volume_from_snapshot(self, extra_size=0, availability_zone=None, volume_type=None): if not availability_zone: return False, "There is no availability_zone" if not volume_type: return False, "There is no volume_type" src_size = CONF.volume.volume_size size = src_size + extra_size src_vol = self.create_volume(size=src_size, availability_zone=availability_zone, volume_type=volume_type) src_snap = self.create_snapshot(src_vol['id']) dst_vol = self.create_volume(snapshot_id=src_snap['id'], size=size, availability_zone=availability_zone, volume_type=volume_type) # NOTE(zhufl): dst_vol is created based on snapshot, so dst_vol # should be deleted before deleting snapshot, otherwise deleting # snapshot will end with status 'error-deleting'. This depends on # the implementation mechanism of vendors, generally speaking, # some verdors will use "virtual disk clone" which will promote # disk clone speed, and in this situation the "disk clone" # is just a relationship between volume and snapshot. self.addCleanup(self.delete_volume, self.volumes_client, dst_vol['id']) volume = self.volumes_client.show_volume(dst_vol['id'])['volume'] # Should allow if volume['status'] != 'available': return False, f"Volume({volume['id']}) is not available status" if volume['snapshot_id'] != src_snap['id']: return False, f"Not matched volume's snapshot id({volume['snapshot_id']}) and source snapshot id({src_snap['id']})" if volume['size'] != size: return False, f"Not matched volume's size({volume['snapshot_id']}) and request size({size})" return True, "" @pytest.mark.positive @pytest.mark.skipif( not CONF.common.availability_zone_list, reason='there is no availability_zone_list', ) @pytest.mark.skipif( not CONF.volume.volume_type_ssd and not CONF.volume.volume_type_hdd, reason='SSD and HDD volume type not available.', ) def test_volume_from_snapshot(self): """Test creating volume from snapshot with extending size :ref: `test_volume_from_snapshot `__ """ if CONF.volume.volume_type_ssd: vol_type = CONF.volume.volume_type_ssd else: vol_type = CONF.volume.volume_type_hdd for vol_az in CONF.common.availability_zone_list: with allure.step(f"Verify volume create from snapshot with extend size AZ:{vol_az} VolumeType:{vol_type}"): is_success, msg = self._create_volume_from_snapshot( extra_size=CONF.volume.volume_size_extend, availability_zone=vol_az, volume_type=vol_type) assert is_success, msg @pytest.mark.positive @pytest.mark.skipif( not CONF.common.availability_zone_list, reason='there is no availability_zone_list', ) @pytest.mark.skipif( not CONF.volume.volume_type_ssd and not CONF.volume.volume_type_hdd, reason='SSD and HDD volume type not available.', ) def test_volume_from_snapshot_no_size(self): """Test creating volume from snapshot with original size :ref: `test_volume_from_snapshot_no_size `__ """ if CONF.volume.volume_type_ssd: vol_type = CONF.volume.volume_type_ssd else: vol_type = CONF.volume.volume_type_hdd for vol_az in CONF.common.availability_zone_list: with allure.step( f"Verify volume create from snapshot with original size AZ:{vol_az} VolumeType:{vol_type}" ): is_success, msg = self._create_volume_from_snapshot(availability_zone=vol_az, volume_type=vol_type) assert is_success, msg @pytest.mark.positive @pytest.mark.skipif( not CONF.common.availability_zone_list, reason='there is no availability_zone_list', ) @pytest.mark.skipif( not CONF.volume.volume_type_ssd and not CONF.volume.volume_type_hdd, reason='SSD and HDD volume type not available.', ) def test_attach_detach_volume_via_snapshot_to_vm(self): """Test attach and detach volume created from snapshot to server""" if CONF.volume.volume_type_ssd: vol_type = CONF.volume.volume_type_ssd else: vol_type = CONF.volume.volume_type_hdd for vol_az in CONF.common.availability_zone_list: # Create a test server with allure.step(f"Create test server with AZ:{vol_az}, VolumeType:{vol_type}"): server = self.create_server(wait_until='SSHABLE', availability_zone=vol_az) assert server['id'], f"Failed to create server: {server['id']}" # Create a test volume with allure.step(f"Create test volume with AZ:{vol_az}, VolumeType:{vol_type}"): volume = self.create_volume( availability_zone=vol_az, size=CONF.volume.volume_size, volume_type=vol_type ) assert volume['id'], f"Failed to create volume: {volume['id']}" # Create a snapshot with allure.step(f"Verify create snapshot with metadata AZ:{vol_az} VolumeType:{vol_type}"): snapshot = self.create_snapshot(volume['id']) assert snapshot['id'], f"Failed to create snapshot: {snapshot['id']}" # Create volume from snapshot with allure.step(f"Verify volume create from snapshot with AZ:{vol_az} VolumeType:{vol_type}"): is_success, msg = self._create_volume_from_snapshot(availability_zone=vol_az, volume_type=vol_type) assert is_success, msg # Attach volume from snapshot to server with allure.step(f"Verify Attach volume from snapshot to server with AZ:{vol_az}, VolumeType:{vol_type}"): self.attach_volume(server['id'], volume['id'], wait_for_detach=False, is_cleanup=False) assert self.volumes_client.show_volume(volume['id'])['volume']['status'] == 'in-use' # Wait reason: There is a problem if run Detach immediately after status 'in-use' changed. time.sleep(5) # Detach volume from snapshot to server with allure.step(f"Verify Detach volume from snapshot to server with AZ:{vol_az}, VolumeType:{vol_type}"): resp = self.servers_client.detach_volume(server['id'], volume_id=volume['id']) assert resp.response['status'] == '202', "Failed to detached volume to instance" with allure.step( f"Wait Volume status to 'available' after volume detach with AZ:{vol_az}, VolumeType:{vol_type}"): try: waiters.wait_for_volume_resource_status(self.volumes_client, volume['id'], 'available') except lib_exc.TimeoutException: assert False, "Failed to wait volume status to 'available' after volume detach"