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
 286 class CreateVolumeWithTypeTests(OSIntegrationTestCase):
 287     """
 288     Test cases for the CreateVolume when attempting to associate it to a
 289     Volume Type
 290     """
 291
 292     def setUp(self):
 293         super(self.__class__, self).__start__()
 294
 295         guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 296         self.volume_name = guid + '-vol'
 297         self.volume_type_name = guid + '-vol-type'
 298
 299         self.volume_type_creator = OpenStackVolumeType(
 300             self.admin_os_creds, VolumeTypeConfig(name=self.volume_type_name))
 301         self.volume_type_creator.create()
 302         self.volume_creator = None
 303
 304     def tearDown(self):
 305         if self.volume_creator:
 306             self.volume_creator.clean()
 307         if self.volume_type_creator:
 308             self.volume_type_creator.clean()
 309
 310         super(self.__class__, self).__clean__()
 311
 312     def test_bad_volume_type(self):
 313         """
 314         Expect a NotFound to be raised when the volume type does not exist
 315         """
 316         self.volume_creator = OpenStackVolume(
 317             self.os_creds,
 318             VolumeConfig(name=self.volume_name, type_name='foo'))
 319
 320         with self.assertRaises(NotFound):
 321             self.volume_creator.create()
 322
 323     def test_valid_volume_type(self):
 324         """
 325         Expect a NotFound to be raised when the volume type does not exist
 326         """
 327         self.volume_creator = OpenStackVolume(
 328             self.admin_os_creds, VolumeConfig(
 329                 name=self.volume_name, type_name=self.volume_type_name))
 330
 331         created_volume = self.volume_creator.create(block=True)
 332         self.assertIsNotNone(created_volume)
 333         self.assertEqual(self.volume_type_name, created_volume.type)
 334
 335
 336 class CreateVolumeWithImageTests(OSIntegrationTestCase):
 337     """
 338     Test cases for the CreateVolume when attempting to associate it to an Image
 339     """
 340
 341     def setUp(self):
 342         super(self.__class__, self).__start__()
 343
 344         self.cinder = cinder_utils.cinder_client(
 345             self.os_creds, self.os_session)
 346
 347         guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 348         self.volume_name = guid + '-vol'
 349         self.image_name = guid + '-image'
 350
 351         os_image_settings = openstack_tests.cirros_image_settings(
 352             name=self.image_name, image_metadata=self.image_metadata)
 353         # Create Image
 354         self.image_creator = OpenStackImage(self.os_creds,
 355                                             os_image_settings)
 356         self.image_creator.create()
 357         self.volume_creator = None
 358
 359     def tearDown(self):
 360         if self.volume_creator:
 361             try:
 362                 self.volume_creator.clean()
 363             except:
 364                 pass
 365         if self.image_creator:
 366             try:
 367                 self.image_creator.clean()
 368             except:
 369                 pass
 370
 371         super(self.__class__, self).__clean__()
 372
 373     def test_bad_image_name(self):
 374         """
 375         Tests OpenStackVolume#create() method to ensure a volume is NOT created
 376         when associating it to an invalid image name
 377         """
 378         self.volume_creator = OpenStackVolume(
 379             self.os_creds,
 380             VolumeConfig(name=self.volume_name, image_name='foo'))
 381
 382         with self.assertRaises(BadRequest):
 383             self.volume_creator.create(block=True)
 384
 385     def test_valid_volume_image(self):
 386         """
 387         Tests OpenStackVolume#create() method to ensure a volume is NOT created
 388         when associating it to an invalid image name
 389         """
 390         self.volume_creator = OpenStackVolume(
 391             self.os_creds,
 392             VolumeConfig(name=self.volume_name, image_name=self.image_name))
 393
 394         created_volume = self.volume_creator.create(block=True)
 395         self.assertIsNotNone(created_volume)
 396         self.assertEqual(
 397             self.volume_creator.volume_settings.name, created_volume.name)
 398         self.assertTrue(self.volume_creator.volume_active())
 399
 400         retrieved_volume = cinder_utils.get_volume_by_id(
 401             self.cinder, created_volume.id)
 402
 403         self.assertEqual(created_volume, retrieved_volume)
 404
 405
 406 class CreateVolMultipleCredsTests(OSIntegrationTestCase):
 407     """
 408     Test for the OpenStackVolume class and how it interacts with volumes
 409     created with differenct credentials and to other projects with the same
 410     name
 411     """
 412     def setUp(self):
 413         super(self.__class__, self).__start__()
 414
 415         self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 416         self.volume_creators = list()
 417
 418     def tearDown(self):
 419         for volume_creator in self.volume_creators:
 420             volume_creator.clean()
 421
 422         super(self.__class__, self).__clean__()
 423
 424     # TODO - activate after cinder API bug has been fixed
 425     # see https://bugs.launchpad.net/cinder/+bug/1641982 as to why this test
 426     # is not activated
 427     # def test_create_by_admin_to_other_proj(self):
 428     #     """
 429     #     Creates a volume as admin to the project of os_creds then instantiates
 430     #     a creator object with the os_creds project to ensure it initializes
 431     #     without creation
 432     #     """
 433     #     self.volume_creators.append(OpenStackVolume(
 434     #         self.admin_os_creds, VolumeConfig(
 435     #             name=self.guid + '-vol',
 436     #             project_name=self.os_creds.project_name)))
 437     #     admin_vol = self.volume_creators[0].create(block=True)
 438     #
 439     #     self.volume_creators.append(OpenStackVolume(
 440     #         self.os_creds, VolumeConfig(name=self.guid + '-vol')))
 441     #     proj_vol = self.volume_creators[1].create(block=True)
 442     #
 443     #     self.assertEqual(admin_vol, proj_vol)
 444
 445     def test_create_two_vol_same_name_diff_proj(self):
 446         """
 447         Creates a volume as admin to the project of os_creds then instantiates
 448         a creator object with the os_creds project to ensure it initializes
 449         without creation
 450         """
 451         vol_name = self.guid + '-vol'
 452         self.volume_creators.append(OpenStackVolume(
 453             self.admin_os_creds, VolumeConfig(name=vol_name)))
 454         admin_vol = self.volume_creators[0].create(block=True)
 455         self.assertIsNotNone(admin_vol)
 456
 457         admin_key = keystone_utils.keystone_client(
 458             self.admin_os_creds, self.admin_os_session)
 459         admin_proj = keystone_utils.get_project(
 460             admin_key, project_name=self.admin_os_creds.project_name)
 461         self.assertEqual(admin_vol.project_id, admin_proj.id)
 462
 463         admin_cinder = cinder_utils.cinder_client(
 464             self.admin_os_creds, self.admin_os_session)
 465         admin_vol_get = cinder_utils.get_volume(
 466             admin_cinder, admin_key, volume_name=vol_name,
 467             project_name=self.admin_os_creds.project_name)
 468         self.assertIsNotNone(admin_vol_get)
 469         self.assertEqual(admin_vol, admin_vol_get)
 470
 471         self.volume_creators.append(OpenStackVolume(
 472             self.os_creds, VolumeConfig(name=vol_name)))
 473         proj_vol = self.volume_creators[1].create(block=True)
 474         self.assertIsNotNone(proj_vol)
 475
 476         self.assertNotEqual(admin_vol, proj_vol)
 477
 478         proj_key = keystone_utils.keystone_client(
 479             self.os_creds, self.os_session)
 480         proj_cinder = cinder_utils.cinder_client(
 481             self.os_creds, self.os_session)
 482         proj_vol_get = cinder_utils.get_volume(
 483             proj_cinder, proj_key, volume_name=vol_name,
 484             project_name=self.os_creds.project_name)
 485
 486         self.assertIsNotNone(proj_vol_get)
 487         self.assertEqual(proj_vol, proj_vol_get)