snaps/openstack/tests/create_volume_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 cinderclient.exceptions import NotFound, BadRequest
  16
  17 from snaps.openstack.create_image import OpenStackImage
  18 from snaps.openstack.create_volume_type import (
  19     VolumeTypeSettings, OpenStackVolumeType)
  20 from snaps.openstack.tests import openstack_tests
  21
  22 try:
  23     from urllib.request import URLError
  24 except ImportError:
  25     from urllib2 import URLError
  26
  27 import logging
  28 import unittest
  29 import uuid
  30
  31 from snaps.openstack import create_volume
  32 from snaps.openstack.create_volume import (
  33     VolumeSettings, VolumeSettingsError, OpenStackVolume)
  34 from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
  35 from snaps.openstack.utils import cinder_utils
  36
  37 __author__ = 'spisarski'
  38
  39 logger = logging.getLogger('create_volume_tests')
  40
  41
  42 class VolumeSettingsUnitTests(unittest.TestCase):
  43     """
  44     Tests the construction of the VolumeSettings class
  45     """
  46
  47     def test_no_params(self):
  48         with self.assertRaises(VolumeSettingsError):
  49             VolumeSettings()
  50
  51     def test_empty_config(self):
  52         with self.assertRaises(VolumeSettingsError):
  53             VolumeSettings(**dict())
  54
  55     def test_name_only(self):
  56         settings = VolumeSettings(name='foo')
  57         self.assertEqual('foo', settings.name)
  58         self.assertIsNone(settings.description)
  59         self.assertEquals(1, settings.size)
  60         self.assertIsNone(settings.image_name)
  61         self.assertIsNone(settings.type_name)
  62         self.assertIsNone(settings.availability_zone)
  63         self.assertFalse(settings.multi_attach)
  64
  65     def test_config_with_name_only(self):
  66         settings = VolumeSettings(**{'name': 'foo'})
  67         self.assertEqual('foo', settings.name)
  68         self.assertIsNone(settings.description)
  69         self.assertEquals(1, settings.size)
  70         self.assertIsNone(settings.image_name)
  71         self.assertIsNone(settings.type_name)
  72         self.assertIsNone(settings.availability_zone)
  73         self.assertFalse(settings.multi_attach)
  74
  75     def test_all_strings(self):
  76         settings = VolumeSettings(
  77             name='foo', description='desc', size='2', image_name='image',
  78             type_name='type', availability_zone='zone1', multi_attach='true')
  79
  80         self.assertEqual('foo', settings.name)
  81         self.assertEqual('desc', settings.description)
  82         self.assertEqual(2, settings.size)
  83         self.assertEqual('image', settings.image_name)
  84         self.assertEqual('type', settings.type_name)
  85         self.assertEqual('zone1', settings.availability_zone)
  86         self.assertTrue(settings.multi_attach)
  87
  88     def test_all_correct_type(self):
  89         settings = VolumeSettings(
  90             name='foo', description='desc', size=2, image_name='image',
  91             type_name='bar', availability_zone='zone1', multi_attach=True)
  92
  93         self.assertEqual('foo', settings.name)
  94         self.assertEqual('desc', settings.description)
  95         self.assertEqual(2, settings.size)
  96         self.assertEqual('image', settings.image_name)
  97         self.assertEqual('bar', settings.type_name)
  98         self.assertEqual('zone1', settings.availability_zone)
  99         self.assertTrue(settings.multi_attach)
 100
 101     def test_config_all(self):
 102         settings = VolumeSettings(
 103             **{'name': 'foo', 'description': 'desc', 'size': '2',
 104                'image_name': 'foo', 'type_name': 'bar',
 105                'availability_zone': 'zone1', 'multi_attach': 'true'})
 106
 107         self.assertEqual('foo', settings.name)
 108         self.assertEqual('desc', settings.description)
 109         self.assertEqual(2, settings.size)
 110         self.assertEqual('foo', settings.image_name)
 111         self.assertEqual('bar', settings.type_name)
 112         self.assertEqual('zone1', settings.availability_zone)
 113         self.assertTrue(settings.multi_attach)
 114
 115
 116 class CreateSimpleVolumeSuccessTests(OSIntegrationTestCase):
 117     """
 118     Test for the CreateVolume class defined in create_volume.py
 119     """
 120
 121     def setUp(self):
 122         """
 123         Instantiates the CreateVolume object that is responsible for
 124         downloading and creating an OS volume file within OpenStack
 125         """
 126         super(self.__class__, self).__start__()
 127
 128         guid = uuid.uuid4()
 129         self.volume_settings = VolumeSettings(
 130             name=self.__class__.__name__ + '-' + str(guid))
 131
 132         self.cinder = cinder_utils.cinder_client(self.os_creds)
 133         self.volume_creator = None
 134
 135     def tearDown(self):
 136         """
 137         Cleans the volume and downloaded volume file
 138         """
 139         if self.volume_creator:
 140             self.volume_creator.clean()
 141
 142         super(self.__class__, self).__clean__()
 143
 144     def test_create_volume_simple(self):
 145         """
 146         Tests the creation of an OpenStack volume from a URL.
 147         """
 148         # Create Volume
 149         self.volume_creator = create_volume.OpenStackVolume(
 150             self.os_creds, self.volume_settings)
 151         created_volume = self.volume_creator.create(block=True)
 152         self.assertIsNotNone(created_volume)
 153
 154         retrieved_volume = cinder_utils.get_volume(
 155             self.cinder, volume_settings=self.volume_settings)
 156
 157         self.assertIsNotNone(retrieved_volume)
 158         self.assertEqual(created_volume.id, retrieved_volume.id)
 159         self.assertTrue(created_volume == retrieved_volume)
 160
 161     def test_create_delete_volume(self):
 162         """
 163         Tests the creation then deletion of an OpenStack volume to ensure
 164         clean() does not raise an Exception.
 165         """
 166         # Create Volume
 167         self.volume_creator = create_volume.OpenStackVolume(
 168             self.os_creds, self.volume_settings)
 169         created_volume = self.volume_creator.create(block=True)
 170         self.assertIsNotNone(created_volume)
 171
 172         retrieved_volume = cinder_utils.get_volume(
 173             self.cinder, volume_settings=self.volume_settings)
 174         self.assertIsNotNone(retrieved_volume)
 175         self.assertEqual(created_volume, retrieved_volume)
 176
 177         # Delete Volume manually
 178         self.volume_creator.clean()
 179
 180         self.assertIsNone(cinder_utils.get_volume(
 181             self.cinder, volume_settings=self.volume_settings))
 182
 183         # Must not throw an exception when attempting to cleanup non-existent
 184         # volume
 185         self.volume_creator.clean()
 186         self.assertIsNone(self.volume_creator.get_volume())
 187
 188     def test_create_same_volume(self):
 189         """
 190         Tests the creation of an OpenStack volume when one already exists.
 191         """
 192         # Create Volume
 193         self.volume_creator = OpenStackVolume(
 194             self.os_creds, self.volume_settings)
 195         volume1 = self.volume_creator.create(block=True)
 196
 197         retrieved_volume = cinder_utils.get_volume(
 198             self.cinder, volume_settings=self.volume_settings)
 199         self.assertEqual(volume1, retrieved_volume)
 200
 201         # Should be retrieving the instance data
 202         os_volume_2 = OpenStackVolume(
 203             self.os_creds, self.volume_settings)
 204         volume2 = os_volume_2.create(block=True)
 205         self.assertEqual(volume1, volume2)
 206
 207
 208 class CreateVolumeWithTypeTests(OSIntegrationTestCase):
 209     """
 210     Test cases for the CreateVolume when attempting to associate it to a
 211     Volume Type
 212     """
 213
 214     def setUp(self):
 215         super(self.__class__, self).__start__()
 216
 217         guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 218         self.volume_name = guid + '-vol'
 219         self.volume_type_name = guid + '-vol-type'
 220
 221         self.volume_type_creator = OpenStackVolumeType(
 222             self.os_creds, VolumeTypeSettings(name=self.volume_type_name))
 223         self.volume_type_creator.create()
 224         self.volume_creator = None
 225
 226     def tearDown(self):
 227         if self.volume_creator:
 228             self.volume_creator.clean()
 229         if self.volume_type_creator:
 230             self.volume_type_creator.clean()
 231
 232         super(self.__class__, self).__clean__()
 233
 234     def test_bad_volume_type(self):
 235         """
 236         Expect a NotFound to be raised when the volume type does not exist
 237         """
 238         self.volume_creator = OpenStackVolume(
 239             self.os_creds,
 240             VolumeSettings(name=self.volume_name, type_name='foo'))
 241
 242         with self.assertRaises(NotFound):
 243             self.volume_creator.create()
 244
 245     def test_valid_volume_type(self):
 246         """
 247         Expect a NotFound to be raised when the volume type does not exist
 248         """
 249         self.volume_creator = OpenStackVolume(
 250             self.os_creds,
 251             VolumeSettings(name=self.volume_name,
 252                            type_name=self.volume_type_name))
 253
 254         created_volume = self.volume_creator.create()
 255         self.assertIsNotNone(created_volume)
 256         self.assertEqual(self.volume_type_name, created_volume.type)
 257
 258
 259 class CreateVolumeWithImageTests(OSIntegrationTestCase):
 260     """
 261     Test cases for the CreateVolume when attempting to associate it to an Image
 262     """
 263
 264     def setUp(self):
 265         super(self.__class__, self).__start__()
 266
 267         guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 268         self.volume_name = guid + '-vol'
 269         self.image_name = guid + '-image'
 270
 271         os_image_settings = openstack_tests.cirros_image_settings(
 272             name=self.image_name, image_metadata=self.image_metadata)
 273         # Create Image
 274         self.image_creator = OpenStackImage(self.os_creds,
 275                                             os_image_settings)
 276         self.image_creator.create()
 277         self.volume_creator = None
 278
 279     def tearDown(self):
 280         if self.volume_creator:
 281             try:
 282                 self.volume_creator.clean()
 283             except:
 284                 pass
 285         if self.image_creator:
 286             try:
 287                 self.image_creator.clean()
 288             except:
 289                 pass
 290
 291         super(self.__class__, self).__clean__()
 292
 293     def test_bad_image_name(self):
 294         """
 295         Expect a NotFound to be raised when the volume type does not exist
 296         """
 297         self.volume_creator = OpenStackVolume(
 298             self.os_creds,
 299             VolumeSettings(name=self.volume_name, image_name='foo'))
 300
 301         with self.assertRaises(BadRequest):
 302             self.volume_creator.create(block=True)
 303
 304     def test_valid_volume_image(self):
 305         """
 306         Expect a NotFound to be raised when the volume type does not exist
 307         """
 308         self.volume_creator = OpenStackVolume(
 309             self.os_creds,
 310             VolumeSettings(name=self.volume_name, image_name=self.image_name))
 311
 312         created_volume = self.volume_creator.create(block=True)
 313         self.assertIsNotNone(created_volume)
 314         self.assertIsNone(created_volume.type)
 315         self.assertTrue(self.volume_creator.volume_active())