1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
# Copyright 2013 Huawei Technologies Co.,LTD
# 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 import decorators
CONF = config.CONF
class SnapshotsActionsTest(base.BaseVolumeAdminTest):
"""Test volume snapshot actions"""
@classmethod
def skip_checks(cls):
super(SnapshotsActionsTest, cls).skip_checks()
if not CONF.volume_feature_enabled.snapshot:
raise cls.skipException("Cinder snapshot feature disabled")
@classmethod
def resource_setup(cls):
super(SnapshotsActionsTest, cls).resource_setup()
# Create a test shared volume for tests
cls.volume = cls.create_volume()
# Create a test shared snapshot for tests
cls.snapshot = cls.create_snapshot(volume_id=cls.volume['id'])
def tearDown(self):
# Set snapshot's status to available after test
status = 'available'
snapshot_id = self.snapshot['id']
self.admin_snapshots_client.reset_snapshot_status(snapshot_id,
status)
waiters.wait_for_volume_resource_status(self.snapshots_client,
snapshot_id, status)
super(SnapshotsActionsTest, self).tearDown()
def _create_reset_and_force_delete_temp_snapshot(self, status=None):
# Create snapshot, reset snapshot status,
# and force delete temp snapshot
temp_snapshot = self.create_snapshot(volume_id=self.volume['id'])
if status:
self.admin_snapshots_client.reset_snapshot_status(
temp_snapshot['id'], status)
waiters.wait_for_volume_resource_status(
self.snapshots_client, temp_snapshot['id'], status)
self.admin_snapshots_client.force_delete_snapshot(temp_snapshot['id'])
self.snapshots_client.wait_for_resource_deletion(temp_snapshot['id'])
def _get_progress_alias(self):
return 'os-extended-snapshot-attributes:progress'
@decorators.idempotent_id('3e13ca2f-48ea-49f3-ae1a-488e9180d535')
def test_reset_snapshot_status(self):
"""Test resetting snapshot status to creating"""
status = 'creating'
self.admin_snapshots_client.reset_snapshot_status(
self.snapshot['id'], status)
waiters.wait_for_volume_resource_status(self.snapshots_client,
self.snapshot['id'], status)
@decorators.idempotent_id('41288afd-d463-485e-8f6e-4eea159413eb')
def test_update_snapshot_status(self):
"""Test updating snapshot
Update snapshot status to 'error' and progress to '80%'.
"""
# Reset snapshot status to creating
status = 'creating'
self.admin_snapshots_client.reset_snapshot_status(
self.snapshot['id'], status)
waiters.wait_for_volume_resource_status(self.snapshots_client,
self.snapshot['id'], status)
# Update snapshot status to error
progress = '80%'
status = 'error'
progress_alias = self._get_progress_alias()
self.snapshots_client.update_snapshot_status(self.snapshot['id'],
status=status,
progress=progress)
snapshot_get = self.admin_snapshots_client.show_snapshot(
self.snapshot['id'])['snapshot']
self.assertEqual(status, snapshot_get['status'])
self.assertEqual(progress, snapshot_get[progress_alias])
@decorators.idempotent_id('05f711b6-e629-4895-8103-7ca069f2073a')
def test_snapshot_force_delete_when_snapshot_is_creating(self):
"""Test force delete when status of snapshot is creating"""
self._create_reset_and_force_delete_temp_snapshot('creating')
@decorators.idempotent_id('92ce8597-b992-43a1-8868-6316b22a969e')
def test_snapshot_force_delete_when_snapshot_is_deleting(self):
"""Test force delete when status of snapshot is deleting"""
self._create_reset_and_force_delete_temp_snapshot('deleting')
@decorators.idempotent_id('645a4a67-a1eb-4e8e-a547-600abac1525d')
def test_snapshot_force_delete_when_snapshot_is_error(self):
"""Test force delete when status of snapshot is error"""
self._create_reset_and_force_delete_temp_snapshot('error')
@decorators.idempotent_id('bf89080f-8129-465e-9327-b2f922666ba5')
def test_snapshot_force_delete_when_snapshot_is_error_deleting(self):
"""Test force delete when status of snapshot is error_deleting"""
self._create_reset_and_force_delete_temp_snapshot('error_deleting')
|
