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(
 131             self.admin_os_creds, self.admin_os_session)
 132         self.volume_type_creator = OpenStackVolumeType(
 133             self.admin_os_creds, self.volume_type_settings)
 134
 135     def tearDown(self):
 136         """
 137         Cleans the volume type
 138         """
 139         if self.volume_type_creator:
 140             self.volume_type_creator.clean()
 141
 142         super(self.__class__, self).__clean__()
 143
 144     def test_create_volume_type(self):
 145         """
 146         Tests the creation of an OpenStack volume.
 147         """
 148         # Create VolumeType
 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         created_volume_type = self.volume_type_creator.create()
 170         self.assertIsNotNone(created_volume_type)
 171
 172         retrieved_volume_type = cinder_utils.get_volume_type(
 173             self.cinder, volume_type_settings=self.volume_type_settings)
 174         self.assertIsNotNone(retrieved_volume_type)
 175         self.assertEqual(created_volume_type, retrieved_volume_type)
 176
 177         # Delete VolumeType manually
 178         cinder_utils.delete_volume_type(self.cinder, created_volume_type)
 179
 180         self.assertIsNone(cinder_utils.get_volume_type(
 181             self.cinder, volume_type_settings=self.volume_type_settings))
 182
 183         # Must not throw an exception when attempting to cleanup non-existent
 184         # volume_type
 185         self.volume_type_creator.clean()
 186         self.assertIsNone(self.volume_type_creator.get_volume_type())
 187
 188     def test_create_same_volume_type(self):
 189         """
 190         Tests the creation of an OpenStack volume_type when one already exists.
 191         """
 192         # Create VolumeType
 193         volume_type1 = self.volume_type_creator.create()
 194
 195         retrieved_volume_type = cinder_utils.get_volume_type(
 196             self.cinder, volume_type_settings=self.volume_type_settings)
 197         self.assertEqual(volume_type1, retrieved_volume_type)
 198
 199         # Should be retrieving the instance data
 200         os_volume_type_2 = create_volume_type.OpenStackVolumeType(
 201             self.admin_os_creds, self.volume_type_settings)
 202         volume_type2 = os_volume_type_2.create()
 203         self.assertEqual(volume_type2, volume_type2)
 204
 205
 206 class CreateVolumeTypeComplexTests(OSIntegrationTestCase):
 207     """
 208     Test cases for the CreateVolumeType class that include QoS Specs and/or
 209     encryption
 210     """
 211
 212     def setUp(self):
 213         super(self.__class__, self).__start__()
 214
 215         self.cinder = cinder_utils.cinder_client(
 216             self.admin_os_creds, self.admin_os_session)
 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.admin_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.admin_os_creds,
 243             VolumeTypeConfig(
 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 = VolumeTypeEncryptionConfig(
 268             name='foo', provider_class='bar',
 269             control_location=ControlLocation.back_end)
 270         self.volume_type_creator = OpenStackVolumeType(
 271             self.admin_os_creds,
 272             VolumeTypeConfig(
 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 = VolumeTypeEncryptionConfig(
 297             name='foo', provider_class='bar',
 298             control_location=ControlLocation.back_end)
 299         self.volume_type_creator = OpenStackVolumeType(
 300             self.admin_os_creds,
 301             VolumeTypeConfig(
 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)