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.openstack import create_volume
22 from snaps.openstack.create_qos import QoSSettings, Consumer
23 from snaps.openstack.create_volume import VolumeSettings
24 from snaps.openstack.create_volume_type import (
25 VolumeTypeSettings, VolumeTypeEncryptionSettings, ControlLocation)
26 from snaps.openstack.tests import validation_utils
27 from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
28 from snaps.openstack.utils import cinder_utils
30 __author__ = 'spisarski'
33 logger = logging.getLogger('cinder_utils_tests')
36 class CinderSmokeTests(OSComponentTestCase):
38 Tests to ensure that the neutron client can communicate with the cloud
41 def test_cinder_connect_success(self):
43 Tests to ensure that the proper credentials can connect.
45 cinder = cinder_utils.cinder_client(self.os_creds)
46 volumes = cinder.volumes.list()
47 self.assertIsNotNone(volumes)
48 self.assertTrue(isinstance(volumes, list))
50 def test_cinder_connect_fail(self):
52 Tests to ensure that the improper credentials cannot connect.
54 from snaps.openstack.os_credentials import OSCreds
56 with self.assertRaises(Exception):
57 cinder = cinder_utils.cinder_client(OSCreds(
58 username='user', password='pass', auth_url='url',
59 project_name='project'))
63 class CinderUtilsVolumeTests(OSComponentTestCase):
65 Test for the CreateVolume class defined in create_volume.py
70 Instantiates the CreateVolume object that is responsible for
71 downloading and creating an OS volume file within OpenStack
74 self.volume_name = self.__class__.__name__ + '-' + str(guid)
76 self.cinder = cinder_utils.cinder_client(self.os_creds)
80 Cleans the remote OpenStack objects
84 cinder_utils.delete_volume(self.cinder, self.volume)
88 self.assertTrue(volume_deleted(self.cinder, self.volume))
90 def test_create_simple_volume(self):
92 Tests the cinder_utils.create_volume()
94 volume_settings = VolumeSettings(name=self.volume_name)
95 self.volume = cinder_utils.create_volume(
96 self.cinder, volume_settings)
97 self.assertIsNotNone(self.volume)
98 self.assertEqual(self.volume_name, self.volume.name)
100 self.assertTrue(volume_active(self.cinder, self.volume))
102 volume = cinder_utils.get_volume(
103 self.cinder, volume_settings=volume_settings)
104 self.assertIsNotNone(volume)
105 validation_utils.objects_equivalent(self.volume, volume)
107 def test_create_delete_volume(self):
109 Tests the cinder_utils.create_volume()
111 volume_settings = VolumeSettings(name=self.volume_name)
112 self.volume = cinder_utils.create_volume(
113 self.cinder, volume_settings)
114 self.assertIsNotNone(self.volume)
115 self.assertEqual(self.volume_name, self.volume.name)
117 self.assertTrue(volume_active(self.cinder, self.volume))
119 volume = cinder_utils.get_volume(
120 self.cinder, volume_settings=volume_settings)
121 self.assertIsNotNone(volume)
122 validation_utils.objects_equivalent(self.volume, volume)
124 cinder_utils.delete_volume(self.cinder, self.volume)
125 self.assertTrue(volume_deleted(self.cinder, self.volume))
127 cinder_utils.get_volume(self.cinder, volume_settings))
130 def volume_active(cinder, volume):
132 Returns true if volume becomes active
137 end_time = time.time() + create_volume.VOLUME_ACTIVE_TIMEOUT
138 while time.time() < end_time:
139 status = cinder_utils.get_volume_status(cinder, volume)
140 if status == create_volume.STATUS_ACTIVE:
142 elif status == create_volume.STATUS_FAILED:
149 def volume_deleted(cinder, volume):
151 Returns true if volume becomes active
156 end_time = time.time() + create_volume.VOLUME_ACTIVE_TIMEOUT
157 while time.time() < end_time:
159 status = cinder_utils.get_volume_status(cinder, volume)
160 if status == create_volume.STATUS_DELETED:
170 class CinderUtilsQoSTests(OSComponentTestCase):
172 Test for the CreateQos class defined in create_qos.py
177 Creates objects for testing cinder_utils.py
180 self.qos_name = self.__class__.__name__ + '-' + str(guid)
181 self.specs = {'foo': 'bar '}
183 self.cinder = cinder_utils.cinder_client(self.os_creds)
187 Cleans the remote OpenStack objects
191 cinder_utils.delete_qos(self.cinder, self.qos)
195 def test_create_qos_both(self):
197 Tests the cinder_utils.create_qos()
199 qos_settings = QoSSettings(name=self.qos_name, specs=self.specs,
200 consumer=Consumer.both)
201 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
202 self.assertIsNotNone(self.qos)
204 qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
205 self.assertIsNotNone(qos1)
206 validation_utils.objects_equivalent(self.qos, qos1)
208 qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
209 self.assertIsNotNone(qos2)
210 validation_utils.objects_equivalent(self.qos, qos2)
212 def test_create_qos_front(self):
214 Tests the cinder_utils.create_qos()
216 qos_settings = QoSSettings(name=self.qos_name, specs=self.specs,
217 consumer=Consumer.front_end)
218 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
219 self.assertIsNotNone(self.qos)
221 qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
222 self.assertIsNotNone(qos1)
223 validation_utils.objects_equivalent(self.qos, qos1)
225 qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
226 self.assertIsNotNone(qos2)
227 validation_utils.objects_equivalent(self.qos, qos2)
229 def test_create_qos_back(self):
231 Tests the cinder_utils.create_qos()
233 qos_settings = QoSSettings(name=self.qos_name, specs=self.specs,
234 consumer=Consumer.back_end)
235 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
236 self.assertIsNotNone(self.qos)
238 qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
239 self.assertIsNotNone(qos1)
240 validation_utils.objects_equivalent(self.qos, qos1)
242 qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
243 self.assertIsNotNone(qos2)
244 validation_utils.objects_equivalent(self.qos, qos2)
246 def test_create_delete_qos(self):
248 Tests the cinder_utils.create_qos()
250 qos_settings = QoSSettings(name=self.qos_name, consumer=Consumer.both)
251 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
252 self.assertIsNotNone(self.qos)
253 self.assertEqual(self.qos_name, self.qos.name)
255 qos = cinder_utils.get_qos(
256 self.cinder, qos_settings=qos_settings)
257 self.assertIsNotNone(qos)
258 validation_utils.objects_equivalent(self.qos, qos)
260 cinder_utils.delete_qos(self.cinder, self.qos)
261 self.assertIsNone(cinder_utils.get_qos(
262 self.cinder, qos_settings=qos_settings))
265 class CinderUtilsSimpleVolumeTypeTests(OSComponentTestCase):
267 Tests the creation of a Volume Type without any external settings such as
268 QoS Specs or encryption
273 Instantiates the CreateVolume object that is responsible for
274 downloading and creating an OS volume file within OpenStack
277 volume_type_name = self.__class__.__name__ + '-' + str(guid)
278 self.volume_type_settings = VolumeTypeSettings(name=volume_type_name)
279 self.volume_type = None
280 self.cinder = cinder_utils.cinder_client(self.os_creds)
284 Cleans the remote OpenStack objects
288 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
292 def test_create_simple_volume_type(self):
294 Tests the cinder_utils.create_volume_type(), get_volume_type(), and
295 get_volume_type_by_id()
297 self.volume_type = cinder_utils.create_volume_type(
298 self.cinder, self.volume_type_settings)
299 self.assertIsNotNone(self.volume_type)
300 self.assertEqual(self.volume_type_settings.name, self.volume_type.name)
302 volume_type1 = cinder_utils.get_volume_type(
303 self.cinder, volume_type_settings=self.volume_type_settings)
304 self.assertEquals(self.volume_type, volume_type1)
305 self.assertEquals(self.volume_type_settings.public,
308 volume_type2 = cinder_utils.get_volume_type_by_id(
309 self.cinder, volume_type1.id)
310 self.assertEquals(self.volume_type, volume_type2)
311 self.assertIsNone(self.volume_type.encryption)
313 def test_create_delete_volume_type(self):
315 Primarily tests the cinder_utils.delete_volume()
317 self.volume_type = cinder_utils.create_volume_type(
318 self.cinder, self.volume_type_settings)
319 self.assertIsNotNone(self.volume_type)
320 self.assertEqual(self.volume_type_settings.name, self.volume_type.name)
322 volume_type = cinder_utils.get_volume_type(
323 self.cinder, volume_type_settings=self.volume_type_settings)
324 self.assertIsNotNone(volume_type)
325 validation_utils.objects_equivalent(self.volume_type, volume_type)
326 self.assertIsNone(self.volume_type.encryption)
328 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
329 self.assertIsNone(cinder_utils.get_volume_type_by_id(
330 self.cinder, self.volume_type.id))
333 class CinderUtilsAddEncryptionTests(OSComponentTestCase):
335 Tests the creation of an encryption and association to and existing
341 Instantiates the CreateVolume object that is responsible for
342 downloading and creating an OS volume file within OpenStack
345 self.encryption_name = self.__class__.__name__ + '-' + str(guid)
346 self.encryption = None
348 self.cinder = cinder_utils.cinder_client(self.os_creds)
350 volume_type_name = self.__class__.__name__ + '-' + str(guid) + '-type'
351 self.volume_type = cinder_utils.create_volume_type(
352 self.cinder, VolumeTypeSettings(name=volume_type_name))
356 Cleans the remote OpenStack objects
360 cinder_utils.delete_volume_type_encryption(
361 self.cinder, self.volume_type)
367 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
371 def test_create_simple_encryption(self):
373 Tests the cinder_utils.create_volume_encryption(),
374 get_volume_encryption(), and get_volume_encryption_by_id()
376 encryption_settings = VolumeTypeEncryptionSettings(
377 name=self.encryption_name, provider_class='foo',
378 control_location=ControlLocation.front_end)
379 self.encryption = cinder_utils.create_volume_encryption(
380 self.cinder, self.volume_type, encryption_settings)
381 self.assertIsNotNone(self.encryption)
382 self.assertEqual('foo', self.encryption.provider)
383 self.assertEqual(ControlLocation.front_end.value,
384 self.encryption.control_location)
386 encryption1 = cinder_utils.get_volume_encryption_by_type(
387 self.cinder, self.volume_type)
388 self.assertEquals(self.encryption, encryption1)
390 def test_create_delete_encryption(self):
392 Primarily tests the cinder_utils.delete_volume_type_encryption()
394 encryption_settings = VolumeTypeEncryptionSettings(
395 name=self.encryption_name, provider_class='LuksEncryptor',
396 control_location=ControlLocation.back_end)
397 self.encryption = cinder_utils.create_volume_encryption(
398 self.cinder, self.volume_type, encryption_settings)
399 self.assertIsNotNone(self.encryption)
400 self.assertEqual('LuksEncryptor', self.encryption.provider)
401 self.assertEqual(ControlLocation.back_end.value,
402 self.encryption.control_location)
404 encryption1 = cinder_utils.get_volume_encryption_by_type(
405 self.cinder, self.volume_type)
406 self.assertEquals(self.encryption, encryption1)
408 cinder_utils.delete_volume_type_encryption(
409 self.cinder, self.volume_type)
411 encryption2 = cinder_utils.get_volume_encryption_by_type(
412 self.cinder, self.volume_type)
413 self.assertIsNone(encryption2)
415 def test_create_with_all_attrs(self):
417 Tests the cinder_utils.create_volume_encryption() with all valid
420 encryption_settings = VolumeTypeEncryptionSettings(
421 name=self.encryption_name, provider_class='foo',
422 cipher='bar', control_location=ControlLocation.back_end,
424 self.encryption = cinder_utils.create_volume_encryption(
425 self.cinder, self.volume_type, encryption_settings)
426 self.assertIsNotNone(self.encryption)
427 self.assertEqual('foo', self.encryption.provider)
428 self.assertEqual('bar', self.encryption.cipher)
429 self.assertEqual(1, self.encryption.key_size)
430 self.assertEqual(ControlLocation.back_end.value,
431 self.encryption.control_location)
433 encryption1 = cinder_utils.get_volume_encryption_by_type(
434 self.cinder, self.volume_type)
435 self.assertEquals(self.encryption, encryption1)
437 def test_create_bad_key_size(self):
439 Tests the cinder_utils.create_volume_encryption() raises an exception
440 when the provider class does not exist
442 encryption_settings = VolumeTypeEncryptionSettings(
443 name=self.encryption_name, provider_class='foo',
444 cipher='bar', control_location=ControlLocation.back_end,
447 with self.assertRaises(BadRequest):
448 self.encryption = cinder_utils.create_volume_encryption(
449 self.cinder, self.volume_type, encryption_settings)
452 class CinderUtilsVolumeTypeCompleteTests(OSComponentTestCase):
454 Tests to ensure that a volume type can have a QoS Spec added to it
459 Creates objects for testing cinder_utils.py
462 self.qos_name = self.__class__.__name__ + '-' + str(guid) + '-qos'
463 self.vol_type_name = self.__class__.__name__ + '-' + str(guid)
464 self.specs = {'foo': 'bar'}
465 self.cinder = cinder_utils.cinder_client(self.os_creds)
466 qos_settings = QoSSettings(name=self.qos_name, specs=self.specs,
467 consumer=Consumer.both)
468 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
469 self.volume_type = None
473 Cleans the remote OpenStack objects
476 if self.volume_type.encryption:
478 cinder_utils.delete_volume_type_encryption(
479 self.cinder, self.volume_type)
483 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
489 cinder_utils.delete_qos(self.cinder, self.qos)
493 def test_create_with_encryption(self):
495 Tests the cinder_utils.create_volume_type() where encryption has been
498 encryption_settings = VolumeTypeEncryptionSettings(
499 name='foo', provider_class='bar',
500 control_location=ControlLocation.back_end)
501 volume_type_settings = VolumeTypeSettings(
502 name=self.vol_type_name, encryption=encryption_settings)
503 self.volume_type = cinder_utils.create_volume_type(
504 self.cinder, volume_type_settings)
506 vol_encrypt = cinder_utils.get_volume_encryption_by_type(
507 self.cinder, self.volume_type)
508 self.assertIsNotNone(vol_encrypt)
509 self.assertIsNone(self.volume_type.qos_spec)
510 self.assertEqual(self.volume_type.encryption, vol_encrypt)
511 self.assertEqual(self.volume_type.id, vol_encrypt.volume_type_id)
513 def test_create_with_qos(self):
515 Tests the cinder_utils.create_volume_type() with an associated QoS Spec
517 volume_type_settings = VolumeTypeSettings(
518 name=self.vol_type_name, qos_spec_name=self.qos_name)
519 self.volume_type = cinder_utils.create_volume_type(
520 self.cinder, volume_type_settings)
522 self.assertIsNotNone(self.volume_type)
523 self.assertIsNone(self.volume_type.encryption)
524 self.assertIsNotNone(self.volume_type.qos_spec)
525 self.assertEqual(self.qos.id, self.volume_type.qos_spec.id)
527 def test_create_with_invalid_qos(self):
529 Tests the cinder_utils.create_volume_type() when the QoS Spec name
532 volume_type_settings = VolumeTypeSettings(
533 name=self.vol_type_name, qos_spec_name='foo')
535 self.volume_type = cinder_utils.create_volume_type(
536 self.cinder, volume_type_settings)
538 self.assertIsNone(self.volume_type.qos_spec)
540 def test_create_with_qos_and_encryption(self):
542 Tests the cinder_utils.create_volume_type() with encryption and an
545 encryption_settings = VolumeTypeEncryptionSettings(
546 name='foo', provider_class='bar',
547 control_location=ControlLocation.back_end)
548 volume_type_settings = VolumeTypeSettings(
549 name=self.vol_type_name, qos_spec_name=self.qos_name,
550 encryption=encryption_settings)
551 self.volume_type = cinder_utils.create_volume_type(
552 self.cinder, volume_type_settings)
554 self.assertIsNotNone(self.volume_type)
555 vol_encrypt = cinder_utils.get_volume_encryption_by_type(
556 self.cinder, self.volume_type)
557 self.assertIsNotNone(vol_encrypt)
558 self.assertEqual(self.volume_type.id, vol_encrypt.volume_type_id)
559 self.assertIsNotNone(self.volume_type.qos_spec)
560 self.assertEqual(self.qos.id, self.volume_type.qos_spec.id)