"Fossies" - the Fresh Open Source Software Archive 
Member "cinder-13.0.7/cinder/tests/unit/api/v2/test_types.py" (4 Oct 2019, 20708 Bytes) of package /linux/misc/openstack/cinder-13.0.7.tar.gz:
As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style:
standard) with prefixed line numbers.
Alternatively you can here
view or
download the uninterpreted source code file.
See also the last
Fossies "Diffs" side-by-side code changes report for "test_types.py":
14.0.2_vs_15.0.0.
1 # Copyright 2011 OpenStack Foundation
2 # All Rights Reserved.
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
5 # not use this file except in compliance with the License. You may obtain
6 # a copy of the License at
7 #
8 # http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 # License for the specific language governing permissions and limitations
14 # under the License.
15
16 import uuid
17
18 import mock
19 from oslo_utils import timeutils
20 import six
21 import webob
22
23 from cinder.api.v2 import types
24 from cinder.api.v2.views import types as views_types
25 from cinder import context
26 from cinder import exception
27 from cinder.policies import volume_type as type_policy
28 from cinder import test
29 from cinder.tests.unit.api import fakes
30 from cinder.tests.unit import fake_constants as fake
31 from cinder.volume import volume_types
32
33
34 def fake_volume_type(id):
35 specs = {
36 "key1": "value1",
37 "key2": "value2",
38 "key3": "value3",
39 "key4": "value4",
40 "key5": "value5"
41 }
42 return dict(
43 id=id,
44 name='vol_type_%s' % six.text_type(id),
45 description='vol_type_desc_%s' % six.text_type(id),
46 extra_specs=specs,
47 )
48
49
50 def return_volume_types_get_all_types(context, filters=None, marker=None,
51 limit=None, sort_keys=None,
52 sort_dirs=None, offset=None,
53 list_result=False):
54 result = dict(vol_type_1=fake_volume_type(1),
55 vol_type_2=fake_volume_type(2),
56 vol_type_3=fake_volume_type(3)
57 )
58 if list_result:
59 return list(result.values())
60 return result
61
62
63 def return_empty_volume_types_get_all_types(context, filters=None, marker=None,
64 limit=None, sort_keys=None,
65 sort_dirs=None, offset=None,
66 list_result=False):
67 if list_result:
68 return []
69 return {}
70
71
72 def return_volume_types_get_volume_type(context, id):
73 if id == fake.WILL_NOT_BE_FOUND_ID:
74 raise exception.VolumeTypeNotFound(volume_type_id=id)
75 return fake_volume_type(id)
76
77
78 def return_volume_types_get_default():
79 return fake_volume_type(1)
80
81
82 class VolumeTypesApiTest(test.TestCase):
83
84 def _create_volume_type(self, volume_type_name, extra_specs=None,
85 is_public=True, projects=None):
86 return volume_types.create(self.ctxt, volume_type_name, extra_specs,
87 is_public, projects).get('id')
88
89 def setUp(self):
90 super(VolumeTypesApiTest, self).setUp()
91 self.controller = types.VolumeTypesController()
92 self.ctxt = context.RequestContext(user_id=fake.USER_ID,
93 project_id=fake.PROJECT_ID,
94 is_admin=True)
95 self.mock_authorize = self.patch(
96 'cinder.context.RequestContext.authorize')
97 self.type_id1 = self._create_volume_type('volume_type1',
98 {'key1': 'value1'})
99 self.type_id2 = self._create_volume_type('volume_type2',
100 {'key2': 'value2'})
101 self.type_id3 = self._create_volume_type('volume_type3',
102 {'key3': 'value3'}, False,
103 [fake.PROJECT_ID])
104
105 def test_volume_types_index(self):
106 self.mock_object(volume_types, 'get_all_types',
107 return_volume_types_get_all_types)
108
109 req = fakes.HTTPRequest.blank('/v2/%s/types' % fake.PROJECT_ID,
110 use_admin_context=True)
111 res_dict = self.controller.index(req)
112
113 self.assertEqual(3, len(res_dict['volume_types']))
114
115 expected_names = ['vol_type_1', 'vol_type_2', 'vol_type_3']
116 actual_names = map(lambda e: e['name'], res_dict['volume_types'])
117 self.assertEqual(set(expected_names), set(actual_names))
118 for entry in res_dict['volume_types']:
119 self.assertEqual('value1', entry['extra_specs']['key1'])
120 self.mock_authorize.assert_any_call(type_policy.GET_ALL_POLICY)
121
122 def test_volume_types_index_no_data(self):
123 self.mock_object(volume_types, 'get_all_types',
124 return_empty_volume_types_get_all_types)
125
126 req = fakes.HTTPRequest.blank('/v2/%s/types' % fake.PROJECT_ID)
127 res_dict = self.controller.index(req)
128
129 self.assertEqual(0, len(res_dict['volume_types']))
130
131 def test_volume_types_index_with_limit(self):
132 req = fakes.HTTPRequest.blank('/v2/%s/types?limit=1' % fake.PROJECT_ID)
133 req.environ['cinder.context'] = self.ctxt
134 res = self.controller.index(req)
135
136 self.assertEqual(1, len(res['volume_types']))
137 self.assertEqual(self.type_id3, res['volume_types'][0]['id'])
138
139 expect_next_link = ('http://localhost/v2/%s/types?limit=1'
140 '&marker=%s' %
141 (fake.PROJECT_ID, res['volume_types'][0]['id']))
142 self.assertEqual(expect_next_link, res['volume_type_links'][0]['href'])
143
144 def test_volume_types_index_with_offset(self):
145 req = fakes.HTTPRequest.blank(
146 '/v2/%s/types?offset=1' % fake.PROJECT_ID)
147 req.environ['cinder.context'] = self.ctxt
148 res = self.controller.index(req)
149
150 self.assertEqual(2, len(res['volume_types']))
151
152 def test_volume_types_index_with_offset_out_of_range(self):
153 url = '/v2/%s/types?offset=424366766556787' % fake.PROJECT_ID
154 req = fakes.HTTPRequest.blank(url)
155 self.assertRaises(webob.exc.HTTPBadRequest,
156 self.controller.index, req)
157
158 def test_volume_types_index_with_limit_and_offset(self):
159 req = fakes.HTTPRequest.blank(
160 '/v2/%s/types?limit=2&offset=1' % fake.PROJECT_ID)
161 req.environ['cinder.context'] = self.ctxt
162 res = self.controller.index(req)
163
164 self.assertEqual(2, len(res['volume_types']))
165 self.assertEqual(self.type_id2, res['volume_types'][0]['id'])
166 self.assertEqual(self.type_id1, res['volume_types'][1]['id'])
167
168 def test_volume_types_index_with_limit_and_marker(self):
169 req = fakes.HTTPRequest.blank('/v2/%s/types?limit=1'
170 '&marker=%s' %
171 (fake.PROJECT_ID,
172 self.type_id2))
173 req.environ['cinder.context'] = self.ctxt
174 res = self.controller.index(req)
175
176 self.assertEqual(1, len(res['volume_types']))
177 self.assertEqual(self.type_id1, res['volume_types'][0]['id'])
178
179 def test_volume_types_index_with_valid_filter(self):
180 req = fakes.HTTPRequest.blank(
181 '/v2/%s/types?is_public=True' % fake.PROJECT_ID)
182 req.environ['cinder.context'] = self.ctxt
183 res = self.controller.index(req)
184
185 self.assertEqual(3, len(res['volume_types']))
186 self.assertEqual(self.type_id3, res['volume_types'][0]['id'])
187 self.assertEqual(self.type_id2, res['volume_types'][1]['id'])
188 self.assertEqual(self.type_id1, res['volume_types'][2]['id'])
189
190 def test_volume_types_index_with_invalid_filter(self):
191 req = fakes.HTTPRequest.blank(
192 '/v2/%s/types?id=%s' % (fake.PROJECT_ID, self.type_id1))
193 req.environ['cinder.context'] = context.RequestContext(
194 user_id=fake.USER_ID, project_id=fake.PROJECT_ID, is_admin=False)
195 res = self.controller.index(req)
196
197 self.assertEqual(3, len(res['volume_types']))
198
199 def test_volume_types_index_with_sort_keys(self):
200 req = fakes.HTTPRequest.blank('/v2/%s/types?sort=id' % fake.PROJECT_ID)
201 req.environ['cinder.context'] = self.ctxt
202 res = self.controller.index(req)
203 expect_result = [self.type_id1, self.type_id2, self.type_id3]
204 expect_result.sort(reverse=True)
205
206 self.assertEqual(3, len(res['volume_types']))
207 self.assertEqual(expect_result[0], res['volume_types'][0]['id'])
208 self.assertEqual(expect_result[1], res['volume_types'][1]['id'])
209 self.assertEqual(expect_result[2], res['volume_types'][2]['id'])
210
211 def test_volume_types_index_with_sort_and_limit(self):
212 req = fakes.HTTPRequest.blank(
213 '/v2/%s/types?sort=id&limit=2' % fake.PROJECT_ID)
214 req.environ['cinder.context'] = self.ctxt
215 res = self.controller.index(req)
216 expect_result = [self.type_id1, self.type_id2, self.type_id3]
217 expect_result.sort(reverse=True)
218
219 self.assertEqual(2, len(res['volume_types']))
220 self.assertEqual(expect_result[0], res['volume_types'][0]['id'])
221 self.assertEqual(expect_result[1], res['volume_types'][1]['id'])
222
223 def test_volume_types_index_with_sort_keys_and_sort_dirs(self):
224 req = fakes.HTTPRequest.blank(
225 '/v2/%s/types?sort=id:asc' % fake.PROJECT_ID)
226 req.environ['cinder.context'] = self.ctxt
227 res = self.controller.index(req)
228 expect_result = [self.type_id1, self.type_id2, self.type_id3]
229 expect_result.sort()
230
231 self.assertEqual(3, len(res['volume_types']))
232 self.assertEqual(expect_result[0], res['volume_types'][0]['id'])
233 self.assertEqual(expect_result[1], res['volume_types'][1]['id'])
234 self.assertEqual(expect_result[2], res['volume_types'][2]['id'])
235
236 def test_volume_types_show(self):
237 self.mock_object(volume_types, 'get_volume_type',
238 return_volume_types_get_volume_type)
239
240 type_id = str(uuid.uuid4())
241 req = fakes.HTTPRequest.blank('/v2/%s/types/' % fake.PROJECT_ID
242 + type_id)
243 res_dict = self.controller.show(req, type_id)
244
245 self.assertEqual(1, len(res_dict))
246 self.assertEqual(type_id, res_dict['volume_type']['id'])
247 type_name = 'vol_type_' + type_id
248 self.assertEqual(type_name, res_dict['volume_type']['name'])
249 self.mock_authorize.assert_any_call(
250 type_policy.GET_POLICY, target_obj=mock.ANY)
251
252 def test_volume_types_show_not_found(self):
253 self.mock_object(volume_types, 'get_volume_type',
254 return_volume_types_get_volume_type)
255
256 req = fakes.HTTPRequest.blank('/v2/%s/types/%s' %
257 (fake.PROJECT_ID,
258 fake.WILL_NOT_BE_FOUND_ID))
259 self.assertRaises(exception.VolumeTypeNotFound, self.controller.show,
260 req, fake.WILL_NOT_BE_FOUND_ID)
261
262 def test_get_default(self):
263 self.mock_object(volume_types, 'get_default_volume_type',
264 return_volume_types_get_default)
265 req = fakes.HTTPRequest.blank('/v2/%s/types/default' % fake.PROJECT_ID)
266 req.method = 'GET'
267 res_dict = self.controller.show(req, 'default')
268 self.assertEqual(1, len(res_dict))
269 self.assertEqual('vol_type_1', res_dict['volume_type']['name'])
270 self.assertEqual('vol_type_desc_1',
271 res_dict['volume_type']['description'])
272
273 def test_get_default_not_found(self):
274 self.mock_object(volume_types, 'get_default_volume_type',
275 return_value={})
276 req = fakes.HTTPRequest.blank('/v2/%s/types/default' % fake.PROJECT_ID)
277 req.method = 'GET'
278
279 self.assertRaises(exception.VolumeTypeNotFound,
280 self.controller.show, req, 'default')
281
282 def test_view_builder_show(self):
283 view_builder = views_types.ViewBuilder()
284 self.mock_authorize.return_value = False
285 now = timeutils.utcnow().isoformat()
286 raw_volume_type = dict(
287 name='new_type',
288 description='new_type_desc',
289 qos_specs_id='new_id',
290 is_public=True,
291 deleted=False,
292 created_at=now,
293 updated_at=now,
294 extra_specs={},
295 deleted_at=None,
296 id=42,
297 )
298
299 request = fakes.HTTPRequest.blank("/v2")
300 output = view_builder.show(request, raw_volume_type)
301
302 self.assertIn('volume_type', output)
303 expected_volume_type = dict(
304 name='new_type',
305 description='new_type_desc',
306 is_public=True,
307 id=42,
308 )
309 self.assertDictEqual(expected_volume_type, output['volume_type'])
310
311 def test_view_builder_show_admin(self):
312 view_builder = views_types.ViewBuilder()
313 self.mock_authorize.return_value = True
314 now = timeutils.utcnow().isoformat()
315 raw_volume_type = dict(
316 name='new_type',
317 description='new_type_desc',
318 qos_specs_id='new_id',
319 is_public=True,
320 deleted=False,
321 created_at=now,
322 updated_at=now,
323 extra_specs={},
324 deleted_at=None,
325 id=42,
326 )
327
328 request = fakes.HTTPRequest.blank("/v2", use_admin_context=True)
329 output = view_builder.show(request, raw_volume_type)
330
331 self.assertIn('volume_type', output)
332 expected_volume_type = dict(
333 name='new_type',
334 description='new_type_desc',
335 qos_specs_id='new_id',
336 is_public=True,
337 extra_specs={},
338 id=42,
339 )
340 self.assertDictEqual(expected_volume_type, output['volume_type'])
341
342 def test_view_builder_show_qos_specs_id_policy(self):
343 with mock.patch('cinder.context.RequestContext.authorize',
344 side_effect=[False, True]):
345 view_builder = views_types.ViewBuilder()
346 now = timeutils.utcnow().isoformat()
347 raw_volume_type = dict(
348 name='new_type',
349 description='new_type_desc',
350 qos_specs_id='new_id',
351 is_public=True,
352 deleted=False,
353 created_at=now,
354 updated_at=now,
355 extra_specs={},
356 deleted_at=None,
357 id=42,
358 )
359
360 request = fakes.HTTPRequest.blank("/v2")
361 output = view_builder.show(request, raw_volume_type)
362
363 self.assertIn('volume_type', output)
364 expected_volume_type = dict(
365 name='new_type',
366 description='new_type_desc',
367 qos_specs_id='new_id',
368 is_public=True,
369 id=42,
370 )
371 self.assertDictEqual(expected_volume_type, output['volume_type'])
372
373 def test_view_builder_show_extra_specs_policy(self):
374 with mock.patch('cinder.context.RequestContext.authorize',
375 side_effect=[True, False]):
376 view_builder = views_types.ViewBuilder()
377 now = timeutils.utcnow().isoformat()
378 raw_volume_type = dict(
379 name='new_type',
380 description='new_type_desc',
381 qos_specs_id='new_id',
382 is_public=True,
383 deleted=False,
384 created_at=now,
385 updated_at=now,
386 extra_specs={},
387 deleted_at=None,
388 id=42,
389 )
390
391 request = fakes.HTTPRequest.blank("/v2")
392 output = view_builder.show(request, raw_volume_type)
393
394 self.assertIn('volume_type', output)
395 expected_volume_type = dict(
396 name='new_type',
397 description='new_type_desc',
398 extra_specs={},
399 is_public=True,
400 id=42,
401 )
402 self.assertDictEqual(expected_volume_type, output['volume_type'])
403
404 with mock.patch('cinder.context.RequestContext.authorize',
405 side_effect=[False, False]):
406 view_builder = views_types.ViewBuilder()
407 now = timeutils.utcnow().isoformat()
408 raw_volume_type = dict(
409 name='new_type',
410 description='new_type_desc',
411 qos_specs_id='new_id',
412 is_public=True,
413 deleted=False,
414 created_at=now,
415 updated_at=now,
416 extra_specs={},
417 deleted_at=None,
418 id=42,
419 )
420
421 request = fakes.HTTPRequest.blank("/v2")
422 output = view_builder.show(request, raw_volume_type)
423
424 self.assertIn('volume_type', output)
425 expected_volume_type = dict(
426 name='new_type',
427 description='new_type_desc',
428 is_public=True,
429 id=42,
430 )
431 self.assertDictEqual(expected_volume_type, output['volume_type'])
432
433 def test_view_builder_show_pass_all_policy(self):
434 with mock.patch('cinder.context.RequestContext.authorize',
435 side_effect=[True, True]):
436 view_builder = views_types.ViewBuilder()
437 now = timeutils.utcnow().isoformat()
438 raw_volume_type = dict(
439 name='new_type',
440 description='new_type_desc',
441 qos_specs_id='new_id',
442 is_public=True,
443 deleted=False,
444 created_at=now,
445 updated_at=now,
446 extra_specs={},
447 deleted_at=None,
448 id=42,
449 )
450
451 request = fakes.HTTPRequest.blank("/v2")
452 output = view_builder.show(request, raw_volume_type)
453
454 self.assertIn('volume_type', output)
455 expected_volume_type = dict(
456 name='new_type',
457 description='new_type_desc',
458 qos_specs_id='new_id',
459 extra_specs={},
460 is_public=True,
461 id=42,
462 )
463 self.assertDictEqual(expected_volume_type, output['volume_type'])
464
465 def test_view_builder_list(self):
466 view_builder = views_types.ViewBuilder()
467 self.mock_authorize.return_value = False
468 now = timeutils.utcnow().isoformat()
469 raw_volume_types = []
470 for i in range(0, 10):
471 raw_volume_types.append(
472 dict(
473 name='new_type',
474 description='new_type_desc',
475 qos_specs_id='new_id',
476 is_public=True,
477 deleted=False,
478 created_at=now,
479 updated_at=now,
480 extra_specs={},
481 deleted_at=None,
482 id=42 + i
483 )
484 )
485
486 request = fakes.HTTPRequest.blank("/v2")
487 output = view_builder.index(request, raw_volume_types)
488
489 self.assertIn('volume_types', output)
490 for i in range(0, 10):
491 expected_volume_type = dict(
492 name='new_type',
493 description='new_type_desc',
494 is_public=True,
495 id=42 + i
496 )
497 self.assertDictEqual(expected_volume_type,
498 output['volume_types'][i])
499
500 def test_view_builder_list_admin(self):
501 view_builder = views_types.ViewBuilder()
502
503 now = timeutils.utcnow().isoformat()
504 raw_volume_types = []
505 for i in range(0, 10):
506 raw_volume_types.append(
507 dict(
508 name='new_type',
509 description='new_type_desc',
510 qos_specs_id='new_id',
511 is_public=True,
512 deleted=False,
513 created_at=now,
514 updated_at=now,
515 extra_specs={},
516 deleted_at=None,
517 id=42 + i
518 )
519 )
520
521 request = fakes.HTTPRequest.blank("/v2", use_admin_context=True)
522 output = view_builder.index(request, raw_volume_types)
523
524 self.assertIn('volume_types', output)
525 for i in range(0, 10):
526 expected_volume_type = dict(
527 name='new_type',
528 description='new_type_desc',
529 qos_specs_id='new_id',
530 is_public=True,
531 extra_specs={},
532 id=42 + i
533 )
534 self.assertDictEqual(expected_volume_type,
535 output['volume_types'][i])