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