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.qos import Consumer, QoSConfig
  22 from snaps.openstack import create_volume
  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
  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 = VolumeTypeSettings(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, VolumeTypeSettings(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 = 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)
 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 = 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)
 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 = VolumeTypeEncryptionSettings(
 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 = VolumeTypeEncryptionSettings(
 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 = 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)
 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 = 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)
 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 = VolumeTypeSettings(
 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 = 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)
 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)