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, OpenStackVolumeType)
  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.admin_os_creds)
 131         self.volume_type_creator = OpenStackVolumeType(
 132             self.admin_os_creds, self.volume_type_settings)
 133
 134     def tearDown(self):
 135         """
 136         Cleans the volume type
 137         """
 138         if self.volume_type_creator:
 139             self.volume_type_creator.clean()
 140
 141         super(self.__class__, self).__clean__()
 142
 143     def test_create_volume_type(self):
 144         """
 145         Tests the creation of an OpenStack volume.
 146         """
 147         # Create VolumeType
 148         created_volume_type = self.volume_type_creator.create()
 149         self.assertIsNotNone(created_volume_type)
 150         self.assertEqual(self.volume_type_settings.name,
 151                          created_volume_type.name)
 152
 153         retrieved_volume_type1 = cinder_utils.get_volume_type(
 154             self.cinder, volume_type_settings=self.volume_type_settings)
 155         self.assertIsNotNone(retrieved_volume_type1)
 156         self.assertEqual(created_volume_type, retrieved_volume_type1)
 157
 158         retrieved_volume_type2 = cinder_utils.get_volume_type_by_id(
 159             self.cinder, created_volume_type.id)
 160         self.assertEqual(created_volume_type, retrieved_volume_type2)
 161
 162     def test_create_delete_volume_type(self):
 163         """
 164         Tests the creation then deletion of an OpenStack volume type to ensure
 165         clean() does not raise an Exception.
 166         """
 167         # Create VolumeType
 168         created_volume_type = self.volume_type_creator.create()
 169         self.assertIsNotNone(created_volume_type)
 170
 171         retrieved_volume_type = cinder_utils.get_volume_type(
 172             self.cinder, volume_type_settings=self.volume_type_settings)
 173         self.assertIsNotNone(retrieved_volume_type)
 174         self.assertEqual(created_volume_type, retrieved_volume_type)
 175
 176         # Delete VolumeType manually
 177         cinder_utils.delete_volume_type(self.cinder, created_volume_type)
 178
 179         self.assertIsNone(cinder_utils.get_volume_type(
 180             self.cinder, volume_type_settings=self.volume_type_settings))
 181
 182         # Must not throw an exception when attempting to cleanup non-existent
 183         # volume_type
 184         self.volume_type_creator.clean()
 185         self.assertIsNone(self.volume_type_creator.get_volume_type())
 186
 187     def test_create_same_volume_type(self):
 188         """
 189         Tests the creation of an OpenStack volume_type when one already exists.
 190         """
 191         # Create VolumeType
 192         volume_type1 = self.volume_type_creator.create()
 193
 194         retrieved_volume_type = cinder_utils.get_volume_type(
 195             self.cinder, volume_type_settings=self.volume_type_settings)
 196         self.assertEqual(volume_type1, retrieved_volume_type)
 197
 198         # Should be retrieving the instance data
 199         os_volume_type_2 = create_volume_type.OpenStackVolumeType(
 200             self.admin_os_creds, self.volume_type_settings)
 201         volume_type2 = os_volume_type_2.create()
 202         self.assertEqual(volume_type2, volume_type2)
 203
 204
 205 class CreateVolumeTypeComplexTests(OSIntegrationTestCase):
 206     """
 207     Test cases for the CreateVolumeType class that include QoS Specs and/or
 208     encryption
 209     """
 210
 211     def setUp(self):
 212         super(self.__class__, self).__start__()
 213
 214         self.cinder = cinder_utils.cinder_client(self.admin_os_creds)
 215
 216         guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 217
 218         self.volume_type_name = guid + '-vol_type'
 219         self.volume_type_creator = None
 220
 221         qos_settings = QoSConfig(
 222             name=guid + '-qos-spec', consumer=Consumer.both)
 223         self.qos_creator = OpenStackQoS(self.admin_os_creds, qos_settings)
 224         self.qos_creator.create()
 225
 226     def tearDown(self):
 227         if self.volume_type_creator:
 228             self.volume_type_creator.clean()
 229
 230         if self.qos_creator:
 231             self.qos_creator.clean()
 232
 233         super(self.__class__, self).__clean__()
 234
 235     def test_volume_type_with_qos(self):
 236         """
 237         Creates a Volume Type object with an associated QoS Spec
 238         """
 239         self.volume_type_creator = OpenStackVolumeType(
 240             self.admin_os_creds,
 241             VolumeTypeConfig(
 242                 name=self.volume_type_name,
 243                 qos_spec_name=self.qos_creator.qos_settings.name))
 244
 245         vol_type = self.volume_type_creator.create()
 246         self.assertEqual(self.volume_type_creator.volume_type_settings.name,
 247                          vol_type.name)
 248         self.assertEqual(self.volume_type_creator.volume_type_settings.name,
 249                          vol_type.name)
 250         self.assertIsNotNone(vol_type.qos_spec)
 251         self.assertEqual(
 252             self.volume_type_creator.volume_type_settings.qos_spec_name,
 253             vol_type.qos_spec.name)
 254         self.assertIsNone(vol_type.encryption)
 255
 256         vol_type_query = cinder_utils.get_volume_type_by_id(
 257             self.cinder, vol_type.id)
 258         self.assertIsNotNone(vol_type_query)
 259         self.assertEqual(vol_type, vol_type_query)
 260
 261     def test_volume_type_with_encryption(self):
 262         """
 263         Creates a Volume Type object with encryption
 264         """
 265         encryption_settings = VolumeTypeEncryptionConfig(
 266             name='foo', provider_class='bar',
 267             control_location=ControlLocation.back_end)
 268         self.volume_type_creator = OpenStackVolumeType(
 269             self.admin_os_creds,
 270             VolumeTypeConfig(
 271                 name=self.volume_type_name,
 272                 encryption=encryption_settings))
 273
 274         vol_type = self.volume_type_creator.create()
 275         self.assertEqual(self.volume_type_creator.volume_type_settings.name,
 276                          vol_type.name)
 277         self.assertEqual(self.volume_type_creator.volume_type_settings.name,
 278                          vol_type.name)
 279         self.assertIsNone(vol_type.qos_spec)
 280         self.assertIsNotNone(vol_type.encryption)
 281
 282         self.assertEqual(encryption_settings.control_location.value,
 283                          vol_type.encryption.control_location)
 284
 285         vol_type_query = cinder_utils.get_volume_type_by_id(
 286             self.cinder, vol_type.id)
 287         self.assertIsNotNone(vol_type_query)
 288         self.assertEqual(vol_type, vol_type_query)
 289
 290     def test_volume_type_with_qos_and_encryption(self):
 291         """
 292         Creates a Volume Type object with encryption and an associated QoS Spec
 293         """
 294         encryption_settings = VolumeTypeEncryptionConfig(
 295             name='foo', provider_class='bar',
 296             control_location=ControlLocation.back_end)
 297         self.volume_type_creator = OpenStackVolumeType(
 298             self.admin_os_creds,
 299             VolumeTypeConfig(
 300                 name=self.volume_type_name,
 301                 encryption=encryption_settings,
 302                 qos_spec_name=self.qos_creator.qos_settings.name))
 303
 304         vol_type = self.volume_type_creator.create()
 305         self.assertEqual(self.volume_type_creator.volume_type_settings.name,
 306                          vol_type.name)
 307         self.assertEqual(self.volume_type_creator.volume_type_settings.name,
 308                          vol_type.name)
 309         self.assertIsNotNone(vol_type.qos_spec)
 310         self.assertEqual(
 311             self.volume_type_creator.volume_type_settings.qos_spec_name,
 312             vol_type.qos_spec.name)
 313         self.assertIsNotNone(vol_type.encryption)
 314
 315         self.assertEqual(encryption_settings.control_location.value,
 316                          vol_type.encryption.control_location)
 317
 318         vol_type_query = cinder_utils.get_volume_type_by_id(
 319             self.cinder, vol_type.id)
 320         self.assertIsNotNone(vol_type_query)
 321         self.assertEqual(vol_type, vol_type_query)