# Copyright 2017 FiberHome Telecommunication Technologies CO.,LTD
# Copyright (C) 2017 Dell Inc. or its subsidiaries.
# 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.
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.common.utils import test_utils
from tempest.lib import decorators
CONF = config.CONF
[docs]
class BaseGroupSnapshotsTest(base.BaseVolumeAdminTest):
@classmethod
def skip_checks(cls):
super(BaseGroupSnapshotsTest, cls).skip_checks()
if not CONF.volume_feature_enabled.snapshot:
raise cls.skipException("Cinder volume snapshots are disabled")
def _create_group_snapshot(self, **kwargs):
if 'name' not in kwargs:
kwargs['name'] = data_utils.rand_name(
prefix=CONF.resource_name_prefix,
name=self.__class__.__name__ + '-Group_Snapshot')
group_snapshot = self.group_snapshots_client.create_group_snapshot(
**kwargs)['group_snapshot']
group_snapshot['group_id'] = kwargs['group_id']
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self._delete_group_snapshot, group_snapshot)
waiters.wait_for_volume_resource_status(
self.group_snapshots_client, group_snapshot['id'], 'available')
return group_snapshot
def _delete_group_snapshot(self, group_snapshot):
self.group_snapshots_client.delete_group_snapshot(group_snapshot['id'])
vols = self.volumes_client.list_volumes(detail=True)['volumes']
snapshots = self.snapshots_client.list_snapshots(
detail=True)['snapshots']
for vol in vols:
for snap in snapshots:
if (vol['group_id'] == group_snapshot['group_id'] and
vol['id'] == snap['volume_id']):
self.snapshots_client.wait_for_resource_deletion(
snap['id'])
self.group_snapshots_client.wait_for_resource_deletion(
group_snapshot['id'])
[docs]
class GroupSnapshotsTest(BaseGroupSnapshotsTest):
"""Test group snapshot"""
volume_min_microversion = '3.14'
volume_max_microversion = 'latest'
[docs]
@decorators.idempotent_id('1298e537-f1f0-47a3-a1dd-8adec8168897')
def test_group_snapshot_create_show_list_delete(self):
"""Test create/show/list/delete group snapshot
1. Create volume type "volume_type1"
2. Create group type "group_type1"
3. Create group "group1" with "group_type1" and "volume_type1"
4. Create volume "volume1" with "volume_type1" and "group1"
5. Create group snapshot "group_snapshot1" with "group1"
6. Check snapshot created from "volume1" reaches available status
7. Check the created group snapshot "group_snapshot1" is in the list
of all group snapshots
8. Delete group snapshot "group_snapshot1"
"""
# Create volume type
volume_type = self.create_volume_type()
# Create group type
group_type = self.create_group_type()
# Create group
grp = self.create_group(group_type=group_type['id'],
volume_types=[volume_type['id']])
# Create volume is instance level, can not be deleted before group.
# Volume delete handled by delete_group method, cleanup method.
prefix = CONF.resource_name_prefix
params = {'name': data_utils.rand_name(prefix=prefix, name="volume"),
'volume_type': volume_type['id'],
'group_id': grp['id'],
'size': CONF.volume.volume_size}
vol = self.volumes_client.create_volume(**params)['volume']
waiters.wait_for_volume_resource_status(
self.volumes_client, vol['id'], 'available')
# Create group snapshot
group_snapshot_name = data_utils.rand_name(prefix=prefix,
name='group_snapshot')
group_snapshot = self._create_group_snapshot(
group_id=grp['id'], name=group_snapshot_name)
snapshots = self.snapshots_client.list_snapshots(
detail=True)['snapshots']
for snap in snapshots:
if vol['id'] == snap['volume_id']:
waiters.wait_for_volume_resource_status(
self.snapshots_client, snap['id'], 'available')
self.assertEqual(group_snapshot_name, group_snapshot['name'])
# Get a given group snapshot
group_snapshot = self.group_snapshots_client.show_group_snapshot(
group_snapshot['id'])['group_snapshot']
self.assertEqual(group_snapshot_name, group_snapshot['name'])
# Get all group snapshots with details, check some detail-specific
# elements, and look for the created group snapshot
group_snapshots = self.group_snapshots_client.list_group_snapshots(
detail=True)['group_snapshots']
for grp_snapshot in group_snapshots:
self.assertIn('created_at', grp_snapshot)
self.assertIn('group_id', grp_snapshot)
self.assertIn((group_snapshot['name'], group_snapshot['id']),
[(m['name'], m['id']) for m in group_snapshots])
# Delete group snapshot
self._delete_group_snapshot(group_snapshot)
group_snapshots = self.group_snapshots_client.list_group_snapshots()[
'group_snapshots']
self.assertNotIn((group_snapshot['name'], group_snapshot['id']),
[(m['name'], m['id']) for m in group_snapshots])
[docs]
@decorators.idempotent_id('eff52c70-efc7-45ed-b47a-4ad675d09b81')
def test_create_group_from_group_snapshot(self):
"""Test creating group from group snapshot
1. Create volume type "volume_type1"
2. Create group type "group_type1"
3. Create group "group1" with "group_type1" and "volume_type1"
4. Create volume "volume1" with "volume_type1" and "group1"
5. Create group snapshot "group_snapshot1" with "group1"
6. Check snapshot created from "volume1" reaches available status
7. Create group "group2" from "group_snapshot1"
8. Check the volumes belonging to "group2" reach available status
9. Check "group2" reaches available status
"""
# Create volume type
volume_type = self.create_volume_type()
# Create group type
group_type = self.create_group_type()
# Create Group
grp = self.create_group(group_type=group_type['id'],
volume_types=[volume_type['id']])
# Create volume is instance level, can not be deleted before group.
# Volume delete handled by delete_group method, cleanup method.
prefix = CONF.resource_name_prefix
params = {'name': data_utils.rand_name(prefix=prefix, name="volume"),
'volume_type': volume_type['id'],
'group_id': grp['id'],
'size': CONF.volume.volume_size}
vol = self.volumes_client.create_volume(**params)['volume']
waiters.wait_for_volume_resource_status(
self.volumes_client, vol['id'], 'available')
# Create group_snapshot
group_snapshot_name = data_utils.rand_name(prefix=prefix,
name='group_snapshot')
group_snapshot = self._create_group_snapshot(
group_id=grp['id'], name=group_snapshot_name)
self.assertEqual(group_snapshot_name, group_snapshot['name'])
snapshots = self.snapshots_client.list_snapshots(
detail=True)['snapshots']
for snap in snapshots:
if vol['id'] == snap['volume_id']:
waiters.wait_for_volume_resource_status(
self.snapshots_client, snap['id'], 'available')
# Create Group from Group snapshot
grp_name2 = data_utils.rand_name(prefix=prefix, name='Group_from_snap')
grp2 = self.groups_client.create_group_from_source(
group_snapshot_id=group_snapshot['id'], name=grp_name2)['group']
self.addCleanup(self.delete_group, grp2['id'])
self.assertEqual(grp_name2, grp2['name'])
vols = self.volumes_client.list_volumes(detail=True)['volumes']
for vol in vols:
if vol['group_id'] == grp2['id']:
waiters.wait_for_volume_resource_status(
self.volumes_client, vol['id'], 'available')
waiters.wait_for_volume_resource_status(
self.groups_client, grp2['id'], 'available')
[docs]
@decorators.idempotent_id('7d7fc000-0b4c-4376-a372-544116d2e127')
@decorators.related_bug('1739031')
def test_delete_group_snapshots_following_updated_volumes(self):
"""Test deleting group snapshot following updated volumes
1. Create volume type "volume_type1"
2. Create group type "group_type1"
3. Create group "group1" with "group_type1" and "volume_type1"
4. Create 2 volumes "volume1" and "volume2"
with "volume_type1" and "group1"
5. For each created volume, removing and then adding back to "group1"
6. Create group snapshot "group_snapshot1" with "group1"
7. Check snapshots created from "volume1" and "volume2" reach
available status
8. Delete "group_snapshot1"
9. Check snapshots created from "volume1" and "volume2" are deleted
"""
volume_type = self.create_volume_type()
group_type = self.create_group_type()
# Create a volume group
grp = self.create_group(group_type=group_type['id'],
volume_types=[volume_type['id']])
prefix = CONF.resource_name_prefix
# Note: When dealing with consistency groups all volumes must
# reside on the same backend. Adding volumes to the same consistency
# group from multiple backends isn't supported. In order to ensure all
# volumes share the same backend, all volumes must share same
# volume-type and group id.
volume_list = []
for _ in range(2):
# Create volume is instance level, can't be deleted before group.
# Volume delete handled by delete_group method, cleanup method.
params = {'name': data_utils.rand_name(prefix=prefix,
name="volume"),
'volume_type': volume_type['id'],
'group_id': grp['id'],
'size': CONF.volume.volume_size}
volume = self.volumes_client.create_volume(**params)['volume']
waiters.wait_for_volume_resource_status(
self.volumes_client, volume['id'], 'available')
volume_list.append(volume['id'])
for vol in volume_list:
self.groups_client.update_group(grp['id'],
remove_volumes=vol)
waiters.wait_for_volume_resource_status(
self.groups_client, grp['id'], 'available')
self.groups_client.update_group(grp['id'],
add_volumes=vol)
waiters.wait_for_volume_resource_status(
self.groups_client, grp['id'], 'available')
# Verify the created volumes are associated with consistency group
vols = self.volumes_client.list_volumes(detail=True)['volumes']
grp_vols = [v for v in vols if v['group_id'] == grp['id']]
self.assertEqual(2, len(grp_vols))
# Create a snapshot group
group_snapshot = self._create_group_snapshot(group_id=grp['id'])
snapshots = self.snapshots_client.list_snapshots(
detail=True)['snapshots']
for snap in snapshots:
if snap['volume_id'] in volume_list:
waiters.wait_for_volume_resource_status(
self.snapshots_client, snap['id'], 'available')
# Delete a snapshot group
self._delete_group_snapshot(group_snapshot)
[docs]
class GroupSnapshotsV319Test(BaseGroupSnapshotsTest):
"""Test group snapshot with volume microversion greater than 3.18"""
volume_min_microversion = '3.19'
volume_max_microversion = 'latest'
[docs]
@decorators.idempotent_id('3b42c9b9-c984-4444-816e-ca2e1ed30b40')
def test_reset_group_snapshot_status(self):
"""Test resetting group snapshot status to creating/available/error"""
# Create volume type
volume_type = self.create_volume_type()
# Create group type
group_type = self.create_group_type()
# Create group
group = self.create_group(group_type=group_type['id'],
volume_types=[volume_type['id']])
prefix = CONF.resource_name_prefix
# Create volume is instance level, can not be deleted before group.
# Volume delete handled by delete_group method, cleanup method.
params = {'name': data_utils.rand_name(prefix=prefix, name="volume"),
'volume_type': volume_type['id'],
'group_id': group['id'],
'size': CONF.volume.volume_size}
volume = self.volumes_client.create_volume(**params)['volume']
waiters.wait_for_volume_resource_status(
self.volumes_client, volume['id'], 'available')
# Create group snapshot
group_snapshot = self._create_group_snapshot(group_id=group['id'])
snapshots = self.snapshots_client.list_snapshots(
detail=True)['snapshots']
for snap in snapshots:
if volume['id'] == snap['volume_id']:
waiters.wait_for_volume_resource_status(
self.snapshots_client, snap['id'], 'available')
# Reset group snapshot status
self.addCleanup(waiters.wait_for_volume_resource_status,
self.group_snapshots_client,
group_snapshot['id'], 'available')
self.addCleanup(
self.admin_group_snapshots_client.reset_group_snapshot_status,
group_snapshot['id'], 'available')
for status in ['creating', 'available', 'error']:
self.admin_group_snapshots_client.reset_group_snapshot_status(
group_snapshot['id'], status)
waiters.wait_for_volume_resource_status(
self.group_snapshots_client, group_snapshot['id'], status)
