/
test_list_image_filters.py
333 lines (277 loc) · 13.9 KB
/
test_list_image_filters.py
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
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
# 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 time
import io
import testtools
from tempest.api.compute 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
from tempest.lib import exceptions
CONF = config.CONF
class ListImageFiltersTestJSON(base.BaseV2ComputeTest):
"""Test listing server images with compute microversion less than 2.36"""
max_microversion = '2.35'
@classmethod
def skip_checks(cls):
super(ListImageFiltersTestJSON, cls).skip_checks()
if not CONF.service_available.glance:
skip_msg = ("%s skipped as glance is not available" % cls.__name__)
raise cls.skipException(skip_msg)
@classmethod
def setup_clients(cls):
super(ListImageFiltersTestJSON, cls).setup_clients()
cls.client = cls.compute_images_client
if CONF.image_feature_enabled.api_v2:
cls.glance_client = cls.os_primary.image_client_v2
else:
raise exceptions.InvalidConfiguration(
'api_v2 must be True in [image-feature-enabled].')
@classmethod
def resource_setup(cls):
super(ListImageFiltersTestJSON, cls).resource_setup()
def _create_image():
image_name_kwargs = {
'prefix': CONF.resource_name_prefix,
'name': cls.__name__ + '-image'
}
params = {
'name': data_utils.rand_name(**image_name_kwargs),
'container_format': 'bare',
'disk_format': 'raw',
'visibility': 'private'
}
body = cls.glance_client.create_image(**params)
body = body['image'] if 'image' in body else body
image_id = body['id']
cls.addClassResourceCleanup(
test_utils.call_and_ignore_notfound_exc,
cls.compute_images_client.delete_image,
image_id)
# Wait 1 second between creation and upload to ensure a delta
# between created_at and updated_at.
time.sleep(1)
image_file = io.BytesIO((b'*' * 1024))
cls.glance_client.store_image_file(image_id, data=image_file)
waiters.wait_for_image_status(cls.client, image_id, 'ACTIVE')
body = cls.client.show_image(image_id)['image']
return body
# Create non-snapshot images via glance
cls.image1 = _create_image()
cls.image1_id = cls.image1['id']
cls.image2 = _create_image()
cls.image2_id = cls.image2['id']
cls.image3 = _create_image()
cls.image3_id = cls.image3['id']
if not CONF.compute_feature_enabled.snapshot:
return
# Create instances and snapshots via nova
cls.server1 = cls.create_test_server()
cls.server2 = cls.create_test_server(wait_until='ACTIVE')
# NOTE(sdague) this is faster than doing the sync wait_util on both
waiters.wait_for_server_status(cls.servers_client,
cls.server1['id'], 'ACTIVE')
# Create images to be used in the filter tests
cls.snapshot1 = cls.create_image_from_server(
cls.server1['id'], wait_until='ACTIVE')
cls.snapshot1_id = cls.snapshot1['id']
# Servers have a hidden property for when they are being imaged
# Performing back-to-back create image calls on a single
# server will sometimes cause failures
cls.snapshot3 = cls.create_image_from_server(
cls.server2['id'], wait_until='ACTIVE')
cls.snapshot3_id = cls.snapshot3['id']
# Wait for the server to be active after the image upload
cls.snapshot2 = cls.create_image_from_server(
cls.server1['id'], wait_until='ACTIVE')
cls.snapshot2_id = cls.snapshot2['id']
@decorators.idempotent_id('a3f5b513-aeb3-42a9-b18e-f091ef73254d')
def test_list_images_filter_by_status(self):
"""Test listing server images filtered by image status
The list of images should contain only images with the
provided image status.
"""
params = {'status': 'ACTIVE'}
images = self.client.list_images(**params)['images']
self.assertNotEmpty([i for i in images if i['id'] == self.image1_id])
self.assertNotEmpty([i for i in images if i['id'] == self.image2_id])
self.assertNotEmpty([i for i in images if i['id'] == self.image3_id])
@decorators.idempotent_id('33163b73-79f5-4d07-a7ea-9213bcc468ff')
def test_list_images_filter_by_name(self):
"""Test listing server images filtered by image name
The list of images should contain only images with the
provided image name.
"""
params = {'name': self.image1['name']}
images = self.client.list_images(**params)['images']
self.assertNotEmpty([i for i in images if i['id'] == self.image1_id])
self.assertEmpty([i for i in images if i['id'] == self.image2_id])
self.assertEmpty([i for i in images if i['id'] == self.image3_id])
@decorators.idempotent_id('9f238683-c763-45aa-b848-232ec3ce3105')
@testtools.skipUnless(CONF.compute_feature_enabled.snapshot,
'Snapshotting is not available.')
def test_list_images_filter_by_server_id(self):
"""Test listing images filtered by server id
The list of images should contain only images with the
provided server id.
"""
params = {'server': self.server1['id']}
images = self.client.list_images(**params)['images']
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot1_id],
"Failed to find image %s in images. "
"Got images %s" % (self.image1_id, images))
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot2_id])
self.assertEmpty([i for i in images if i['id'] == self.snapshot3_id])
@decorators.idempotent_id('05a377b8-28cf-4734-a1e6-2ab5c38bf606')
@testtools.skipUnless(CONF.compute_feature_enabled.snapshot,
'Snapshotting is not available.')
def test_list_images_filter_by_server_ref(self):
"""Test listing images filtered by server link href
The list of images should contain only images with the
provided server link href.
"""
server_links = self.server2['links']
# Try all server link types
for link in server_links:
params = {'server': link['href']}
images = self.client.list_images(**params)['images']
self.assertEmpty([i for i in images
if i['id'] == self.snapshot1_id])
self.assertEmpty([i for i in images
if i['id'] == self.snapshot2_id])
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot3_id])
@decorators.idempotent_id('e3356918-4d3e-4756-81d5-abc4524ba29f')
@testtools.skipUnless(CONF.compute_feature_enabled.snapshot,
'Snapshotting is not available.')
def test_list_images_filter_by_type(self):
"""Test listing images filtered by image type
The list of images should contain only images with the
provided image type.
"""
params = {'type': 'snapshot'}
images = self.client.list_images(**params)['images']
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot1_id])
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot2_id])
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot3_id])
self.assertEmpty([i for i in images if i['id'] == self.image_ref])
@decorators.idempotent_id('3a484ca9-67ba-451e-b494-7fcf28d32d62')
def test_list_images_limit_results(self):
"""Test listing images with limited count
If we use limit=1 when listing images, then only 1 image should be
returned.
"""
params = {'limit': '1'}
images = self.client.list_images(**params)['images']
self.assertEqual(1, len([x for x in images if 'id' in x]))
@decorators.idempotent_id('18bac3ae-da27-436c-92a9-b22474d13aab')
def test_list_images_filter_by_changes_since(self):
"""Test listing images filtered by changes-since
The list of images should contain only images updated since the
provided changes-since value.
"""
# Verify only updated images are returned in the detailed list
# Becoming ACTIVE will modify the updated time
# Filter by the image's created time
params = {'changes-since': self.image3['created']}
images = self.client.list_images(**params)['images']
found = [i for i in images if i['id'] == self.image3_id]
self.assertNotEmpty(found)
@decorators.idempotent_id('9b0ea018-6185-4f71-948a-a123a107988e')
def test_list_images_with_detail_filter_by_status(self):
"""Test listing server images details filtered by image status
The list of images should contain only images with the
provided image status.
"""
params = {'status': 'ACTIVE'}
images = self.client.list_images(detail=True, **params)['images']
self.assertNotEmpty([i for i in images if i['id'] == self.image1_id])
self.assertNotEmpty([i for i in images if i['id'] == self.image2_id])
self.assertNotEmpty([i for i in images if i['id'] == self.image3_id])
@decorators.idempotent_id('644ea267-9bd9-4f3b-af9f-dffa02396a17')
def test_list_images_with_detail_filter_by_name(self):
"""Test listing server images details filtered by image name
The list of images should contain only images with the
provided image name.
"""
params = {'name': self.image1['name']}
images = self.client.list_images(detail=True, **params)['images']
self.assertNotEmpty([i for i in images if i['id'] == self.image1_id])
self.assertEmpty([i for i in images if i['id'] == self.image2_id])
self.assertEmpty([i for i in images if i['id'] == self.image3_id])
@decorators.idempotent_id('ba2fa9a9-b672-47cc-b354-3b4c0600e2cb')
def test_list_images_with_detail_limit_results(self):
"""Test listing images details with limited count
If we use limit=1 when listing images with full details, then only 1
image should be returned.
"""
params = {'limit': '1'}
images = self.client.list_images(detail=True, **params)['images']
self.assertEqual(1, len(images))
@decorators.idempotent_id('8c78f822-203b-4bf6-8bba-56ebd551cf84')
@testtools.skipUnless(CONF.compute_feature_enabled.snapshot,
'Snapshotting is not available.')
def test_list_images_with_detail_filter_by_server_ref(self):
"""Test listing images details filtered by server link href
The list of images should contain only images with the
provided server link href.
"""
server_links = self.server2['links']
# Try all server link types
for link in server_links:
params = {'server': link['href']}
images = self.client.list_images(detail=True, **params)['images']
self.assertEmpty([i for i in images
if i['id'] == self.snapshot1_id])
self.assertEmpty([i for i in images
if i['id'] == self.snapshot2_id])
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot3_id])
@decorators.idempotent_id('888c0cc0-7223-43c5-9db0-b125fd0a393b')
@testtools.skipUnless(CONF.compute_feature_enabled.snapshot,
'Snapshotting is not available.')
def test_list_images_with_detail_filter_by_type(self):
"""Test listing images details filtered by image type
The list of images should contain only images with the
provided image type.
"""
params = {'type': 'snapshot'}
images = self.client.list_images(detail=True, **params)['images']
self.client.show_image(self.image_ref)
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot1_id])
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot2_id])
self.assertNotEmpty([i for i in images
if i['id'] == self.snapshot3_id])
self.assertEmpty([i for i in images if i['id'] == self.image_ref])
@decorators.idempotent_id('7d439e18-ac2e-4827-b049-7e18004712c4')
def test_list_images_with_detail_filter_by_changes_since(self):
"""Test listing images details filtered by changes-since
The list of images should contain only images updated since the
provided changes-since value.
"""
# Becoming ACTIVE will modify the updated time
# Filter by the image's created time
params = {'changes-since': self.image1['created']}
images = self.client.list_images(detail=True, **params)['images']
self.assertNotEmpty([i for i in images if i['id'] == self.image1_id])