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.config.volume import VolumeConfig, VolumeConfigError
  18 from snaps.config.volume_type import VolumeTypeConfig
  19 from snaps.openstack.create_image import OpenStackImage
  20 from snaps.openstack.create_volume_type import OpenStackVolumeType
  21 from snaps.openstack.tests import openstack_tests
  22
  23 try:
  24     from urllib.request import URLError
  25 except ImportError:
  26     from urllib2 import URLError
  27
  28 import logging
  29 import unittest
  30 import uuid
  31
  32 from snaps.openstack.create_volume import (
  33     VolumeSettings, OpenStackVolume)
  34 from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
  35 from snaps.openstack.utils import cinder_utils, keystone_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(VolumeConfigError):
  49             VolumeSettings()
  50
  51     def test_empty_config(self):
  52         with self.assertRaises(VolumeConfigError):
  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 = VolumeConfig(
 130             name=self.__class__.__name__ + '-' + str(guid))
 131
 132         self.cinder = cinder_utils.cinder_client(
 133             self.os_creds, self.os_session)
 134         self.keystone = keystone_utils.keystone_client(
 135             self.os_creds, self.os_session)
 136         self.volume_creator = None
 137
 138     def tearDown(self):
 139         """
 140         Cleans the volume and downloaded volume file
 141         """
 142         if self.volume_creator:
 143             self.volume_creator.clean()
 144
 145         super(self.__class__, self).__clean__()
 146
 147     def test_create_volume_simple(self):
 148         """
 149         Tests the creation of a simple OpenStack volume.
 150         """
 151         # Create Volume
 152         self.volume_creator = OpenStackVolume(
 153             self.os_creds, self.volume_settings)
 154         created_volume = self.volume_creator.create(block=True)
 155         self.assertIsNotNone(created_volume)
 156
 157         retrieved_volume = cinder_utils.get_volume(
 158             self.cinder, self.keystone, volume_settings=self.volume_settings,
 159             project_name=self.os_creds.project_name)
 160
 161         self.assertIsNotNone(retrieved_volume)
 162         self.assertEqual(created_volume.id, retrieved_volume.id)
 163         self.assertTrue(created_volume == retrieved_volume)
 164
 165     def test_create_delete_volume(self):
 166         """
 167         Tests the creation then deletion of an OpenStack volume to ensure
 168         clean() does not raise an Exception.
 169         """
 170         # Create Volume
 171         self.volume_creator = OpenStackVolume(
 172             self.os_creds, self.volume_settings)
 173         created_volume = self.volume_creator.create(block=True)
 174         self.assertIsNotNone(created_volume)
 175
 176         retrieved_volume = cinder_utils.get_volume(
 177             self.cinder, self.keystone, volume_settings=self.volume_settings,
 178             project_name=self.os_creds.project_name)
 179         self.assertIsNotNone(retrieved_volume)
 180         self.assertEqual(created_volume, retrieved_volume)
 181
 182         # Delete Volume manually
 183         self.volume_creator.clean()
 184
 185         self.assertIsNone(cinder_utils.get_volume(
 186             self.cinder, self.keystone, volume_settings=self.volume_settings,
 187             project_name=self.os_creds.project_name))
 188
 189         # Must not throw an exception when attempting to cleanup non-existent
 190         # volume
 191         self.volume_creator.clean()
 192         self.assertIsNone(self.volume_creator.get_volume())
 193
 194     def test_create_same_volume(self):
 195         """
 196         Tests the creation of an OpenStack volume when one already exists.
 197         """
 198         # Create Volume
 199         self.volume_creator = OpenStackVolume(
 200             self.os_creds, self.volume_settings)
 201         volume1 = self.volume_creator.create(block=True)
 202
 203         retrieved_volume = cinder_utils.get_volume(
 204             self.cinder, self.keystone, volume_settings=self.volume_settings,
 205             project_name=self.os_creds.project_name)
 206         self.assertEqual(volume1, retrieved_volume)
 207
 208         # Should be retrieving the instance data
 209         os_volume_2 = OpenStackVolume(
 210             self.os_creds, self.volume_settings)
 211         volume2 = os_volume_2.create(block=True)
 212         self.assertEqual(volume1, volume2)
 213
 214
 215 class CreateSimpleVolumeFailureTests(OSIntegrationTestCase):
 216     """
 217     Test for the CreateVolume class defined in create_volume.py
 218     """
 219
 220     def setUp(self):
 221         """
 222         Instantiates the CreateVolume object that is responsible for
 223         downloading and creating an OS volume file within OpenStack
 224         """
 225         super(self.__class__, self).__start__()
 226
 227         self.guid = uuid.uuid4()
 228         self.cinder = cinder_utils.cinder_client(
 229             self.os_creds, self.os_session)
 230         self.volume_creator = None
 231
 232     def tearDown(self):
 233         """
 234         Cleans the volume and downloaded volume file
 235         """
 236         if self.volume_creator:
 237             self.volume_creator.clean()
 238
 239         super(self.__class__, self).__clean__()
 240
 241     def test_create_volume_bad_size(self):
 242         """
 243         Tests the creation of an OpenStack volume with a negative size to
 244         ensure it raises a BadRequest exception.
 245         """
 246         volume_settings = VolumeConfig(
 247             name=self.__class__.__name__ + '-' + str(self.guid), size=-1)
 248
 249         # Create Volume
 250         self.volume_creator = OpenStackVolume(self.os_creds, volume_settings)
 251
 252         with self.assertRaises(BadRequest):
 253             self.volume_creator.create(block=True)
 254
 255     def test_create_volume_bad_type(self):
 256         """
 257         Tests the creation of an OpenStack volume with a type that does not
 258         exist to ensure it raises a NotFound exception.
 259         """
 260         volume_settings = VolumeConfig(
 261             name=self.__class__.__name__ + '-' + str(self.guid),
 262             type_name='foo')
 263
 264         # Create Volume
 265         self.volume_creator = OpenStackVolume(self.os_creds, volume_settings)
 266
 267         with self.assertRaises(NotFound):
 268             self.volume_creator.create(block=True)
 269
 270     def test_create_volume_bad_image(self):
 271         """
 272         Tests the creation of an OpenStack volume with an image that does not
 273         exist to ensure it raises a BadRequest exception.
 274         """
 275         volume_settings = VolumeConfig(
 276             name=self.__class__.__name__ + '-' + str(self.guid),
 277             image_name='foo')
 278
 279         # Create Volume
 280         self.volume_creator = OpenStackVolume(self.os_creds, volume_settings)
 281
 282         with self.assertRaises(BadRequest):
 283             self.volume_creator.create(block=True)
 284
 285     def test_create_volume_bad_zone(self):
 286         """
 287         Tests the creation of an OpenStack volume with an availability zone
 288         that does not exist to ensure it raises a BadRequest exception.
 289         """
 290         volume_settings = VolumeConfig(
 291             name=self.__class__.__name__ + '-' + str(self.guid),
 292             availability_zone='foo')
 293
 294         # Create Volume
 295         self.volume_creator = OpenStackVolume(self.os_creds, volume_settings)
 296
 297         with self.assertRaises(BadRequest):
 298             self.volume_creator.create(block=True)
 299
 300
 301 class CreateVolumeWithTypeTests(OSIntegrationTestCase):
 302     """
 303     Test cases for the CreateVolume when attempting to associate it to a
 304     Volume Type
 305     """
 306
 307     def setUp(self):
 308         super(self.__class__, self).__start__()
 309
 310         guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 311         self.volume_name = guid + '-vol'
 312         self.volume_type_name = guid + '-vol-type'
 313
 314         self.volume_type_creator = OpenStackVolumeType(
 315             self.admin_os_creds, VolumeTypeConfig(name=self.volume_type_name))
 316         self.volume_type_creator.create()
 317         self.volume_creator = None
 318
 319     def tearDown(self):
 320         if self.volume_creator:
 321             self.volume_creator.clean()
 322         if self.volume_type_creator:
 323             self.volume_type_creator.clean()
 324
 325         super(self.__class__, self).__clean__()
 326
 327     def test_bad_volume_type(self):
 328         """
 329         Expect a NotFound to be raised when the volume type does not exist
 330         """
 331         self.volume_creator = OpenStackVolume(
 332             self.os_creds,
 333             VolumeConfig(name=self.volume_name, type_name='foo'))
 334
 335         with self.assertRaises(NotFound):
 336             self.volume_creator.create()
 337
 338     def test_valid_volume_type(self):
 339         """
 340         Expect a NotFound to be raised when the volume type does not exist
 341         """
 342         self.volume_creator = OpenStackVolume(
 343             self.admin_os_creds, VolumeConfig(
 344                 name=self.volume_name, type_name=self.volume_type_name))
 345
 346         created_volume = self.volume_creator.create(block=True)
 347         self.assertIsNotNone(created_volume)
 348         self.assertEqual(self.volume_type_name, created_volume.type)
 349
 350
 351 class CreateVolumeWithImageTests(OSIntegrationTestCase):
 352     """
 353     Test cases for the CreateVolume when attempting to associate it to an Image
 354     """
 355
 356     def setUp(self):
 357         super(self.__class__, self).__start__()
 358
 359         self.cinder = cinder_utils.cinder_client(
 360             self.os_creds, self.os_session)
 361
 362         guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 363         self.volume_name = guid + '-vol'
 364         self.image_name = guid + '-image'
 365
 366         os_image_settings = openstack_tests.cirros_image_settings(
 367             name=self.image_name, image_metadata=self.image_metadata)
 368         # Create Image
 369         self.image_creator = OpenStackImage(self.os_creds,
 370                                             os_image_settings)
 371         self.image_creator.create()
 372         self.volume_creator = None
 373
 374     def tearDown(self):
 375         if self.volume_creator:
 376             try:
 377                 self.volume_creator.clean()
 378             except:
 379                 pass
 380         if self.image_creator:
 381             try:
 382                 self.image_creator.clean()
 383             except:
 384                 pass
 385
 386         super(self.__class__, self).__clean__()
 387
 388     def test_bad_image_name(self):
 389         """
 390         Tests OpenStackVolume#create() method to ensure a volume is NOT created
 391         when associating it to an invalid image name
 392         """
 393         self.volume_creator = OpenStackVolume(
 394             self.os_creds,
 395             VolumeConfig(name=self.volume_name, image_name='foo'))
 396
 397         with self.assertRaises(BadRequest):
 398             self.volume_creator.create(block=True)
 399
 400     def test_valid_volume_image(self):
 401         """
 402         Tests OpenStackVolume#create() method to ensure a volume is NOT created
 403         when associating it to an invalid image name
 404         """
 405         self.volume_creator = OpenStackVolume(
 406             self.os_creds,
 407             VolumeConfig(name=self.volume_name, image_name=self.image_name))
 408
 409         created_volume = self.volume_creator.create(block=True)
 410         self.assertIsNotNone(created_volume)
 411         self.assertEqual(
 412             self.volume_creator.volume_settings.name, created_volume.name)
 413         self.assertTrue(self.volume_creator.volume_active())
 414
 415         retrieved_volume = cinder_utils.get_volume_by_id(
 416             self.cinder, created_volume.id)
 417
 418         self.assertEqual(created_volume, retrieved_volume)
 419
 420
 421 class CreateVolMultipleCredsTests(OSIntegrationTestCase):
 422     """
 423     Test for the OpenStackVolume class and how it interacts with volumes
 424     created with differenct credentials and to other projects with the same
 425     name
 426     """
 427     def setUp(self):
 428         super(self.__class__, self).__start__()
 429
 430         self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 431         self.volume_creators = list()
 432
 433     def tearDown(self):
 434         for volume_creator in self.volume_creators:
 435             volume_creator.clean()
 436
 437         super(self.__class__, self).__clean__()
 438
 439     # TODO - activate after cinder API bug has been fixed
 440     # see https://bugs.launchpad.net/cinder/+bug/1641982 as to why this test
 441     # is not activated
 442     # def test_create_by_admin_to_other_proj(self):
 443     #     """
 444     #     Creates a volume as admin to the project of os_creds then instantiates
 445     #     a creator object with the os_creds project to ensure it initializes
 446     #     without creation
 447     #     """
 448     #     self.volume_creators.append(OpenStackVolume(
 449     #         self.admin_os_creds, VolumeConfig(
 450     #             name=self.guid + '-vol',
 451     #             project_name=self.os_creds.project_name)))
 452     #     admin_vol = self.volume_creators[0].create(block=True)
 453     #
 454     #     self.volume_creators.append(OpenStackVolume(
 455     #         self.os_creds, VolumeConfig(name=self.guid + '-vol')))
 456     #     proj_vol = self.volume_creators[1].create(block=True)
 457     #
 458     #     self.assertEqual(admin_vol, proj_vol)
 459
 460     def test_create_two_vol_same_name_diff_proj(self):
 461         """
 462         Creates a volume as admin to the project of os_creds then instantiates
 463         a creator object with the os_creds project to ensure it initializes
 464         without creation
 465         """
 466         vol_name = self.guid + '-vol'
 467         self.volume_creators.append(OpenStackVolume(
 468             self.admin_os_creds, VolumeConfig(name=vol_name)))
 469         admin_vol = self.volume_creators[0].create(block=True)
 470         self.assertIsNotNone(admin_vol)
 471
 472         admin_key = keystone_utils.keystone_client(
 473             self.admin_os_creds, self.admin_os_session)
 474         admin_proj = keystone_utils.get_project(
 475             admin_key, project_name=self.admin_os_creds.project_name)
 476         self.assertEqual(admin_vol.project_id, admin_proj.id)
 477
 478         admin_cinder = cinder_utils.cinder_client(
 479             self.admin_os_creds, self.admin_os_session)
 480         admin_vol_get = cinder_utils.get_volume(
 481             admin_cinder, admin_key, volume_name=vol_name,
 482             project_name=self.admin_os_creds.project_name)
 483         self.assertIsNotNone(admin_vol_get)
 484         self.assertEqual(admin_vol, admin_vol_get)
 485
 486         self.volume_creators.append(OpenStackVolume(
 487             self.os_creds, VolumeConfig(name=vol_name)))
 488         proj_vol = self.volume_creators[1].create(block=True)
 489         self.assertIsNotNone(proj_vol)
 490
 491         self.assertNotEqual(admin_vol, proj_vol)
 492
 493         proj_key = keystone_utils.keystone_client(
 494             self.os_creds, self.os_session)
 495         proj_cinder = cinder_utils.cinder_client(
 496             self.os_creds, self.os_session)
 497         proj_vol_get = cinder_utils.get_volume(
 498             proj_cinder, proj_key, volume_name=vol_name,
 499             project_name=self.os_creds.project_name)
 500
 501         self.assertIsNotNone(proj_vol_get)
 502         self.assertEqual(proj_vol, proj_vol_get)