snaps/openstack/tests/create_image_tests.py

   1 # Copyright (c) 2016 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 import os
  16 import shutil
  17 import uuid
  18 import unittest
  19
  20 from snaps import file_utils
  21 from snaps.openstack.create_image import ImageSettings
  22
  23 import openstack_tests
  24 from snaps.openstack.utils import glance_utils, nova_utils
  25 from snaps.openstack import create_image
  26 from snaps.openstack import os_credentials
  27 from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
  28
  29 __author__ = 'spisarski'
  30
  31
  32 class ImageSettingsUnitTests(unittest.TestCase):
  33     """
  34     Tests the construction of the ImageSettings class
  35     """
  36
  37     def test_no_params(self):
  38         with self.assertRaises(Exception):
  39             ImageSettings()
  40
  41     def test_empty_config(self):
  42         with self.assertRaises(Exception):
  43             ImageSettings(config=dict())
  44
  45     def test_name_only(self):
  46         with self.assertRaises(Exception):
  47             ImageSettings(name='foo')
  48
  49     def test_config_with_name_only(self):
  50         with self.assertRaises(Exception):
  51             ImageSettings(config={'name': 'foo'})
  52
  53     def test_name_user_only(self):
  54         with self.assertRaises(Exception):
  55             ImageSettings(name='foo', image_user='bar')
  56
  57     def test_config_with_name_user_only(self):
  58         with self.assertRaises(Exception):
  59             ImageSettings(config={'name': 'foo', 'image_user': 'bar'})
  60
  61     def test_name_user_format_only(self):
  62         with self.assertRaises(Exception):
  63             ImageSettings(name='foo', image_user='bar', img_format='qcow2')
  64
  65     def test_config_with_name_user_format_only(self):
  66         with self.assertRaises(Exception):
  67             ImageSettings(config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2'})
  68
  69     def test_name_user_format_url_file_only(self):
  70         with self.assertRaises(Exception):
  71             ImageSettings(name='foo', image_user='bar', img_format='qcow2', url='http://foo.com',
  72                           image_file='/foo/bar.qcow')
  73
  74     def test_config_with_name_user_format_url_file_only(self):
  75         with self.assertRaises(Exception):
  76             ImageSettings(config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2',
  77                                   'download_url': 'http://foo.com', 'image_file': '/foo/bar.qcow'})
  78
  79     def test_name_user_format_url_only(self):
  80         settings = ImageSettings(name='foo', image_user='bar', img_format='qcow2', url='http://foo.com')
  81         self.assertEquals('foo', settings.name)
  82         self.assertEquals('bar', settings.image_user)
  83         self.assertEquals('qcow2', settings.format)
  84         self.assertEquals('http://foo.com', settings.url)
  85         self.assertIsNone(settings.image_file)
  86         self.assertIsNone(settings.nic_config_pb_loc)
  87
  88     def test_name_user_format_url_only_properties(self):
  89         properties = {}
  90         properties['hw_video_model'] = 'vga'
  91         settings = ImageSettings(name='foo', image_user='bar', img_format='qcow2', url='http://foo.com', extra_properties=properties)
  92         self.assertEquals('foo', settings.name)
  93         self.assertEquals('bar', settings.image_user)
  94         self.assertEquals('qcow2', settings.format)
  95         self.assertEquals('http://foo.com', settings.url)
  96         self.assertEquals(properties, settings.extra_properties)
  97         self.assertIsNone(settings.image_file)
  98         self.assertIsNone(settings.nic_config_pb_loc)
  99
 100
 101     def test_config_with_name_user_format_url_only(self):
 102         settings = ImageSettings(config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2',
 103                                          'download_url': 'http://foo.com'})
 104         self.assertEquals('foo', settings.name)
 105         self.assertEquals('bar', settings.image_user)
 106         self.assertEquals('qcow2', settings.format)
 107         self.assertEquals('http://foo.com', settings.url)
 108         self.assertIsNone(settings.image_file)
 109         self.assertIsNone(settings.nic_config_pb_loc)
 110
 111     def test_name_user_format_file_only(self):
 112         settings = ImageSettings(name='foo', image_user='bar', img_format='qcow2', image_file='/foo/bar.qcow')
 113         self.assertEquals('foo', settings.name)
 114         self.assertEquals('bar', settings.image_user)
 115         self.assertEquals('qcow2', settings.format)
 116         self.assertIsNone(settings.url)
 117         self.assertEquals('/foo/bar.qcow', settings.image_file)
 118         self.assertIsNone(settings.nic_config_pb_loc)
 119
 120     def test_config_with_name_user_format_file_only(self):
 121         settings = ImageSettings(config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2',
 122                                          'image_file': '/foo/bar.qcow'})
 123         self.assertEquals('foo', settings.name)
 124         self.assertEquals('bar', settings.image_user)
 125         self.assertEquals('qcow2', settings.format)
 126         self.assertIsNone(settings.url)
 127         self.assertEquals('/foo/bar.qcow', settings.image_file)
 128         self.assertIsNone(settings.nic_config_pb_loc)
 129
 130     def test_all_url(self):
 131         properties = {}
 132         properties['hw_video_model'] = 'vga'
 133         settings = ImageSettings(name='foo', image_user='bar', img_format='qcow2', url='http://foo.com',
 134                                  extra_properties=properties, nic_config_pb_loc='/foo/bar')
 135         self.assertEquals('foo', settings.name)
 136         self.assertEquals('bar', settings.image_user)
 137         self.assertEquals('qcow2', settings.format)
 138         self.assertEquals('http://foo.com', settings.url)
 139         self.assertEquals(properties, settings.extra_properties)
 140         self.assertIsNone(settings.image_file)
 141         self.assertEquals('/foo/bar', settings.nic_config_pb_loc)
 142
 143     def test_config_all_url(self):
 144         settings = ImageSettings(config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2',
 145                                          'download_url': 'http://foo.com',
 146                                          'extra_properties' : '{\'hw_video_model\': \'vga\'}',
 147                                          'nic_config_pb_loc': '/foo/bar'})
 148         self.assertEquals('foo', settings.name)
 149         self.assertEquals('bar', settings.image_user)
 150         self.assertEquals('qcow2', settings.format)
 151         self.assertEquals('http://foo.com', settings.url)
 152         self.assertEquals('{\'hw_video_model\': \'vga\'}', settings.extra_properties)
 153         self.assertIsNone(settings.image_file)
 154         self.assertEquals('/foo/bar', settings.nic_config_pb_loc)
 155
 156     def test_all_file(self):
 157         properties = {}
 158         properties['hw_video_model'] = 'vga'
 159         settings = ImageSettings(name='foo', image_user='bar', img_format='qcow2', image_file='/foo/bar.qcow',
 160                                  extra_properties=properties, nic_config_pb_loc='/foo/bar')
 161         self.assertEquals('foo', settings.name)
 162         self.assertEquals('bar', settings.image_user)
 163         self.assertEquals('qcow2', settings.format)
 164         self.assertIsNone(settings.url)
 165         self.assertEquals('/foo/bar.qcow', settings.image_file)
 166         self.assertEquals(properties, settings.extra_properties)
 167         self.assertEquals('/foo/bar', settings.nic_config_pb_loc)
 168
 169     def test_config_all_file(self):
 170         settings = ImageSettings(config={'name': 'foo', 'image_user': 'bar', 'format': 'qcow2',
 171                                          'image_file': '/foo/bar.qcow',
 172                                          'extra_properties' : '{\'hw_video_model\' : \'vga\'}',
 173                                          'nic_config_pb_loc': '/foo/bar'})
 174         self.assertEquals('foo', settings.name)
 175         self.assertEquals('bar', settings.image_user)
 176         self.assertEquals('qcow2', settings.format)
 177         self.assertIsNone(settings.url)
 178         self.assertEquals('/foo/bar.qcow', settings.image_file)
 179         self.assertEquals('{\'hw_video_model\' : \'vga\'}', settings.extra_properties)
 180         self.assertEquals('/foo/bar', settings.nic_config_pb_loc)
 181
 182
 183 class CreateImageSuccessTests(OSIntegrationTestCase):
 184     """
 185     Test for the CreateImage class defined in create_image.py
 186     """
 187
 188     def setUp(self):
 189         """
 190         Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
 191         within OpenStack
 192         """
 193         super(self.__class__, self).__start__()
 194
 195         guid = uuid.uuid4()
 196         self.image_name = self.__class__.__name__ + '-' + str(guid)
 197
 198         self.nova = nova_utils.nova_client(self.os_creds)
 199         self.glance = glance_utils.glance_client(self.os_creds)
 200
 201         self.tmp_dir = 'tmp/' + str(guid)
 202         if not os.path.exists(self.tmp_dir):
 203             os.makedirs(self.tmp_dir)
 204
 205     def tearDown(self):
 206         """
 207         Cleans the image and downloaded image file
 208         """
 209         if self.image_creator:
 210             self.image_creator.clean()
 211
 212         if os.path.exists(self.tmp_dir) and os.path.isdir(self.tmp_dir):
 213             shutil.rmtree(self.tmp_dir)
 214
 215         super(self.__class__, self).__clean__()
 216
 217     def test_create_image_clean_url(self):
 218         """
 219         Tests the creation of an OpenStack image from a URL.
 220         """
 221         # Create Image
 222         os_image_settings = openstack_tests.cirros_url_image(name=self.image_name)
 223         self.image_creator = create_image.OpenStackImage(self.os_creds, os_image_settings)
 224
 225         created_image = self.image_creator.create()
 226         self.assertIsNotNone(created_image)
 227
 228         retrieved_image = glance_utils.get_image(self.nova, self.glance, os_image_settings.name)
 229         self.assertIsNotNone(retrieved_image)
 230
 231         self.assertEquals(created_image.name, retrieved_image.name)
 232         self.assertEquals(created_image.id, retrieved_image.id)
 233
 234     def test_create_image_clean_url_properties(self):
 235         """
 236         Tests the creation of an OpenStack image from a URL and set properties.
 237         """
 238         # Set properties
 239         properties = {}
 240         properties['hw_video_model'] = 'vga'
 241
 242         # Create Image
 243         os_image_settings = openstack_tests.cirros_url_image(name=self.image_name)
 244         os_image_settings.extra_properties = properties
 245         self.image_creator = create_image.OpenStackImage(self.os_creds, os_image_settings)
 246
 247         created_image = self.image_creator.create()
 248         self.assertIsNotNone(created_image)
 249
 250         retrieved_image = glance_utils.get_image(self.nova, self.glance, os_image_settings.name)
 251         self.assertIsNotNone(retrieved_image)
 252
 253         self.assertEquals(created_image.name, retrieved_image.name)
 254         self.assertEquals(created_image.id, retrieved_image.id)
 255         self.assertEquals(created_image.properties, retrieved_image.properties)
 256
 257     def test_create_image_clean_file(self):
 258         """
 259         Tests the creation of an OpenStack image from a file.
 260         """
 261         url_image_settings = openstack_tests.cirros_url_image('foo')
 262         image_file = file_utils.download(url_image_settings.url, self.tmp_dir)
 263         file_image_settings = openstack_tests.file_image_test_settings(name=self.image_name, file_path=image_file.name)
 264         self.image_creator = create_image.OpenStackImage(self.os_creds, file_image_settings)
 265
 266         self.image = self.image_creator.create()
 267         self.assertIsNotNone(self.image)
 268         self.assertEqual(self.image_name, self.image.name)
 269
 270         created_image = self.image_creator.create()
 271         self.assertIsNotNone(created_image)
 272
 273         retrieved_image = glance_utils.get_image(self.nova, self.glance, file_image_settings.name)
 274         self.assertIsNotNone(retrieved_image)
 275
 276         self.assertEquals(created_image.name, retrieved_image.name)
 277         self.assertEquals(created_image.id, retrieved_image.id)
 278
 279     def test_create_delete_image(self):
 280         """
 281         Tests the creation then deletion of an OpenStack image to ensure clean() does not raise an Exception.
 282         """
 283         # Create Image
 284         os_image_settings = openstack_tests.cirros_url_image(name=self.image_name)
 285         self.image_creator = create_image.OpenStackImage(self.os_creds, os_image_settings)
 286         created_image = self.image_creator.create()
 287         self.assertIsNotNone(created_image)
 288
 289         # Delete Image manually
 290         glance_utils.delete_image(self.glance, created_image)
 291
 292         self.assertIsNone(glance_utils.get_image(self.nova, self.glance, self.image_creator.image_settings.name))
 293
 294         # Must not throw an exception when attempting to cleanup non-existent image
 295         self.image_creator.clean()
 296         self.assertIsNone(self.image_creator.get_image())
 297
 298     def test_create_same_image(self):
 299         """
 300         Tests the creation of an OpenStack image when the image already exists.
 301         """
 302         # Create Image
 303         os_image_settings = openstack_tests.cirros_url_image(name=self.image_name)
 304         self.image_creator = create_image.OpenStackImage(self.os_creds, os_image_settings)
 305         image1 = self.image_creator.create()
 306         # Should be retrieving the instance data
 307         os_image_2 = create_image.OpenStackImage(self.os_creds, os_image_settings)
 308         image2 = os_image_2.create()
 309         self.assertEquals(image1.id, image2.id)
 310
 311
 312 class CreateImageNegativeTests(OSIntegrationTestCase):
 313     """
 314     Negative test cases for the CreateImage class
 315     """
 316
 317     def setUp(self):
 318         super(self.__class__, self).__start__()
 319
 320         self.image_name = self.__class__.__name__ + '-' + str(uuid.uuid4())
 321         self.image_creator = None
 322
 323     def tearDown(self):
 324         if self.image_creator:
 325             self.image_creator.clean()
 326
 327         super(self.__class__, self).__clean__()
 328
 329     def test_none_image_name(self):
 330         """
 331         Expect an exception when the image name is None
 332         """
 333         os_image_settings = openstack_tests.cirros_url_image(name=self.image_name)
 334         with self.assertRaises(Exception):
 335             self.image_creator = create_image.OpenStackImage(
 336                 self.os_creds, create_image.ImageSettings(
 337                     name=None, image_user=os_image_settings.image_user, img_format=os_image_settings.format,
 338                     url=os_image_settings.url))
 339
 340             self.fail('Exception should have been thrown prior to this line')
 341
 342     def test_bad_image_url(self):
 343         """
 344         Expect an exception when the image download url is bad
 345         """
 346         os_image_settings = openstack_tests.cirros_url_image(name=self.image_name)
 347         self.image_creator = create_image.OpenStackImage(self.os_creds, create_image.ImageSettings(
 348             name=os_image_settings.name, image_user=os_image_settings.image_user,
 349             img_format=os_image_settings.format, url="http://foo.bar"))
 350         with self.assertRaises(Exception):
 351             self.image_creator.create()
 352
 353     def test_bad_image_file(self):
 354         """
 355         Expect an exception when the image file does not exist
 356         """
 357         os_image_settings = openstack_tests.cirros_url_image(name=self.image_name)
 358         self.image_creator = create_image.OpenStackImage(
 359             self.os_creds,
 360             create_image.ImageSettings(name=os_image_settings.name, image_user=os_image_settings.image_user,
 361                                        img_format=os_image_settings.format, image_file="/foo/bar.qcow"))
 362         with self.assertRaises(Exception):
 363             self.image_creator.create()
 364
 365     def test_none_proj_name(self):
 366         """
 367         Expect an exception when the project name is None
 368         """
 369         os_image_settings = openstack_tests.cirros_url_image(name=self.image_name)
 370         with self.assertRaises(Exception):
 371             self.image_creator = create_image.OpenStackImage(
 372                 os_credentials.OSCreds(self.os_creds.username, self.os_creds.password, self.os_creds.auth_url, None,
 373                                        proxy_settings=self.os_creds.proxy_settings),
 374                 os_image_settings)
 375             self.image_creator.create()
 376
 377     def test_none_auth_url(self):
 378         """
 379         Expect an exception when the project name is None
 380         """
 381         os_image_settings = openstack_tests.cirros_url_image(name=self.image_name)
 382         with self.assertRaises(Exception):
 383             self.image_creator = create_image.OpenStackImage(
 384                 os_credentials.OSCreds(self.os_creds.username, self.os_creds.password, None,
 385                                        self.os_creds.project_name, proxy_settings=self.os_creds.proxy_settings),
 386                 os_image_settings)
 387             self.image_creator.create()
 388
 389     def test_none_password(self):
 390         """
 391         Expect an exception when the project name is None
 392         """
 393         os_image_settings = openstack_tests.cirros_url_image(name=self.image_name)
 394         with self.assertRaises(Exception):
 395             self.image_creator = create_image.OpenStackImage(
 396                 os_credentials.OSCreds(self.os_creds.username, None, self.os_creds.os_auth_url,
 397                                        self.os_creds.project_name, proxy_settings=self.os_creds.proxy_settings),
 398                 os_image_settings)
 399
 400     def test_none_user(self):
 401         """
 402         Expect an exception when the project name is None
 403         """
 404         os_image_settings = openstack_tests.cirros_url_image(name=self.image_name)
 405         with self.assertRaises(Exception):
 406             self.image_creator = create_image.OpenStackImage(
 407                 os_credentials.OSCreds(None, self.os_creds.password, self.os_creds.os_auth_url,
 408                                        self.os_creds.project_name,
 409                                        proxy_settings=self.os_creds.proxy_settings),
 410                 os_image_settings)
 411
 412
 413 class CreateMultiPartImageTests(OSIntegrationTestCase):
 414     """
 415     Test for creating a 3-part image
 416     """
 417     def setUp(self):
 418         """
 419         Instantiates the CreateImage object that is responsible for
 420         downloading and creating an OS image file within OpenStack
 421         """
 422         super(self.__class__, self).__start__()
 423
 424         guid = uuid.uuid4()
 425         self.image_creators = list()
 426         self.image_name = self.__class__.__name__ + '-' + str(guid)
 427
 428         self.nova = nova_utils.nova_client(self.os_creds)
 429         self.glance = glance_utils.glance_client(self.os_creds)
 430
 431         self.tmp_dir = 'tmp/' + str(guid)
 432         if not os.path.exists(self.tmp_dir):
 433             os.makedirs(self.tmp_dir)
 434
 435     def tearDown(self):
 436         """
 437         Cleans the images and downloaded image file
 438         """
 439         while self.image_creators:
 440             self.image_creators[0].clean()
 441             self.image_creators.pop(0)
 442
 443         if os.path.exists(self.tmp_dir) and os.path.isdir(self.tmp_dir):
 444             shutil.rmtree(self.tmp_dir)
 445
 446         super(self.__class__, self).__clean__()
 447
 448     def test_create_three_part_image_from_url(self):
 449         """
 450         Tests the creation of a 3-part OpenStack image from a URL.
 451         """
 452         # Set properties
 453         properties = {}
 454
 455         # Create the kernel image
 456         kernel_image_settings = openstack_tests.cirros_url_image(name=self.image_name+'_kernel',
 457             url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel')
 458         self.image_creators.append(create_image.OpenStackImage(self.os_creds, kernel_image_settings))
 459         kernel_image = self.image_creators[-1].create()
 460         self.assertIsNotNone(kernel_image)
 461
 462         # Create the ramdisk image
 463         ramdisk_image_settings = openstack_tests.cirros_url_image(name=self.image_name+'_ramdisk',
 464             url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs')
 465         self.image_creators.append(create_image.OpenStackImage(self.os_creds, ramdisk_image_settings))
 466         ramdisk_image = self.image_creators[-1].create()
 467         self.assertIsNotNone(ramdisk_image)
 468
 469         # Create the main image
 470         os_image_settings = openstack_tests.cirros_url_image(name=self.image_name,
 471             url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img')
 472         properties['kernel_id'] = kernel_image.id
 473         properties['ramdisk_id'] = ramdisk_image.id
 474         os_image_settings.extra_properties = properties
 475         self.image_creators.append(create_image.OpenStackImage(self.os_creds, os_image_settings))
 476         created_image = self.image_creators[-1].create()
 477         self.assertIsNotNone(created_image)
 478         self.assertEqual(self.image_name, created_image.name)
 479
 480         retrieved_image = glance_utils.get_image(self.nova, self.glance, os_image_settings.name)
 481         self.assertIsNotNone(retrieved_image)
 482
 483         self.assertEquals(created_image.name, retrieved_image.name)
 484         self.assertEquals(created_image.id, retrieved_image.id)
 485         self.assertEquals(created_image.properties, retrieved_image.properties)
 486
 487     def test_create_three_part_image_from_file(self):
 488         """
 489         Tests the creation of a 3-part OpenStack image from files.
 490         """
 491         # Set properties
 492         properties = {}
 493
 494         # Create the kernel image
 495         url_image_settings = openstack_tests.cirros_url_image('foo_kernel',
 496             url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel')
 497         kernel_image_file = file_utils.download(url_image_settings.url, self.tmp_dir)
 498         kernel_file_image_settings = openstack_tests.file_image_test_settings(
 499             name=self.image_name+'_kernel', file_path=kernel_image_file.name)
 500         self.image_creators.append(create_image.OpenStackImage(self.os_creds, kernel_file_image_settings))
 501         kernel_image = self.image_creators[-1].create()
 502         self.assertIsNotNone(kernel_image)
 503
 504         # Create the ramdisk image
 505         url_image_settings = openstack_tests.cirros_url_image('foo_ramdisk',
 506             url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs')
 507         ramdisk_image_file = file_utils.download(url_image_settings.url, self.tmp_dir)
 508         ramdisk_file_image_settings = openstack_tests.file_image_test_settings(
 509             name=self.image_name+'_ramdisk', file_path=ramdisk_image_file.name)
 510         self.image_creators.append(create_image.OpenStackImage(self.os_creds, ramdisk_file_image_settings))
 511         ramdisk_image = self.image_creators[-1].create()
 512         self.assertIsNotNone(ramdisk_image)
 513
 514         # Create the main image
 515         url_image_settings = openstack_tests.cirros_url_image('foo',
 516             url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img')
 517         image_file = file_utils.download(url_image_settings.url, self.tmp_dir)
 518         file_image_settings = openstack_tests.file_image_test_settings(name=self.image_name, file_path=image_file.name)
 519         properties['kernel_id'] = kernel_image.id
 520         properties['ramdisk_id'] = ramdisk_image.id
 521         file_image_settings.extra_properties = properties
 522         self.image_creators.append(create_image.OpenStackImage(self.os_creds, file_image_settings))
 523         created_image = self.image_creators[-1].create()
 524
 525         self.assertIsNotNone(created_image)
 526         self.assertEqual(self.image_name, created_image.name)
 527
 528         retrieved_image = glance_utils.get_image(self.nova, self.glance, file_image_settings.name)
 529         self.assertIsNotNone(retrieved_image)
 530
 531         self.assertEquals(created_image.name, retrieved_image.name)
 532         self.assertEquals(created_image.id, retrieved_image.id)
 533         self.assertEquals(created_image.properties, retrieved_image.properties)