1 # Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
2 # and others. All rights reserved.
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at:
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
19 from cinderclient.exceptions import NotFound, BadRequest
21 from snaps.config.volume import VolumeConfig
22 from snaps.config.volume_type import (
23 VolumeTypeConfig, ControlLocation, VolumeTypeEncryptionConfig)
24 from snaps.config.qos import Consumer, QoSConfig
25 from snaps.openstack import create_volume
26 from snaps.openstack.create_qos import Consumer
27 from snaps.openstack.tests import validation_utils
28 from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
29 from snaps.openstack.utils import cinder_utils, keystone_utils
31 __author__ = 'spisarski'
34 logger = logging.getLogger('cinder_utils_tests')
37 class CinderSmokeTests(OSComponentTestCase):
39 Tests to ensure that the neutron client can communicate with the cloud
42 def test_cinder_connect_success(self):
44 Tests to ensure that the proper credentials can connect.
46 cinder = cinder_utils.cinder_client(self.os_creds)
47 volumes = cinder.volumes.list()
48 self.assertIsNotNone(volumes)
49 self.assertTrue(isinstance(volumes, list))
51 def test_cinder_connect_fail(self):
53 Tests to ensure that the improper credentials cannot connect.
55 from snaps.openstack.os_credentials import OSCreds
57 with self.assertRaises(Exception):
58 cinder = cinder_utils.cinder_client(OSCreds(
59 username='user', password='pass', auth_url='url',
60 project_name='project'))
64 class CinderUtilsVolumeTests(OSComponentTestCase):
66 Test for the CreateVolume class defined in create_volume.py
71 Instantiates the CreateVolume object that is responsible for
72 downloading and creating an OS volume file within OpenStack
75 self.volume_name = self.__class__.__name__ + '-' + str(guid)
77 self.cinder = cinder_utils.cinder_client(self.os_creds)
78 self.keystone = keystone_utils.keystone_client(self.os_creds)
82 Cleans the remote OpenStack objects
86 cinder_utils.delete_volume(self.cinder, self.volume)
90 self.assertTrue(volume_deleted(self.cinder, self.volume))
92 def test_create_simple_volume(self):
94 Tests the cinder_utils.create_volume()
96 volume_settings = VolumeConfig(name=self.volume_name)
97 self.volume = cinder_utils.create_volume(
98 self.cinder, self.keystone, volume_settings)
99 self.assertIsNotNone(self.volume)
100 self.assertEqual(self.volume_name, self.volume.name)
102 self.assertTrue(volume_active(self.cinder, self.volume))
104 volume = cinder_utils.get_volume(
105 self.cinder, self.keystone, volume_settings=volume_settings,
106 project_name=self.os_creds.project_name)
107 self.assertIsNotNone(volume)
108 validation_utils.objects_equivalent(self.volume, volume)
110 def test_create_delete_volume(self):
112 Tests the cinder_utils.create_volume()
114 volume_settings = VolumeConfig(name=self.volume_name)
115 self.volume = cinder_utils.create_volume(
116 self.cinder, self.keystone, volume_settings)
117 self.assertIsNotNone(self.volume)
118 self.assertEqual(self.volume_name, self.volume.name)
120 self.assertTrue(volume_active(self.cinder, self.volume))
122 volume = cinder_utils.get_volume(
123 self.cinder, self.keystone, volume_settings=volume_settings,
124 project_name=self.os_creds.project_name)
125 self.assertIsNotNone(volume)
126 validation_utils.objects_equivalent(self.volume, volume)
128 cinder_utils.delete_volume(self.cinder, self.volume)
129 self.assertTrue(volume_deleted(self.cinder, self.volume))
131 cinder_utils.get_volume(
132 self.cinder, self.keystone, volume_settings,
133 project_name=self.os_creds.project_name))
136 def volume_active(cinder, volume):
138 Returns true if volume becomes active
143 end_time = time.time() + create_volume.VOLUME_ACTIVE_TIMEOUT
144 while time.time() < end_time:
145 status = cinder_utils.get_volume_status(cinder, volume)
146 if status == create_volume.STATUS_ACTIVE:
148 elif status == create_volume.STATUS_FAILED:
155 def volume_deleted(cinder, volume):
157 Returns true if volume becomes active
162 end_time = time.time() + create_volume.VOLUME_ACTIVE_TIMEOUT
163 while time.time() < end_time:
165 status = cinder_utils.get_volume_status(cinder, volume)
166 if status == create_volume.STATUS_DELETED:
176 class CinderUtilsQoSTests(OSComponentTestCase):
178 Test for the CreateQos class defined in create_qos.py
183 Creates objects for testing cinder_utils.py
186 self.qos_name = self.__class__.__name__ + '-' + str(guid)
187 self.specs = {'foo': 'bar '}
189 self.cinder = cinder_utils.cinder_client(self.os_creds)
193 Cleans the remote OpenStack objects
197 cinder_utils.delete_qos(self.cinder, self.qos)
201 def test_create_qos_both(self):
203 Tests the cinder_utils.create_qos()
205 qos_settings = QoSConfig(
206 name=self.qos_name, specs=self.specs, consumer=Consumer.both)
207 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
208 self.assertIsNotNone(self.qos)
210 qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
211 self.assertIsNotNone(qos1)
212 validation_utils.objects_equivalent(self.qos, qos1)
214 qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
215 self.assertIsNotNone(qos2)
216 validation_utils.objects_equivalent(self.qos, qos2)
218 def test_create_qos_front(self):
220 Tests the cinder_utils.create_qos()
222 qos_settings = QoSConfig(
223 name=self.qos_name, specs=self.specs, consumer=Consumer.front_end)
224 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
225 self.assertIsNotNone(self.qos)
227 qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
228 self.assertIsNotNone(qos1)
229 validation_utils.objects_equivalent(self.qos, qos1)
231 qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
232 self.assertIsNotNone(qos2)
233 validation_utils.objects_equivalent(self.qos, qos2)
235 def test_create_qos_back(self):
237 Tests the cinder_utils.create_qos()
239 qos_settings = QoSConfig(
240 name=self.qos_name, specs=self.specs, consumer=Consumer.back_end)
241 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
242 self.assertIsNotNone(self.qos)
244 qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
245 self.assertIsNotNone(qos1)
246 validation_utils.objects_equivalent(self.qos, qos1)
248 qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
249 self.assertIsNotNone(qos2)
250 validation_utils.objects_equivalent(self.qos, qos2)
252 def test_create_delete_qos(self):
254 Tests the cinder_utils.create_qos()
256 qos_settings = QoSConfig(name=self.qos_name, consumer=Consumer.both)
257 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
258 self.assertIsNotNone(self.qos)
259 self.assertEqual(self.qos_name, self.qos.name)
261 qos = cinder_utils.get_qos(
262 self.cinder, qos_settings=qos_settings)
263 self.assertIsNotNone(qos)
264 validation_utils.objects_equivalent(self.qos, qos)
266 cinder_utils.delete_qos(self.cinder, self.qos)
267 self.assertIsNone(cinder_utils.get_qos(
268 self.cinder, qos_settings=qos_settings))
271 class CinderUtilsSimpleVolumeTypeTests(OSComponentTestCase):
273 Tests the creation of a Volume Type without any external settings such as
274 QoS Specs or encryption
279 Instantiates the CreateVolume object that is responsible for
280 downloading and creating an OS volume file within OpenStack
283 volume_type_name = self.__class__.__name__ + '-' + str(guid)
284 self.volume_type_settings = VolumeTypeConfig(name=volume_type_name)
285 self.volume_type = None
286 self.cinder = cinder_utils.cinder_client(self.os_creds)
290 Cleans the remote OpenStack objects
294 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
298 def test_create_simple_volume_type(self):
300 Tests the cinder_utils.create_volume_type(), get_volume_type(), and
301 get_volume_type_by_id()
303 self.volume_type = cinder_utils.create_volume_type(
304 self.cinder, self.volume_type_settings)
305 self.assertIsNotNone(self.volume_type)
306 self.assertEqual(self.volume_type_settings.name, self.volume_type.name)
308 volume_type1 = cinder_utils.get_volume_type(
309 self.cinder, volume_type_settings=self.volume_type_settings)
310 self.assertEquals(self.volume_type, volume_type1)
311 self.assertEquals(self.volume_type_settings.public,
314 volume_type2 = cinder_utils.get_volume_type_by_id(
315 self.cinder, volume_type1.id)
316 self.assertEquals(self.volume_type, volume_type2)
317 self.assertIsNone(self.volume_type.encryption)
319 def test_create_delete_volume_type(self):
321 Primarily tests the cinder_utils.delete_volume()
323 self.volume_type = cinder_utils.create_volume_type(
324 self.cinder, self.volume_type_settings)
325 self.assertIsNotNone(self.volume_type)
326 self.assertEqual(self.volume_type_settings.name, self.volume_type.name)
328 volume_type = cinder_utils.get_volume_type(
329 self.cinder, volume_type_settings=self.volume_type_settings)
330 self.assertIsNotNone(volume_type)
331 validation_utils.objects_equivalent(self.volume_type, volume_type)
332 self.assertIsNone(self.volume_type.encryption)
334 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
335 self.assertIsNone(cinder_utils.get_volume_type_by_id(
336 self.cinder, self.volume_type.id))
339 class CinderUtilsAddEncryptionTests(OSComponentTestCase):
341 Tests the creation of an encryption and association to and existing
347 Instantiates the CreateVolume object that is responsible for
348 downloading and creating an OS volume file within OpenStack
351 self.encryption_name = self.__class__.__name__ + '-' + str(guid)
352 self.encryption = None
354 self.cinder = cinder_utils.cinder_client(self.os_creds)
356 volume_type_name = self.__class__.__name__ + '-' + str(guid) + '-type'
357 self.volume_type = cinder_utils.create_volume_type(
358 self.cinder, VolumeTypeConfig(name=volume_type_name))
362 Cleans the remote OpenStack objects
366 cinder_utils.delete_volume_type_encryption(
367 self.cinder, self.volume_type)
373 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
377 def test_create_simple_encryption(self):
379 Tests the cinder_utils.create_volume_encryption(),
380 get_volume_encryption(), and get_volume_encryption_by_id()
382 encryption_settings = VolumeTypeEncryptionConfig(
383 name=self.encryption_name, provider_class='foo',
384 control_location=ControlLocation.front_end)
385 self.encryption = cinder_utils.create_volume_encryption(
386 self.cinder, self.volume_type, encryption_settings)
387 self.assertIsNotNone(self.encryption)
388 self.assertEqual('foo', self.encryption.provider)
389 self.assertEqual(ControlLocation.front_end.value,
390 self.encryption.control_location)
392 encryption1 = cinder_utils.get_volume_encryption_by_type(
393 self.cinder, self.volume_type)
394 self.assertEquals(self.encryption, encryption1)
396 def test_create_delete_encryption(self):
398 Primarily tests the cinder_utils.delete_volume_type_encryption()
400 encryption_settings = VolumeTypeEncryptionConfig(
401 name=self.encryption_name, provider_class='LuksEncryptor',
402 control_location=ControlLocation.back_end)
403 self.encryption = cinder_utils.create_volume_encryption(
404 self.cinder, self.volume_type, encryption_settings)
405 self.assertIsNotNone(self.encryption)
406 self.assertEqual('LuksEncryptor', self.encryption.provider)
407 self.assertEqual(ControlLocation.back_end.value,
408 self.encryption.control_location)
410 encryption1 = cinder_utils.get_volume_encryption_by_type(
411 self.cinder, self.volume_type)
412 self.assertEquals(self.encryption, encryption1)
414 cinder_utils.delete_volume_type_encryption(
415 self.cinder, self.volume_type)
417 encryption2 = cinder_utils.get_volume_encryption_by_type(
418 self.cinder, self.volume_type)
419 self.assertIsNone(encryption2)
421 def test_create_with_all_attrs(self):
423 Tests the cinder_utils.create_volume_encryption() with all valid
426 encryption_settings = VolumeTypeEncryptionConfig(
427 name=self.encryption_name, provider_class='foo',
428 cipher='bar', control_location=ControlLocation.back_end,
430 self.encryption = cinder_utils.create_volume_encryption(
431 self.cinder, self.volume_type, encryption_settings)
432 self.assertIsNotNone(self.encryption)
433 self.assertEqual('foo', self.encryption.provider)
434 self.assertEqual('bar', self.encryption.cipher)
435 self.assertEqual(1, self.encryption.key_size)
436 self.assertEqual(ControlLocation.back_end.value,
437 self.encryption.control_location)
439 encryption1 = cinder_utils.get_volume_encryption_by_type(
440 self.cinder, self.volume_type)
441 self.assertEquals(self.encryption, encryption1)
443 def test_create_bad_key_size(self):
445 Tests the cinder_utils.create_volume_encryption() raises an exception
446 when the provider class does not exist
448 encryption_settings = VolumeTypeEncryptionConfig(
449 name=self.encryption_name, provider_class='foo',
450 cipher='bar', control_location=ControlLocation.back_end,
453 with self.assertRaises(BadRequest):
454 self.encryption = cinder_utils.create_volume_encryption(
455 self.cinder, self.volume_type, encryption_settings)
458 class CinderUtilsVolumeTypeCompleteTests(OSComponentTestCase):
460 Tests to ensure that a volume type can have a QoS Spec added to it
465 Creates objects for testing cinder_utils.py
468 self.qos_name = self.__class__.__name__ + '-' + str(guid) + '-qos'
469 self.vol_type_name = self.__class__.__name__ + '-' + str(guid)
470 self.specs = {'foo': 'bar'}
471 self.cinder = cinder_utils.cinder_client(self.os_creds)
472 qos_settings = QoSConfig(
473 name=self.qos_name, specs=self.specs, consumer=Consumer.both)
474 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
475 self.volume_type = None
479 Cleans the remote OpenStack objects
482 if self.volume_type.encryption:
484 cinder_utils.delete_volume_type_encryption(
485 self.cinder, self.volume_type)
489 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
495 cinder_utils.delete_qos(self.cinder, self.qos)
499 def test_create_with_encryption(self):
501 Tests the cinder_utils.create_volume_type() where encryption has been
504 encryption_settings = VolumeTypeEncryptionConfig(
505 name='foo', provider_class='bar',
506 control_location=ControlLocation.back_end)
507 volume_type_settings = VolumeTypeConfig(
508 name=self.vol_type_name, encryption=encryption_settings)
509 self.volume_type = cinder_utils.create_volume_type(
510 self.cinder, volume_type_settings)
512 vol_encrypt = cinder_utils.get_volume_encryption_by_type(
513 self.cinder, self.volume_type)
514 self.assertIsNotNone(vol_encrypt)
515 self.assertIsNone(self.volume_type.qos_spec)
516 self.assertEqual(self.volume_type.encryption, vol_encrypt)
517 self.assertEqual(self.volume_type.id, vol_encrypt.volume_type_id)
519 def test_create_with_qos(self):
521 Tests the cinder_utils.create_volume_type() with an associated QoS Spec
523 volume_type_settings = VolumeTypeConfig(
524 name=self.vol_type_name, qos_spec_name=self.qos_name)
525 self.volume_type = cinder_utils.create_volume_type(
526 self.cinder, volume_type_settings)
528 self.assertIsNotNone(self.volume_type)
529 self.assertIsNone(self.volume_type.encryption)
530 self.assertIsNotNone(self.volume_type.qos_spec)
531 self.assertEqual(self.qos.id, self.volume_type.qos_spec.id)
533 def test_create_with_invalid_qos(self):
535 Tests the cinder_utils.create_volume_type() when the QoS Spec name
538 volume_type_settings = VolumeTypeConfig(
539 name=self.vol_type_name, qos_spec_name='foo')
541 self.volume_type = cinder_utils.create_volume_type(
542 self.cinder, volume_type_settings)
544 self.assertIsNone(self.volume_type.qos_spec)
546 def test_create_with_qos_and_encryption(self):
548 Tests the cinder_utils.create_volume_type() with encryption and an
551 encryption_settings = VolumeTypeEncryptionConfig(
552 name='foo', provider_class='bar',
553 control_location=ControlLocation.back_end)
554 volume_type_settings = VolumeTypeConfig(
555 name=self.vol_type_name, qos_spec_name=self.qos_name,
556 encryption=encryption_settings)
557 self.volume_type = cinder_utils.create_volume_type(
558 self.cinder, volume_type_settings)
560 self.assertIsNotNone(self.volume_type)
561 vol_encrypt = cinder_utils.get_volume_encryption_by_type(
562 self.cinder, self.volume_type)
563 self.assertIsNotNone(vol_encrypt)
564 self.assertEqual(self.volume_type.id, vol_encrypt.volume_type_id)
565 self.assertIsNotNone(self.volume_type.qos_spec)
566 self.assertEqual(self.qos.id, self.volume_type.qos_spec.id)