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, keystone_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         self.keystone = keystone_utils.keystone_client(self.os_creds)
  79
  80     def tearDown(self):
  81         """
  82         Cleans the remote OpenStack objects
  83         """
  84         if self.volume:
  85             try:
  86                 cinder_utils.delete_volume(self.cinder, self.volume)
  87             except NotFound:
  88                 pass
  89
  90         self.assertTrue(volume_deleted(self.cinder, self.volume))
  91
  92     def test_create_simple_volume(self):
  93         """
  94         Tests the cinder_utils.create_volume()
  95         """
  96         volume_settings = VolumeConfig(name=self.volume_name)
  97         self.volume = cinder_utils.create_volume(
  98             self.cinder, self.keystone, volume_settings)
  99         self.assertIsNotNone(self.volume)
 100         self.assertEqual(self.volume_name, self.volume.name)
 101
 102         self.assertTrue(volume_active(self.cinder, self.volume))
 103
 104         volume = cinder_utils.get_volume(
 105             self.cinder, self.keystone, volume_settings=volume_settings,
 106             project_name=self.os_creds.project_name)
 107         self.assertIsNotNone(volume)
 108         validation_utils.objects_equivalent(self.volume, volume)
 109
 110     def test_create_delete_volume(self):
 111         """
 112         Tests the cinder_utils.create_volume()
 113         """
 114         volume_settings = VolumeConfig(name=self.volume_name)
 115         self.volume = cinder_utils.create_volume(
 116             self.cinder, self.keystone, volume_settings)
 117         self.assertIsNotNone(self.volume)
 118         self.assertEqual(self.volume_name, self.volume.name)
 119
 120         self.assertTrue(volume_active(self.cinder, self.volume))
 121
 122         volume = cinder_utils.get_volume(
 123             self.cinder, self.keystone, volume_settings=volume_settings,
 124             project_name=self.os_creds.project_name)
 125         self.assertIsNotNone(volume)
 126         validation_utils.objects_equivalent(self.volume, volume)
 127
 128         cinder_utils.delete_volume(self.cinder, self.volume)
 129         self.assertTrue(volume_deleted(self.cinder, self.volume))
 130         self.assertIsNone(
 131             cinder_utils.get_volume(
 132                 self.cinder, self.keystone, volume_settings,
 133                 project_name=self.os_creds.project_name))
 134
 135
 136 def volume_active(cinder, volume):
 137     """
 138     Returns true if volume becomes active
 139     :param cinder:
 140     :param volume:
 141     :return:
 142     """
 143     end_time = time.time() + create_volume.VOLUME_ACTIVE_TIMEOUT
 144     while time.time() < end_time:
 145         status = cinder_utils.get_volume_status(cinder, volume)
 146         if status == create_volume.STATUS_ACTIVE:
 147             return True
 148         elif status == create_volume.STATUS_FAILED:
 149             return False
 150         time.sleep(3)
 151
 152     return False
 153
 154
 155 def volume_deleted(cinder, volume):
 156     """
 157     Returns true if volume becomes active
 158     :param cinder:
 159     :param volume:
 160     :return:
 161     """
 162     end_time = time.time() + create_volume.VOLUME_ACTIVE_TIMEOUT
 163     while time.time() < end_time:
 164         try:
 165             status = cinder_utils.get_volume_status(cinder, volume)
 166             if status == create_volume.STATUS_DELETED:
 167                 return True
 168         except NotFound:
 169             return True
 170
 171         time.sleep(3)
 172
 173     return False
 174
 175
 176 class CinderUtilsQoSTests(OSComponentTestCase):
 177     """
 178     Test for the CreateQos class defined in create_qos.py
 179     """
 180
 181     def setUp(self):
 182         """
 183         Creates objects for testing cinder_utils.py
 184         """
 185         guid = uuid.uuid4()
 186         self.qos_name = self.__class__.__name__ + '-' + str(guid)
 187         self.specs = {'foo': 'bar '}
 188         self.qos = None
 189         self.cinder = cinder_utils.cinder_client(self.os_creds)
 190
 191     def tearDown(self):
 192         """
 193         Cleans the remote OpenStack objects
 194         """
 195         if self.qos:
 196             try:
 197                 cinder_utils.delete_qos(self.cinder, self.qos)
 198             except NotFound:
 199                 pass
 200
 201     def test_create_qos_both(self):
 202         """
 203         Tests the cinder_utils.create_qos()
 204         """
 205         qos_settings = QoSConfig(
 206             name=self.qos_name, specs=self.specs, consumer=Consumer.both)
 207         self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
 208         self.assertIsNotNone(self.qos)
 209
 210         qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
 211         self.assertIsNotNone(qos1)
 212         validation_utils.objects_equivalent(self.qos, qos1)
 213
 214         qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
 215         self.assertIsNotNone(qos2)
 216         validation_utils.objects_equivalent(self.qos, qos2)
 217
 218     def test_create_qos_front(self):
 219         """
 220         Tests the cinder_utils.create_qos()
 221         """
 222         qos_settings = QoSConfig(
 223             name=self.qos_name, specs=self.specs, consumer=Consumer.front_end)
 224         self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
 225         self.assertIsNotNone(self.qos)
 226
 227         qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
 228         self.assertIsNotNone(qos1)
 229         validation_utils.objects_equivalent(self.qos, qos1)
 230
 231         qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
 232         self.assertIsNotNone(qos2)
 233         validation_utils.objects_equivalent(self.qos, qos2)
 234
 235     def test_create_qos_back(self):
 236         """
 237         Tests the cinder_utils.create_qos()
 238         """
 239         qos_settings = QoSConfig(
 240             name=self.qos_name, specs=self.specs, consumer=Consumer.back_end)
 241         self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
 242         self.assertIsNotNone(self.qos)
 243
 244         qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
 245         self.assertIsNotNone(qos1)
 246         validation_utils.objects_equivalent(self.qos, qos1)
 247
 248         qos2 = cinder_utils.get_qos_by_id(self.cinder, qos1.id)
 249         self.assertIsNotNone(qos2)
 250         validation_utils.objects_equivalent(self.qos, qos2)
 251
 252     def test_create_delete_qos(self):
 253         """
 254         Tests the cinder_utils.create_qos()
 255         """
 256         qos_settings = QoSConfig(name=self.qos_name, consumer=Consumer.both)
 257         self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
 258         self.assertIsNotNone(self.qos)
 259         self.assertEqual(self.qos_name, self.qos.name)
 260
 261         qos = cinder_utils.get_qos(
 262             self.cinder, qos_settings=qos_settings)
 263         self.assertIsNotNone(qos)
 264         validation_utils.objects_equivalent(self.qos, qos)
 265
 266         cinder_utils.delete_qos(self.cinder, self.qos)
 267         self.assertIsNone(cinder_utils.get_qos(
 268             self.cinder, qos_settings=qos_settings))
 269
 270
 271 class CinderUtilsSimpleVolumeTypeTests(OSComponentTestCase):
 272     """
 273     Tests the creation of a Volume Type without any external settings such as
 274     QoS Specs or encryption
 275     """
 276
 277     def setUp(self):
 278         """
 279         Instantiates the CreateVolume object that is responsible for
 280         downloading and creating an OS volume file within OpenStack
 281         """
 282         guid = uuid.uuid4()
 283         volume_type_name = self.__class__.__name__ + '-' + str(guid)
 284         self.volume_type_settings = VolumeTypeConfig(name=volume_type_name)
 285         self.volume_type = None
 286         self.cinder = cinder_utils.cinder_client(self.os_creds)
 287
 288     def tearDown(self):
 289         """
 290         Cleans the remote OpenStack objects
 291         """
 292         if self.volume_type:
 293             try:
 294                 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
 295             except NotFound:
 296                 pass
 297
 298     def test_create_simple_volume_type(self):
 299         """
 300         Tests the cinder_utils.create_volume_type(), get_volume_type(), and
 301         get_volume_type_by_id()
 302         """
 303         self.volume_type = cinder_utils.create_volume_type(
 304             self.cinder, self.volume_type_settings)
 305         self.assertIsNotNone(self.volume_type)
 306         self.assertEqual(self.volume_type_settings.name, self.volume_type.name)
 307
 308         volume_type1 = cinder_utils.get_volume_type(
 309             self.cinder, volume_type_settings=self.volume_type_settings)
 310         self.assertEquals(self.volume_type, volume_type1)
 311         self.assertEquals(self.volume_type_settings.public,
 312                           volume_type1.public)
 313
 314         volume_type2 = cinder_utils.get_volume_type_by_id(
 315             self.cinder, volume_type1.id)
 316         self.assertEquals(self.volume_type, volume_type2)
 317         self.assertIsNone(self.volume_type.encryption)
 318
 319     def test_create_delete_volume_type(self):
 320         """
 321         Primarily tests the cinder_utils.delete_volume()
 322         """
 323         self.volume_type = cinder_utils.create_volume_type(
 324             self.cinder, self.volume_type_settings)
 325         self.assertIsNotNone(self.volume_type)
 326         self.assertEqual(self.volume_type_settings.name, self.volume_type.name)
 327
 328         volume_type = cinder_utils.get_volume_type(
 329             self.cinder, volume_type_settings=self.volume_type_settings)
 330         self.assertIsNotNone(volume_type)
 331         validation_utils.objects_equivalent(self.volume_type, volume_type)
 332         self.assertIsNone(self.volume_type.encryption)
 333
 334         cinder_utils.delete_volume_type(self.cinder, self.volume_type)
 335         self.assertIsNone(cinder_utils.get_volume_type_by_id(
 336             self.cinder, self.volume_type.id))
 337
 338
 339 class CinderUtilsAddEncryptionTests(OSComponentTestCase):
 340     """
 341     Tests the creation of an encryption and association to and existing
 342     VolumeType object
 343     """
 344
 345     def setUp(self):
 346         """
 347         Instantiates the CreateVolume object that is responsible for
 348         downloading and creating an OS volume file within OpenStack
 349         """
 350         guid = uuid.uuid4()
 351         self.encryption_name = self.__class__.__name__ + '-' + str(guid)
 352         self.encryption = None
 353
 354         self.cinder = cinder_utils.cinder_client(self.os_creds)
 355
 356         volume_type_name = self.__class__.__name__ + '-' + str(guid) + '-type'
 357         self.volume_type = cinder_utils.create_volume_type(
 358             self.cinder, VolumeTypeConfig(name=volume_type_name))
 359
 360     def tearDown(self):
 361         """
 362         Cleans the remote OpenStack objects
 363         """
 364         if self.encryption:
 365             try:
 366                 cinder_utils.delete_volume_type_encryption(
 367                     self.cinder, self.volume_type)
 368             except NotFound:
 369                 pass
 370
 371         if self.volume_type:
 372             try:
 373                 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
 374             except NotFound:
 375                 pass
 376
 377     def test_create_simple_encryption(self):
 378         """
 379         Tests the cinder_utils.create_volume_encryption(),
 380         get_volume_encryption(), and get_volume_encryption_by_id()
 381         """
 382         encryption_settings = VolumeTypeEncryptionConfig(
 383             name=self.encryption_name, provider_class='foo',
 384             control_location=ControlLocation.front_end)
 385         self.encryption = cinder_utils.create_volume_encryption(
 386             self.cinder, self.volume_type, encryption_settings)
 387         self.assertIsNotNone(self.encryption)
 388         self.assertEqual('foo', self.encryption.provider)
 389         self.assertEqual(ControlLocation.front_end.value,
 390                          self.encryption.control_location)
 391
 392         encryption1 = cinder_utils.get_volume_encryption_by_type(
 393             self.cinder, self.volume_type)
 394         self.assertEquals(self.encryption, encryption1)
 395
 396     def test_create_delete_encryption(self):
 397         """
 398         Primarily tests the cinder_utils.delete_volume_type_encryption()
 399         """
 400         encryption_settings = VolumeTypeEncryptionConfig(
 401             name=self.encryption_name, provider_class='LuksEncryptor',
 402             control_location=ControlLocation.back_end)
 403         self.encryption = cinder_utils.create_volume_encryption(
 404             self.cinder, self.volume_type, encryption_settings)
 405         self.assertIsNotNone(self.encryption)
 406         self.assertEqual('LuksEncryptor', self.encryption.provider)
 407         self.assertEqual(ControlLocation.back_end.value,
 408                          self.encryption.control_location)
 409
 410         encryption1 = cinder_utils.get_volume_encryption_by_type(
 411             self.cinder, self.volume_type)
 412         self.assertEquals(self.encryption, encryption1)
 413
 414         cinder_utils.delete_volume_type_encryption(
 415             self.cinder, self.volume_type)
 416
 417         encryption2 = cinder_utils.get_volume_encryption_by_type(
 418             self.cinder, self.volume_type)
 419         self.assertIsNone(encryption2)
 420
 421     def test_create_with_all_attrs(self):
 422         """
 423         Tests the cinder_utils.create_volume_encryption() with all valid
 424         settings
 425         """
 426         encryption_settings = VolumeTypeEncryptionConfig(
 427             name=self.encryption_name, provider_class='foo',
 428             cipher='bar', control_location=ControlLocation.back_end,
 429             key_size=1)
 430         self.encryption = cinder_utils.create_volume_encryption(
 431             self.cinder, self.volume_type, encryption_settings)
 432         self.assertIsNotNone(self.encryption)
 433         self.assertEqual('foo', self.encryption.provider)
 434         self.assertEqual('bar', self.encryption.cipher)
 435         self.assertEqual(1, self.encryption.key_size)
 436         self.assertEqual(ControlLocation.back_end.value,
 437                          self.encryption.control_location)
 438
 439         encryption1 = cinder_utils.get_volume_encryption_by_type(
 440             self.cinder, self.volume_type)
 441         self.assertEquals(self.encryption, encryption1)
 442
 443     def test_create_bad_key_size(self):
 444         """
 445         Tests the cinder_utils.create_volume_encryption() raises an exception
 446         when the provider class does not exist
 447         """
 448         encryption_settings = VolumeTypeEncryptionConfig(
 449             name=self.encryption_name, provider_class='foo',
 450             cipher='bar', control_location=ControlLocation.back_end,
 451             key_size=-1)
 452
 453         with self.assertRaises(BadRequest):
 454             self.encryption = cinder_utils.create_volume_encryption(
 455                 self.cinder, self.volume_type, encryption_settings)
 456
 457
 458 class CinderUtilsVolumeTypeCompleteTests(OSComponentTestCase):
 459     """
 460     Tests to ensure that a volume type can have a QoS Spec added to it
 461     """
 462
 463     def setUp(self):
 464         """
 465         Creates objects for testing cinder_utils.py
 466         """
 467         guid = uuid.uuid4()
 468         self.qos_name = self.__class__.__name__ + '-' + str(guid) + '-qos'
 469         self.vol_type_name = self.__class__.__name__ + '-' + str(guid)
 470         self.specs = {'foo': 'bar'}
 471         self.cinder = cinder_utils.cinder_client(self.os_creds)
 472         qos_settings = QoSConfig(
 473             name=self.qos_name, specs=self.specs, consumer=Consumer.both)
 474         self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
 475         self.volume_type = None
 476
 477     def tearDown(self):
 478         """
 479         Cleans the remote OpenStack objects
 480         """
 481         if self.volume_type:
 482             if self.volume_type.encryption:
 483                 try:
 484                     cinder_utils.delete_volume_type_encryption(
 485                         self.cinder, self.volume_type)
 486                 except NotFound:
 487                     pass
 488             try:
 489                 cinder_utils.delete_volume_type(self.cinder, self.volume_type)
 490             except NotFound:
 491                 pass
 492
 493         if self.qos:
 494             try:
 495                 cinder_utils.delete_qos(self.cinder, self.qos)
 496             except NotFound:
 497                 pass
 498
 499     def test_create_with_encryption(self):
 500         """
 501         Tests the cinder_utils.create_volume_type() where encryption has been
 502         configured
 503         """
 504         encryption_settings = VolumeTypeEncryptionConfig(
 505             name='foo', provider_class='bar',
 506             control_location=ControlLocation.back_end)
 507         volume_type_settings = VolumeTypeConfig(
 508             name=self.vol_type_name, encryption=encryption_settings)
 509         self.volume_type = cinder_utils.create_volume_type(
 510             self.cinder, volume_type_settings)
 511
 512         vol_encrypt = cinder_utils.get_volume_encryption_by_type(
 513             self.cinder, self.volume_type)
 514         self.assertIsNotNone(vol_encrypt)
 515         self.assertIsNone(self.volume_type.qos_spec)
 516         self.assertEqual(self.volume_type.encryption, vol_encrypt)
 517         self.assertEqual(self.volume_type.id, vol_encrypt.volume_type_id)
 518
 519     def test_create_with_qos(self):
 520         """
 521         Tests the cinder_utils.create_volume_type() with an associated QoS Spec
 522         """
 523         volume_type_settings = VolumeTypeConfig(
 524             name=self.vol_type_name, qos_spec_name=self.qos_name)
 525         self.volume_type = cinder_utils.create_volume_type(
 526             self.cinder, volume_type_settings)
 527
 528         self.assertIsNotNone(self.volume_type)
 529         self.assertIsNone(self.volume_type.encryption)
 530         self.assertIsNotNone(self.volume_type.qos_spec)
 531         self.assertEqual(self.qos.id, self.volume_type.qos_spec.id)
 532
 533     def test_create_with_invalid_qos(self):
 534         """
 535         Tests the cinder_utils.create_volume_type() when the QoS Spec name
 536         does not exist
 537         """
 538         volume_type_settings = VolumeTypeConfig(
 539             name=self.vol_type_name, qos_spec_name='foo')
 540
 541         self.volume_type = cinder_utils.create_volume_type(
 542             self.cinder, volume_type_settings)
 543
 544         self.assertIsNone(self.volume_type.qos_spec)
 545
 546     def test_create_with_qos_and_encryption(self):
 547         """
 548         Tests the cinder_utils.create_volume_type() with encryption and an
 549         associated QoS Spec
 550         """
 551         encryption_settings = VolumeTypeEncryptionConfig(
 552             name='foo', provider_class='bar',
 553             control_location=ControlLocation.back_end)
 554         volume_type_settings = VolumeTypeConfig(
 555             name=self.vol_type_name, qos_spec_name=self.qos_name,
 556             encryption=encryption_settings)
 557         self.volume_type = cinder_utils.create_volume_type(
 558             self.cinder, volume_type_settings)
 559
 560         self.assertIsNotNone(self.volume_type)
 561         vol_encrypt = cinder_utils.get_volume_encryption_by_type(
 562             self.cinder, self.volume_type)
 563         self.assertIsNotNone(vol_encrypt)
 564         self.assertEqual(self.volume_type.id, vol_encrypt.volume_type_id)
 565         self.assertIsNotNone(self.volume_type.qos_spec)
 566         self.assertEqual(self.qos.id, self.volume_type.qos_spec.id)