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