snaps/openstack/tests/create_volume_type_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 from snaps.config.volume_type import (
  16     VolumeTypeConfig, VolumeTypeEncryptionConfig, VolumeTypeConfigError,
  17     ControlLocation)
  18 from snaps.config.qos import QoSConfig, Consumer
  19 from snaps.openstack.create_qos import OpenStackQoS
  20
  21 try:
  22     from urllib.request import URLError
  23 except ImportError:
  24     from urllib2 import URLError
  25
  26 import logging
  27 import unittest
  28 import uuid
  29
  30 from snaps.openstack import create_volume_type
  31 from snaps.openstack.create_volume_type import (
  32     VolumeTypeSettings, VolumeTypeEncryptionSettings)
  33 from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
  34 from snaps.openstack.utils import cinder_utils
  35
  36 __author__ = 'spisarski'
  37
  38 logger = logging.getLogger('create_volume_type_tests')
  39
  40
  41 class VolumeTypeSettingsUnitTests(unittest.TestCase):
  42     """
  43     Tests the construction of the VolumeTypeSettings class
  44     """
  45
  46     def test_no_params(self):
  47         with self.assertRaises(VolumeTypeConfigError):
  48             VolumeTypeSettings()
  49
  50     def test_empty_config(self):
  51         with self.assertRaises(VolumeTypeConfigError):
  52             VolumeTypeSettings(**dict())
  53
  54     def test_name_only(self):
  55         settings = VolumeTypeSettings(name='foo')
  56         self.assertEqual('foo', settings.name)
  57         self.assertIsNone(settings.description)
  58         self.assertIsNone(settings.qos_spec_name)
  59         self.assertIsNone(settings.encryption)
  60         self.assertFalse(settings.public)
  61
  62     def test_config_with_name_only(self):
  63         settings = VolumeTypeSettings(**{'name': 'foo'})
  64         self.assertEqual('foo', settings.name)
  65         self.assertIsNone(settings.description)
  66         self.assertIsNone(settings.qos_spec_name)
  67         self.assertIsNone(settings.encryption)
  68         self.assertFalse(settings.public)
  69
  70     def test_all(self):
  71         encryption_settings = VolumeTypeEncryptionSettings(
  72             name='foo', provider_class='bar',
  73             control_location=ControlLocation.back_end)
  74         settings = VolumeTypeSettings(
  75             name='foo', description='desc', encryption=encryption_settings,
  76             qos_spec_name='spec_name', public=True)
  77         self.assertEqual('foo', settings.name)
  78         self.assertEqual('desc', settings.description)
  79         self.assertEqual('spec_name', settings.qos_spec_name)
  80         self.assertEqual(encryption_settings, settings.encryption)
  81         self.assertTrue(True, settings.public)
  82
  83     def test_all_string(self):
  84         encryption_settings = {
  85             'name': 'foo', 'provider_class': 'bar',
  86             'control_location': 'back-end'}
  87         settings = VolumeTypeSettings(
  88             name='foo', description='desc', encryption=encryption_settings,
  89             qos_spec_name='spec_name', public='true')
  90         self.assertEqual('foo', settings.name)
  91         self.assertEqual('desc', settings.description)
  92         self.assertEqual('spec_name', settings.qos_spec_name)
  93         self.assertEqual(VolumeTypeEncryptionSettings(**encryption_settings),
  94                          settings.encryption)
  95         self.assertTrue(settings.public)
  96
  97     def test_config_all(self):
  98         encryption_settings = {
  99             'name': 'foo', 'provider_class': 'bar',
 100             'control_location': 'back-end'}
 101         settings = VolumeTypeSettings(
 102             **{'name': 'foo', 'description': 'desc',
 103                'encryption': encryption_settings,
 104                'qos_spec_name': 'spec_name', 'public': 'false'})
 105         self.assertEqual('foo', settings.name)
 106         self.assertEqual('desc', settings.description)
 107         self.assertEqual('spec_name', settings.qos_spec_name)
 108         self.assertEqual(VolumeTypeEncryptionSettings(**encryption_settings),
 109                          settings.encryption)
 110         self.assertFalse(settings.public)
 111
 112
 113 class CreateSimpleVolumeTypeSuccessTests(OSIntegrationTestCase):
 114     """
 115     Test for the OpenStackVolumeType class defined in py
 116     without any QoS Specs or Encryption
 117     """
 118
 119     def setUp(self):
 120         """
 121         Instantiates the CreateVolumeType object that is responsible for
 122         downloading and creating an OS volume type file within OpenStack
 123         """
 124         super(self.__class__, self).__start__()
 125
 126         guid = uuid.uuid4()
 127         self.volume_type_settings = VolumeTypeConfig(
 128             name=self.__class__.__name__ + '-' + str(guid))
 129
 130         self.cinder = cinder_utils.cinder_client(self.os_creds)
 131         self.volume_type_creator = None
 132
 133     def tearDown(self):
 134         """
 135         Cleans the volume type
 136         """
 137         if self.volume_type_creator:
 138             self.volume_type_creator.clean()
 139
 140         super(self.__class__, self).__clean__()
 141
 142     def test_create_volume_type(self):
 143         """
 144         Tests the creation of an OpenStack volume.
 145         """
 146         # Create VolumeType
 147         self.volume_type_creator = create_volume_type.OpenStackVolumeType(
 148             self.os_creds, self.volume_type_settings)
 149         created_volume_type = self.volume_type_creator.create()
 150         self.assertIsNotNone(created_volume_type)
 151         self.assertEqual(self.volume_type_settings.name,
 152                          created_volume_type.name)
 153
 154         retrieved_volume_type1 = cinder_utils.get_volume_type(
 155             self.cinder, volume_type_settings=self.volume_type_settings)
 156         self.assertIsNotNone(retrieved_volume_type1)
 157         self.assertEqual(created_volume_type, retrieved_volume_type1)
 158
 159         retrieved_volume_type2 = cinder_utils.get_volume_type_by_id(
 160             self.cinder, created_volume_type.id)
 161         self.assertEqual(created_volume_type, retrieved_volume_type2)
 162
 163     def test_create_delete_volume_type(self):
 164         """
 165         Tests the creation then deletion of an OpenStack volume type to ensure
 166         clean() does not raise an Exception.
 167         """
 168         # Create VolumeType
 169         self.volume_type_creator = create_volume_type.OpenStackVolumeType(
 170             self.os_creds, self.volume_type_settings)
 171         created_volume_type = self.volume_type_creator.create()
 172         self.assertIsNotNone(created_volume_type)
 173
 174         retrieved_volume_type = cinder_utils.get_volume_type(
 175             self.cinder, volume_type_settings=self.volume_type_settings)
 176         self.assertIsNotNone(retrieved_volume_type)
 177         self.assertEqual(created_volume_type, retrieved_volume_type)
 178
 179         # Delete VolumeType manually
 180         cinder_utils.delete_volume_type(self.cinder, created_volume_type)
 181
 182         self.assertIsNone(cinder_utils.get_volume_type(
 183             self.cinder, volume_type_settings=self.volume_type_settings))
 184
 185         # Must not throw an exception when attempting to cleanup non-existent
 186         # volume_type
 187         self.volume_type_creator.clean()
 188         self.assertIsNone(self.volume_type_creator.get_volume_type())
 189
 190     def test_create_same_volume_type(self):
 191         """
 192         Tests the creation of an OpenStack volume_type when one already exists.
 193         """
 194         # Create VolumeType
 195         self.volume_type_creator = create_volume_type.OpenStackVolumeType(
 196             self.os_creds, self.volume_type_settings)
 197         volume_type1 = self.volume_type_creator.create()
 198
 199         retrieved_volume_type = cinder_utils.get_volume_type(
 200             self.cinder, volume_type_settings=self.volume_type_settings)
 201         self.assertEqual(volume_type1, retrieved_volume_type)
 202
 203         # Should be retrieving the instance data
 204         os_volume_type_2 = create_volume_type.OpenStackVolumeType(
 205             self.os_creds, self.volume_type_settings)
 206         volume_type2 = os_volume_type_2.create()
 207         self.assertEqual(volume_type2, volume_type2)
 208
 209
 210 class CreateVolumeTypeComplexTests(OSIntegrationTestCase):
 211     """
 212     Test cases for the CreateVolumeType class that include QoS Specs and/or
 213     encryption
 214     """
 215
 216     def setUp(self):
 217         super(self.__class__, self).__start__()
 218
 219         self.cinder = cinder_utils.cinder_client(self.os_creds)
 220
 221         guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 222
 223         self.volume_type_name = guid + '-vol_type'
 224         self.volume_type_creator = None
 225
 226         qos_settings = QoSConfig(
 227             name=guid + '-qos-spec', consumer=Consumer.both)
 228         self.qos_creator = OpenStackQoS(self.os_creds, qos_settings)
 229         self.qos_creator.create()
 230
 231     def tearDown(self):
 232         if self.volume_type_creator:
 233             self.volume_type_creator.clean()
 234
 235         if self.qos_creator:
 236             self.qos_creator.clean()
 237
 238         super(self.__class__, self).__clean__()
 239
 240     def test_volume_type_with_qos(self):
 241         """
 242         Creates a Volume Type object with an associated QoS Spec
 243         """
 244         self.volume_type_creator = create_volume_type.OpenStackVolumeType(
 245             self.os_creds,
 246             VolumeTypeConfig(
 247                 name=self.volume_type_name,
 248                 qos_spec_name=self.qos_creator.qos_settings.name))
 249
 250         vol_type = self.volume_type_creator.create()
 251         self.assertEqual(self.volume_type_creator.volume_type_settings.name,
 252                          vol_type.name)
 253         self.assertEqual(self.volume_type_creator.volume_type_settings.name,
 254                          vol_type.name)
 255         self.assertIsNotNone(vol_type.qos_spec)
 256         self.assertEqual(
 257             self.volume_type_creator.volume_type_settings.qos_spec_name,
 258             vol_type.qos_spec.name)
 259         self.assertIsNone(vol_type.encryption)
 260
 261         vol_type_query = cinder_utils.get_volume_type_by_id(
 262             self.cinder, vol_type.id)
 263         self.assertIsNotNone(vol_type_query)
 264         self.assertEqual(vol_type, vol_type_query)
 265
 266     def test_volume_type_with_encryption(self):
 267         """
 268         Creates a Volume Type object with encryption
 269         """
 270         encryption_settings = VolumeTypeEncryptionConfig(
 271             name='foo', provider_class='bar',
 272             control_location=ControlLocation.back_end)
 273         self.volume_type_creator = create_volume_type.OpenStackVolumeType(
 274             self.os_creds,
 275             VolumeTypeConfig(
 276                 name=self.volume_type_name,
 277                 encryption=encryption_settings))
 278
 279         vol_type = self.volume_type_creator.create()
 280         self.assertEqual(self.volume_type_creator.volume_type_settings.name,
 281                          vol_type.name)
 282         self.assertEqual(self.volume_type_creator.volume_type_settings.name,
 283                          vol_type.name)
 284         self.assertIsNone(vol_type.qos_spec)
 285         self.assertIsNotNone(vol_type.encryption)
 286
 287         self.assertEqual(encryption_settings.control_location.value,
 288                          vol_type.encryption.control_location)
 289
 290         vol_type_query = cinder_utils.get_volume_type_by_id(
 291             self.cinder, vol_type.id)
 292         self.assertIsNotNone(vol_type_query)
 293         self.assertEqual(vol_type, vol_type_query)
 294
 295     def test_volume_type_with_qos_and_encryption(self):
 296         """
 297         Creates a Volume Type object with encryption and an associated QoS Spec
 298         """
 299         encryption_settings = VolumeTypeEncryptionConfig(
 300             name='foo', provider_class='bar',
 301             control_location=ControlLocation.back_end)
 302         self.volume_type_creator = create_volume_type.OpenStackVolumeType(
 303             self.os_creds,
 304             VolumeTypeConfig(
 305                 name=self.volume_type_name,
 306                 encryption=encryption_settings,
 307                 qos_spec_name=self.qos_creator.qos_settings.name))
 308
 309         vol_type = self.volume_type_creator.create()
 310         self.assertEqual(self.volume_type_creator.volume_type_settings.name,
 311                          vol_type.name)
 312         self.assertEqual(self.volume_type_creator.volume_type_settings.name,
 313                          vol_type.name)
 314         self.assertIsNotNone(vol_type.qos_spec)
 315         self.assertEqual(
 316             self.volume_type_creator.volume_type_settings.qos_spec_name,
 317             vol_type.qos_spec.name)
 318         self.assertIsNotNone(vol_type.encryption)
 319
 320         self.assertEqual(encryption_settings.control_location.value,
 321                          vol_type.encryption.control_location)
 322
 323         vol_type_query = cinder_utils.get_volume_type_by_id(
 324             self.cinder, vol_type.id)
 325         self.assertIsNotNone(vol_type_query)
 326         self.assertEqual(vol_type, vol_type_query)