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
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)
81 Cleans the remote OpenStack objects
85 cinder_utils.delete_volume(self.cinder, self.volume)
89 self.assertTrue(volume_deleted(self.cinder, self.volume))
91 def test_create_simple_volume(self):
93 Tests the cinder_utils.create_volume()
95 volume_settings = VolumeConfig(name=self.volume_name)
96 self.volume = cinder_utils.create_volume(
97 self.cinder, volume_settings)
98 self.assertIsNotNone(self.volume)
99 self.assertEqual(self.volume_name, self.volume.name)
101 self.assertTrue(volume_active(self.cinder, self.volume))
103 volume = cinder_utils.get_volume(
104 self.cinder, volume_settings=volume_settings)
105 self.assertIsNotNone(volume)
106 validation_utils.objects_equivalent(self.volume, volume)
108 def test_create_delete_volume(self):
110 Tests the cinder_utils.create_volume()
112 volume_settings = VolumeConfig(name=self.volume_name)
113 self.volume = cinder_utils.create_volume(
114 self.cinder, volume_settings)
115 self.assertIsNotNone(self.volume)
116 self.assertEqual(self.volume_name, self.volume.name)
118 self.assertTrue(volume_active(self.cinder, self.volume))
120 volume = cinder_utils.get_volume(
121 self.cinder, volume_settings=volume_settings)
122 self.assertIsNotNone(volume)
123 validation_utils.objects_equivalent(self.volume, volume)
125 cinder_utils.delete_volume(self.cinder, self.volume)
126 self.assertTrue(volume_deleted(self.cinder, self.volume))
128 cinder_utils.get_volume(self.cinder, volume_settings))
131 def volume_active(cinder, volume):
133 Returns true if volume becomes active
138 end_time = time.time() + create_volume.VOLUME_ACTIVE_TIMEOUT
139 while time.time() < end_time:
140 status = cinder_utils.get_volume_status(cinder, volume)
141 if status == create_volume.STATUS_ACTIVE:
143 elif status == create_volume.STATUS_FAILED:
150 def volume_deleted(cinder, volume):
152 Returns true if volume becomes active
157 end_time = time.time() + create_volume.VOLUME_ACTIVE_TIMEOUT
158 while time.time() < end_time:
160 status = cinder_utils.get_volume_status(cinder, volume)
161 if status == create_volume.STATUS_DELETED:
171 class CinderUtilsQoSTests(OSComponentTestCase):
173 Test for the CreateQos class defined in create_qos.py
178 Creates objects for testing cinder_utils.py
181 self.qos_name = self.__class__.__name__ + '-' + str(guid)
182 self.specs = {'foo': 'bar '}
184 self.cinder = cinder_utils.cinder_client(self.os_creds)
188 Cleans the remote OpenStack objects
192 cinder_utils.delete_qos(self.cinder, self.qos)
196 def test_create_qos_both(self):
198 Tests the cinder_utils.create_qos()
200 qos_settings = QoSConfig(
201 name=self.qos_name, specs=self.specs, consumer=Consumer.both)
202 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
203 self.assertIsNotNone(self.qos)
205 qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
206 self.assertIsNotNone(qos1)
207 validation_utils.objects_equivalent(self.qos, qos1)
209 qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
210 self.assertIsNotNone(qos2)
211 validation_utils.objects_equivalent(self.qos, qos2)
213 def test_create_qos_front(self):
215 Tests the cinder_utils.create_qos()
217 qos_settings = QoSConfig(
218 name=self.qos_name, specs=self.specs, consumer=Consumer.front_end)
219 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
220 self.assertIsNotNone(self.qos)
222 qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
223 self.assertIsNotNone(qos1)
224 validation_utils.objects_equivalent(self.qos, qos1)
226 qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
227 self.assertIsNotNone(qos2)
228 validation_utils.objects_equivalent(self.qos, qos2)
230 def test_create_qos_back(self):
232 Tests the cinder_utils.create_qos()
234 qos_settings = QoSConfig(
235 name=self.qos_name, specs=self.specs, consumer=Consumer.back_end)
236 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
237 self.assertIsNotNone(self.qos)
239 qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
240 self.assertIsNotNone(qos1)
241 validation_utils.objects_equivalent(self.qos, qos1)
243 qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
244 self.assertIsNotNone(qos2)
245 validation_utils.objects_equivalent(self.qos, qos2)
247 def test_create_delete_qos(self):
249 Tests the cinder_utils.create_qos()
251 qos_settings = QoSConfig(name=self.qos_name, consumer=Consumer.both)
252 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
253 self.assertIsNotNone(self.qos)
254 self.assertEqual(self.qos_name, self.qos.name)
256 qos = cinder_utils.get_qos(
257 self.cinder, qos_settings=qos_settings)
258 self.assertIsNotNone(qos)
259 validation_utils.objects_equivalent(self.qos, qos)
261 cinder_utils.delete_qos(self.cinder, self.qos)
262 self.assertIsNone(cinder_utils.get_qos(
263 self.cinder, qos_settings=qos_settings))
266 class CinderUtilsSimpleVolumeTypeTests(OSComponentTestCase):
268 Tests the creation of a Volume Type without any external settings such as
269 QoS Specs or encryption
274 Instantiates the CreateVolume object that is responsible for
275 downloading and creating an OS volume file within OpenStack
278 volume_type_name = self.__class__.__name__ + '-' + str(guid)
279 self.volume_type_settings = VolumeTypeConfig(name=volume_type_name)
280 self.volume_type = None
281 self.cinder = cinder_utils.cinder_client(self.os_creds)
285 Cleans the remote OpenStack objects
289 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
293 def test_create_simple_volume_type(self):
295 Tests the cinder_utils.create_volume_type(), get_volume_type(), and
296 get_volume_type_by_id()
298 self.volume_type = cinder_utils.create_volume_type(
299 self.cinder, self.volume_type_settings)
300 self.assertIsNotNone(self.volume_type)
301 self.assertEqual(self.volume_type_settings.name, self.volume_type.name)
303 volume_type1 = cinder_utils.get_volume_type(
304 self.cinder, volume_type_settings=self.volume_type_settings)
305 self.assertEquals(self.volume_type, volume_type1)
306 self.assertEquals(self.volume_type_settings.public,
309 volume_type2 = cinder_utils.get_volume_type_by_id(
310 self.cinder, volume_type1.id)
311 self.assertEquals(self.volume_type, volume_type2)
312 self.assertIsNone(self.volume_type.encryption)
314 def test_create_delete_volume_type(self):
316 Primarily tests the cinder_utils.delete_volume()
318 self.volume_type = cinder_utils.create_volume_type(
319 self.cinder, self.volume_type_settings)
320 self.assertIsNotNone(self.volume_type)
321 self.assertEqual(self.volume_type_settings.name, self.volume_type.name)
323 volume_type = cinder_utils.get_volume_type(
324 self.cinder, volume_type_settings=self.volume_type_settings)
325 self.assertIsNotNone(volume_type)
326 validation_utils.objects_equivalent(self.volume_type, volume_type)
327 self.assertIsNone(self.volume_type.encryption)
329 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
330 self.assertIsNone(cinder_utils.get_volume_type_by_id(
331 self.cinder, self.volume_type.id))
334 class CinderUtilsAddEncryptionTests(OSComponentTestCase):
336 Tests the creation of an encryption and association to and existing
342 Instantiates the CreateVolume object that is responsible for
343 downloading and creating an OS volume file within OpenStack
346 self.encryption_name = self.__class__.__name__ + '-' + str(guid)
347 self.encryption = None
349 self.cinder = cinder_utils.cinder_client(self.os_creds)
351 volume_type_name = self.__class__.__name__ + '-' + str(guid) + '-type'
352 self.volume_type = cinder_utils.create_volume_type(
353 self.cinder, VolumeTypeConfig(name=volume_type_name))
357 Cleans the remote OpenStack objects
361 cinder_utils.delete_volume_type_encryption(
362 self.cinder, self.volume_type)
368 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
372 def test_create_simple_encryption(self):
374 Tests the cinder_utils.create_volume_encryption(),
375 get_volume_encryption(), and get_volume_encryption_by_id()
377 encryption_settings = VolumeTypeEncryptionConfig(
378 name=self.encryption_name, provider_class='foo',
379 control_location=ControlLocation.front_end)
380 self.encryption = cinder_utils.create_volume_encryption(
381 self.cinder, self.volume_type, encryption_settings)
382 self.assertIsNotNone(self.encryption)
383 self.assertEqual('foo', self.encryption.provider)
384 self.assertEqual(ControlLocation.front_end.value,
385 self.encryption.control_location)
387 encryption1 = cinder_utils.get_volume_encryption_by_type(
388 self.cinder, self.volume_type)
389 self.assertEquals(self.encryption, encryption1)
391 def test_create_delete_encryption(self):
393 Primarily tests the cinder_utils.delete_volume_type_encryption()
395 encryption_settings = VolumeTypeEncryptionConfig(
396 name=self.encryption_name, provider_class='LuksEncryptor',
397 control_location=ControlLocation.back_end)
398 self.encryption = cinder_utils.create_volume_encryption(
399 self.cinder, self.volume_type, encryption_settings)
400 self.assertIsNotNone(self.encryption)
401 self.assertEqual('LuksEncryptor', self.encryption.provider)
402 self.assertEqual(ControlLocation.back_end.value,
403 self.encryption.control_location)
405 encryption1 = cinder_utils.get_volume_encryption_by_type(
406 self.cinder, self.volume_type)
407 self.assertEquals(self.encryption, encryption1)
409 cinder_utils.delete_volume_type_encryption(
410 self.cinder, self.volume_type)
412 encryption2 = cinder_utils.get_volume_encryption_by_type(
413 self.cinder, self.volume_type)
414 self.assertIsNone(encryption2)
416 def test_create_with_all_attrs(self):
418 Tests the cinder_utils.create_volume_encryption() with all valid
421 encryption_settings = VolumeTypeEncryptionConfig(
422 name=self.encryption_name, provider_class='foo',
423 cipher='bar', control_location=ControlLocation.back_end,
425 self.encryption = cinder_utils.create_volume_encryption(
426 self.cinder, self.volume_type, encryption_settings)
427 self.assertIsNotNone(self.encryption)
428 self.assertEqual('foo', self.encryption.provider)
429 self.assertEqual('bar', self.encryption.cipher)
430 self.assertEqual(1, self.encryption.key_size)
431 self.assertEqual(ControlLocation.back_end.value,
432 self.encryption.control_location)
434 encryption1 = cinder_utils.get_volume_encryption_by_type(
435 self.cinder, self.volume_type)
436 self.assertEquals(self.encryption, encryption1)
438 def test_create_bad_key_size(self):
440 Tests the cinder_utils.create_volume_encryption() raises an exception
441 when the provider class does not exist
443 encryption_settings = VolumeTypeEncryptionConfig(
444 name=self.encryption_name, provider_class='foo',
445 cipher='bar', control_location=ControlLocation.back_end,
448 with self.assertRaises(BadRequest):
449 self.encryption = cinder_utils.create_volume_encryption(
450 self.cinder, self.volume_type, encryption_settings)
453 class CinderUtilsVolumeTypeCompleteTests(OSComponentTestCase):
455 Tests to ensure that a volume type can have a QoS Spec added to it
460 Creates objects for testing cinder_utils.py
463 self.qos_name = self.__class__.__name__ + '-' + str(guid) + '-qos'
464 self.vol_type_name = self.__class__.__name__ + '-' + str(guid)
465 self.specs = {'foo': 'bar'}
466 self.cinder = cinder_utils.cinder_client(self.os_creds)
467 qos_settings = QoSConfig(
468 name=self.qos_name, specs=self.specs, consumer=Consumer.both)
469 self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
470 self.volume_type = None
474 Cleans the remote OpenStack objects
477 if self.volume_type.encryption:
479 cinder_utils.delete_volume_type_encryption(
480 self.cinder, self.volume_type)
484 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
490 cinder_utils.delete_qos(self.cinder, self.qos)
494 def test_create_with_encryption(self):
496 Tests the cinder_utils.create_volume_type() where encryption has been
499 encryption_settings = VolumeTypeEncryptionConfig(
500 name='foo', provider_class='bar',
501 control_location=ControlLocation.back_end)
502 volume_type_settings = VolumeTypeConfig(
503 name=self.vol_type_name, encryption=encryption_settings)
504 self.volume_type = cinder_utils.create_volume_type(
505 self.cinder, volume_type_settings)
507 vol_encrypt = cinder_utils.get_volume_encryption_by_type(
508 self.cinder, self.volume_type)
509 self.assertIsNotNone(vol_encrypt)
510 self.assertIsNone(self.volume_type.qos_spec)
511 self.assertEqual(self.volume_type.encryption, vol_encrypt)
512 self.assertEqual(self.volume_type.id, vol_encrypt.volume_type_id)
514 def test_create_with_qos(self):
516 Tests the cinder_utils.create_volume_type() with an associated QoS Spec
518 volume_type_settings = VolumeTypeConfig(
519 name=self.vol_type_name, qos_spec_name=self.qos_name)
520 self.volume_type = cinder_utils.create_volume_type(
521 self.cinder, volume_type_settings)
523 self.assertIsNotNone(self.volume_type)
524 self.assertIsNone(self.volume_type.encryption)
525 self.assertIsNotNone(self.volume_type.qos_spec)
526 self.assertEqual(self.qos.id, self.volume_type.qos_spec.id)
528 def test_create_with_invalid_qos(self):
530 Tests the cinder_utils.create_volume_type() when the QoS Spec name
533 volume_type_settings = VolumeTypeConfig(
534 name=self.vol_type_name, qos_spec_name='foo')
536 self.volume_type = cinder_utils.create_volume_type(
537 self.cinder, volume_type_settings)
539 self.assertIsNone(self.volume_type.qos_spec)
541 def test_create_with_qos_and_encryption(self):
543 Tests the cinder_utils.create_volume_type() with encryption and an
546 encryption_settings = VolumeTypeEncryptionConfig(
547 name='foo', provider_class='bar',
548 control_location=ControlLocation.back_end)
549 volume_type_settings = VolumeTypeConfig(
550 name=self.vol_type_name, qos_spec_name=self.qos_name,
551 encryption=encryption_settings)
552 self.volume_type = cinder_utils.create_volume_type(
553 self.cinder, volume_type_settings)
555 self.assertIsNotNone(self.volume_type)
556 vol_encrypt = cinder_utils.get_volume_encryption_by_type(
557 self.cinder, self.volume_type)
558 self.assertIsNotNone(vol_encrypt)
559 self.assertEqual(self.volume_type.id, vol_encrypt.volume_type_id)
560 self.assertIsNotNone(self.volume_type.qos_spec)
561 self.assertEqual(self.qos.id, self.volume_type.qos_spec.id)