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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
|
# 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)
|
