snaps/openstack/utils/tests/cinder_utils_tests.py

   1 # Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
   2 #                    and others.  All rights reserved.
   3 #
   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:
   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,
  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.
  15 import logging
  16 import uuid
  17
  18 import time
  19 from cinderclient.exceptions import NotFound, BadRequest
  20
  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
  30
  31 __author__ = 'spisarski'
  32
  33
  34 logger = logging.getLogger('cinder_utils_tests')
  35
  36
  37 class CinderSmokeTests(OSComponentTestCase):
  38     """
  39     Tests to ensure that the neutron client can communicate with the cloud
  40     """
  41
  42     def test_cinder_connect_success(self):
  43         """
  44         Tests to ensure that the proper credentials can connect.
  45         """
  46         cinder = cinder_utils.cinder_client(self.os_creds)
  47         volumes = cinder.volumes.list()
  48         self.assertIsNotNone(volumes)
  49         self.assertTrue(isinstance(volumes, list))
  50
  51     def test_cinder_connect_fail(self):
  52         """
  53         Tests to ensure that the improper credentials cannot connect.
  54         """
  55         from snaps.openstack.os_credentials import OSCreds
  56
  57         with self.assertRaises(Exception):
  58             cinder = cinder_utils.cinder_client(OSCreds(
  59                 username='user', password='pass', auth_url='url',
  60                 project_name='project'))
  61             cinder.volumes.list()
  62
  63
  64 class CinderUtilsVolumeTests(OSComponentTestCase):
  65     """
  66     Test for the CreateVolume class defined in create_volume.py
  67     """
  68
  69     def setUp(self):
  70         """
  71         Instantiates the CreateVolume object that is responsible for
  72         downloading and creating an OS volume file within OpenStack
  73         """
  74         guid = uuid.uuid4()
  75         self.volume_name = self.__class__.__name__ + '-' + str(guid)
  76         self.volume = None
  77         self.cinder = cinder_utils.cinder_client(self.os_creds)
  78
  79     def tearDown(self):
  80         """
  81         Cleans the remote OpenStack objects
  82         """
  83         if self.volume:
  84             try:
  85                 cinder_utils.delete_volume(self.cinder, self.volume)
  86             except NotFound:
  87                 pass
  88
  89         self.assertTrue(volume_deleted(self.cinder, self.volume))
  90
  91     def test_create_simple_volume(self):
  92         """
  93         Tests the cinder_utils.create_volume()
  94         """
  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)
 100
 101         self.assertTrue(volume_active(self.cinder, self.volume))
 102
 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)
 107
 108     def test_create_delete_volume(self):
 109         """
 110         Tests the cinder_utils.create_volume()
 111         """
 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)
 117
 118         self.assertTrue(volume_active(self.cinder, self.volume))
 119
 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)
 124
 125         cinder_utils.delete_volume(self.cinder, self.volume)
 126         self.assertTrue(volume_deleted(self.cinder, self.volume))
 127         self.assertIsNone(
 128             cinder_utils.get_volume(self.cinder, volume_settings))
 129
 130
 131 def volume_active(cinder, volume):
 132     """
 133     Returns true if volume becomes active
 134     :param cinder:
 135     :param volume:
 136     :return:
 137     """
 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:
 142             return True
 143         elif status == create_volume.STATUS_FAILED:
 144             return False
 145         time.sleep(3)
 146
 147     return False
 148
 149
 150 def volume_deleted(cinder, volume):
 151     """
 152     Returns true if volume becomes active
 153     :param cinder:
 154     :param volume:
 155     :return:
 156     """
 157     end_time = time.time() + create_volume.VOLUME_ACTIVE_TIMEOUT
 158     while time.time() < end_time:
 159         try:
 160             status = cinder_utils.get_volume_status(cinder, volume)
 161             if status == create_volume.STATUS_DELETED:
 162                 return True
 163         except NotFound:
 164             return True
 165
 166         time.sleep(3)
 167
 168     return False
 169
 170
 171 class CinderUtilsQoSTests(OSComponentTestCase):
 172     """
 173     Test for the CreateQos class defined in create_qos.py
 174     """
 175
 176     def setUp(self):
 177         """
 178         Creates objects for testing cinder_utils.py
 179         """
 180         guid = uuid.uuid4()
 181         self.qos_name = self.__class__.__name__ + '-' + str(guid)
 182         self.specs = {'foo': 'bar '}
 183         self.qos = None
 184         self.cinder = cinder_utils.cinder_client(self.os_creds)
 185
 186     def tearDown(self):
 187         """
 188         Cleans the remote OpenStack objects
 189         """
 190         if self.qos:
 191             try:
 192                 cinder_utils.delete_qos(self.cinder, self.qos)
 193             except NotFound:
 194                 pass
 195
 196     def test_create_qos_both(self):
 197         """
 198         Tests the cinder_utils.create_qos()
 199         """
 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)
 204
 205         qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
 206         self.assertIsNotNone(qos1)
 207         validation_utils.objects_equivalent(self.qos, qos1)
 208
 209         qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
 210         self.assertIsNotNone(qos2)
 211         validation_utils.objects_equivalent(self.qos, qos2)
 212
 213     def test_create_qos_front(self):
 214         """
 215         Tests the cinder_utils.create_qos()
 216         """
 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)
 221
 222         qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
 223         self.assertIsNotNone(qos1)
 224         validation_utils.objects_equivalent(self.qos, qos1)
 225
 226         qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
 227         self.assertIsNotNone(qos2)
 228         validation_utils.objects_equivalent(self.qos, qos2)
 229
 230     def test_create_qos_back(self):
 231         """
 232         Tests the cinder_utils.create_qos()
 233         """
 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)
 238
 239         qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
 240         self.assertIsNotNone(qos1)
 241         validation_utils.objects_equivalent(self.qos, qos1)
 242
 243         qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
 244         self.assertIsNotNone(qos2)
 245         validation_utils.objects_equivalent(self.qos, qos2)
 246
 247     def test_create_delete_qos(self):
 248         """
 249         Tests the cinder_utils.create_qos()
 250         """
 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)
 255
 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)
 260
 261         cinder_utils.delete_qos(self.cinder, self.qos)
 262         self.assertIsNone(cinder_utils.get_qos(
 263             self.cinder, qos_settings=qos_settings))
 264
 265
 266 class CinderUtilsSimpleVolumeTypeTests(OSComponentTestCase):
 267     """
 268     Tests the creation of a Volume Type without any external settings such as
 269     QoS Specs or encryption
 270     """
 271
 272     def setUp(self):
 273         """
 274         Instantiates the CreateVolume object that is responsible for
 275         downloading and creating an OS volume file within OpenStack
 276         """
 277         guid = uuid.uuid4()
 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)
 282
 283     def tearDown(self):
 284         """
 285         Cleans the remote OpenStack objects
 286         """
 287         if self.volume_type:
 288             try:
 289                 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
 290             except NotFound:
 291                 pass
 292
 293     def test_create_simple_volume_type(self):
 294         """
 295         Tests the cinder_utils.create_volume_type(), get_volume_type(), and
 296         get_volume_type_by_id()
 297         """
 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)
 302
 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,
 307                           volume_type1.public)
 308
 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)
 313
 314     def test_create_delete_volume_type(self):
 315         """
 316         Primarily tests the cinder_utils.delete_volume()
 317         """
 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)
 322
 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)
 328
 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))
 332
 333
 334 class CinderUtilsAddEncryptionTests(OSComponentTestCase):
 335     """
 336     Tests the creation of an encryption and association to and existing
 337     VolumeType object
 338     """
 339
 340     def setUp(self):
 341         """
 342         Instantiates the CreateVolume object that is responsible for
 343         downloading and creating an OS volume file within OpenStack
 344         """
 345         guid = uuid.uuid4()
 346         self.encryption_name = self.__class__.__name__ + '-' + str(guid)
 347         self.encryption = None
 348
 349         self.cinder = cinder_utils.cinder_client(self.os_creds)
 350
 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))
 354
 355     def tearDown(self):
 356         """
 357         Cleans the remote OpenStack objects
 358         """
 359         if self.encryption:
 360             try:
 361                 cinder_utils.delete_volume_type_encryption(
 362                     self.cinder, self.volume_type)
 363             except NotFound:
 364                 pass
 365
 366         if self.volume_type:
 367             try:
 368                 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
 369             except NotFound:
 370                 pass
 371
 372     def test_create_simple_encryption(self):
 373         """
 374         Tests the cinder_utils.create_volume_encryption(),
 375         get_volume_encryption(), and get_volume_encryption_by_id()
 376         """
 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)
 386
 387         encryption1 = cinder_utils.get_volume_encryption_by_type(
 388             self.cinder, self.volume_type)
 389         self.assertEquals(self.encryption, encryption1)
 390
 391     def test_create_delete_encryption(self):
 392         """
 393         Primarily tests the cinder_utils.delete_volume_type_encryption()
 394         """
 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)
 404
 405         encryption1 = cinder_utils.get_volume_encryption_by_type(
 406             self.cinder, self.volume_type)
 407         self.assertEquals(self.encryption, encryption1)
 408
 409         cinder_utils.delete_volume_type_encryption(
 410             self.cinder, self.volume_type)
 411
 412         encryption2 = cinder_utils.get_volume_encryption_by_type(
 413             self.cinder, self.volume_type)
 414         self.assertIsNone(encryption2)
 415
 416     def test_create_with_all_attrs(self):
 417         """
 418         Tests the cinder_utils.create_volume_encryption() with all valid
 419         settings
 420         """
 421         encryption_settings = VolumeTypeEncryptionConfig(
 422             name=self.encryption_name, provider_class='foo',
 423             cipher='bar', control_location=ControlLocation.back_end,
 424             key_size=1)
 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)
 433
 434         encryption1 = cinder_utils.get_volume_encryption_by_type(
 435             self.cinder, self.volume_type)
 436         self.assertEquals(self.encryption, encryption1)
 437
 438     def test_create_bad_key_size(self):
 439         """
 440         Tests the cinder_utils.create_volume_encryption() raises an exception
 441         when the provider class does not exist
 442         """
 443         encryption_settings = VolumeTypeEncryptionConfig(
 444             name=self.encryption_name, provider_class='foo',
 445             cipher='bar', control_location=ControlLocation.back_end,
 446             key_size=-1)
 447
 448         with self.assertRaises(BadRequest):
 449             self.encryption = cinder_utils.create_volume_encryption(
 450                 self.cinder, self.volume_type, encryption_settings)
 451
 452
 453 class CinderUtilsVolumeTypeCompleteTests(OSComponentTestCase):
 454     """
 455     Tests to ensure that a volume type can have a QoS Spec added to it
 456     """
 457
 458     def setUp(self):
 459         """
 460         Creates objects for testing cinder_utils.py
 461         """
 462         guid = uuid.uuid4()
 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
 471
 472     def tearDown(self):
 473         """
 474         Cleans the remote OpenStack objects
 475         """
 476         if self.volume_type:
 477             if self.volume_type.encryption:
 478                 try:
 479                     cinder_utils.delete_volume_type_encryption(
 480                         self.cinder, self.volume_type)
 481                 except NotFound:
 482                     pass
 483             try:
 484                 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
 485             except NotFound:
 486                 pass
 487
 488         if self.qos:
 489             try:
 490                 cinder_utils.delete_qos(self.cinder, self.qos)
 491             except NotFound:
 492                 pass
 493
 494     def test_create_with_encryption(self):
 495         """
 496         Tests the cinder_utils.create_volume_type() where encryption has been
 497         configured
 498         """
 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)
 506
 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)
 513
 514     def test_create_with_qos(self):
 515         """
 516         Tests the cinder_utils.create_volume_type() with an associated QoS Spec
 517         """
 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)
 522
 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)
 527
 528     def test_create_with_invalid_qos(self):
 529         """
 530         Tests the cinder_utils.create_volume_type() when the QoS Spec name
 531         does not exist
 532         """
 533         volume_type_settings = VolumeTypeConfig(
 534             name=self.vol_type_name, qos_spec_name='foo')
 535
 536         self.volume_type = cinder_utils.create_volume_type(
 537             self.cinder, volume_type_settings)
 538
 539         self.assertIsNone(self.volume_type.qos_spec)
 540
 541     def test_create_with_qos_and_encryption(self):
 542         """
 543         Tests the cinder_utils.create_volume_type() with encryption and an
 544         associated QoS Spec
 545         """
 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)
 554
 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)