snaps/openstack/utils/tests/settings_utils_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 import logging
  16 import unittest
  17
  18 import os
  19 import uuid
  20
  21 from snaps.config.flavor import FlavorConfig
  22 from snaps.domain.flavor import Flavor
  23 from snaps.domain.volume import (
  24     Volume, VolumeType, VolumeTypeEncryption, QoSSpec)
  25 from snaps.openstack import (
  26     create_image, create_network, create_router, create_flavor,
  27     create_keypairs, create_instance)
  28 from snaps.openstack.create_network import (
  29     NetworkSettings, OpenStackNetwork, SubnetSettings)
  30 from snaps.openstack.create_qos import Consumer
  31 from snaps.openstack.create_security_group import (
  32     SecurityGroupRuleSettings,  Direction, Protocol, OpenStackSecurityGroup,
  33     SecurityGroupSettings)
  34 from snaps.openstack.tests import openstack_tests
  35 from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
  36 from snaps.openstack.utils import (
  37     neutron_utils, settings_utils, nova_utils, glance_utils)
  38
  39 __author__ = 'spisarski'
  40
  41 logger = logging.getLogger('nova_utils_tests')
  42
  43
  44 class SettingsUtilsNetworkingTests(OSComponentTestCase):
  45     """
  46     Tests the ability to reverse engineer NetworkSettings objects from existing
  47     networks deployed to OpenStack
  48     """
  49
  50     def setUp(self):
  51         """
  52         Instantiates OpenStack instances that cannot be spawned by Heat
  53         """
  54         guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
  55         self.network_name = guid + '-net'
  56         self.subnet_name = guid + '-subnet'
  57         self.net_creator = None
  58         self.neutron = neutron_utils.neutron_client(self.os_creds)
  59
  60     def tearDown(self):
  61         """
  62         Cleans the image and downloaded image file
  63         """
  64         if self.net_creator:
  65             try:
  66                 self.net_creator.clean()
  67             except:
  68                 pass
  69
  70     def test_derive_net_settings_no_subnet(self):
  71         """
  72         Validates the utility function settings_utils#create_network_settings
  73         returns an acceptable NetworkSettings object and ensures that the
  74         new settings object will not cause the new OpenStackNetwork instance
  75         to create another network
  76         """
  77         net_settings = NetworkSettings(name=self.network_name)
  78         self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
  79         network = self.net_creator.create()
  80
  81         derived_settings = settings_utils.create_network_settings(
  82             self.neutron, network)
  83
  84         self.assertIsNotNone(derived_settings)
  85         self.assertEqual(net_settings.name, derived_settings.name)
  86         self.assertEqual(net_settings.admin_state_up,
  87                          derived_settings.admin_state_up)
  88         self.assertEqual(net_settings.external, derived_settings.external)
  89         self.assertEqual(len(net_settings.subnet_settings),
  90                          len(derived_settings.subnet_settings))
  91
  92         net_creator = OpenStackNetwork(self.os_creds, derived_settings)
  93         derived_network = net_creator.create()
  94
  95         self.assertEqual(network, derived_network)
  96
  97     def test_derive_net_settings_two_subnets(self):
  98         """
  99         Validates the utility function settings_utils#create_network_settings
 100         returns an acceptable NetworkSettings object
 101         """
 102         subnet_settings = list()
 103         subnet_settings.append(SubnetSettings(name='sub1', cidr='10.0.0.0/24'))
 104         subnet_settings.append(SubnetSettings(name='sub2', cidr='10.0.1.0/24'))
 105         net_settings = NetworkSettings(name=self.network_name,
 106                                        subnet_settings=subnet_settings)
 107         self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
 108         network = self.net_creator.create()
 109
 110         derived_settings = settings_utils.create_network_settings(
 111             self.neutron, network)
 112
 113         self.assertIsNotNone(derived_settings)
 114         self.assertEqual(net_settings.name, derived_settings.name)
 115         self.assertEqual(net_settings.admin_state_up,
 116                          derived_settings.admin_state_up)
 117         self.assertEqual(net_settings.external, derived_settings.external)
 118         self.assertEqual(len(net_settings.subnet_settings),
 119                          len(derived_settings.subnet_settings))
 120
 121         # Validate the first subnet
 122         orig_sub1 = net_settings.subnet_settings[0]
 123         found = False
 124         for derived_sub in derived_settings.subnet_settings:
 125             if orig_sub1.name == derived_sub.name:
 126                 self.assertEqual(orig_sub1.cidr, derived_sub.cidr)
 127                 found = True
 128
 129         self.assertTrue(found)
 130
 131         # Validate the second subnet
 132         orig_sub2 = net_settings.subnet_settings[1]
 133         found = False
 134         for derived_sub in derived_settings.subnet_settings:
 135             if orig_sub2.name == derived_sub.name:
 136                 self.assertEqual(orig_sub2.cidr, derived_sub.cidr)
 137                 self.assertEqual(orig_sub2.ip_version, derived_sub.ip_version)
 138                 found = True
 139
 140         self.assertTrue(found)
 141
 142
 143 class SettingsUtilsVmInstTests(OSComponentTestCase):
 144     """
 145     Tests the ability to reverse engineer VmInstanceSettings objects from
 146     existing VMs/servers deployed to OpenStack
 147     """
 148
 149     def setUp(self):
 150         """
 151         Instantiates the CreateImage object that is responsible for downloading
 152         and creating an OS image file within OpenStack
 153         """
 154         self.nova = nova_utils.nova_client(self.os_creds)
 155         self.glance = glance_utils.glance_client(self.os_creds)
 156         self.neutron = neutron_utils.neutron_client(self.os_creds)
 157
 158         guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 159         self.keypair_priv_filepath = 'tmp/' + guid
 160         self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub'
 161         self.keypair_name = guid + '-kp'
 162         self.vm_inst_name = guid + '-inst'
 163         self.test_file_local_path = 'tmp/' + guid + '-hello.txt'
 164         self.port_1_name = guid + '-port-1'
 165         self.port_2_name = guid + '-port-2'
 166         self.floating_ip_name = guid + 'fip1'
 167
 168         # Setup members to cleanup just in case they don't get created
 169         self.inst_creator = None
 170         self.keypair_creator = None
 171         self.sec_grp_creator = None
 172         self.flavor_creator = None
 173         self.router_creator = None
 174         self.network_creator = None
 175         self.image_creator = None
 176
 177         try:
 178             # Create Image
 179             os_image_settings = openstack_tests.cirros_image_settings(
 180                 name=guid + '-' + '-image',
 181                 image_metadata=self.image_metadata)
 182             self.image_creator = create_image.OpenStackImage(self.os_creds,
 183                                                              os_image_settings)
 184             self.image_creator.create()
 185
 186             # First network is public
 187             self.pub_net_config = openstack_tests.get_pub_net_config(
 188                 net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
 189                 router_name=guid + '-pub-router',
 190                 external_net=self.ext_net_name)
 191
 192             self.network_creator = create_network.OpenStackNetwork(
 193                 self.os_creds, self.pub_net_config.network_settings)
 194             self.network_creator.create()
 195
 196             # Create routers
 197             self.router_creator = create_router.OpenStackRouter(
 198                 self.os_creds, self.pub_net_config.router_settings)
 199             self.router_creator.create()
 200
 201             # Create Flavor
 202             self.flavor_creator = create_flavor.OpenStackFlavor(
 203                 self.os_creds,
 204                 FlavorConfig(
 205                     name=guid + '-flavor-name', ram=256, disk=1, vcpus=1))
 206             self.flavor_creator.create()
 207
 208             # Create Key/Pair
 209             self.keypair_creator = create_keypairs.OpenStackKeypair(
 210                 self.os_creds, create_keypairs.KeypairSettings(
 211                     name=self.keypair_name,
 212                     public_filepath=self.keypair_pub_filepath,
 213                     private_filepath=self.keypair_priv_filepath))
 214             self.keypair_creator.create()
 215
 216             # Create Security Group
 217             sec_grp_name = guid + '-sec-grp'
 218             rule1 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
 219                                               direction=Direction.ingress,
 220                                               protocol=Protocol.icmp)
 221             rule2 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
 222                                               direction=Direction.ingress,
 223                                               protocol=Protocol.tcp,
 224                                               port_range_min=22,
 225                                               port_range_max=22)
 226             self.sec_grp_creator = OpenStackSecurityGroup(
 227                 self.os_creds,
 228                 SecurityGroupSettings(name=sec_grp_name,
 229                                       rule_settings=[rule1, rule2]))
 230             self.sec_grp_creator.create()
 231
 232             # Create instance
 233             ports_settings = list()
 234             ports_settings.append(
 235                 create_network.PortSettings(
 236                     name=self.port_1_name,
 237                     network_name=self.pub_net_config.network_settings.name))
 238
 239             instance_settings = create_instance.VmInstanceSettings(
 240                 name=self.vm_inst_name,
 241                 flavor=self.flavor_creator.flavor_settings.name,
 242                 port_settings=ports_settings,
 243                 floating_ip_settings=[create_instance.FloatingIpSettings(
 244                     name=self.floating_ip_name, port_name=self.port_1_name,
 245                     router_name=self.pub_net_config.router_settings.name)])
 246
 247             self.inst_creator = create_instance.OpenStackVmInstance(
 248                 self.os_creds, instance_settings,
 249                 self.image_creator.image_settings,
 250                 keypair_settings=self.keypair_creator.keypair_settings)
 251         except:
 252             self.tearDown()
 253             raise
 254
 255     def tearDown(self):
 256         """
 257         Cleans the created objects
 258         """
 259         if self.inst_creator:
 260             try:
 261                 self.inst_creator.clean()
 262             except:
 263                 pass
 264
 265         if self.sec_grp_creator:
 266             try:
 267                 self.sec_grp_creator.clean()
 268             except:
 269                 pass
 270
 271         if self.keypair_creator:
 272             try:
 273                 self.keypair_creator.clean()
 274             except:
 275                 pass
 276
 277         if self.flavor_creator:
 278             try:
 279                 self.flavor_creator.clean()
 280             except:
 281                 pass
 282
 283         if os.path.isfile(self.keypair_pub_filepath):
 284             try:
 285                 os.remove(self.keypair_pub_filepath)
 286             except:
 287                 pass
 288
 289         if os.path.isfile(self.keypair_priv_filepath):
 290             try:
 291                 os.remove(self.keypair_priv_filepath)
 292             except:
 293                 pass
 294
 295         if self.router_creator:
 296             try:
 297                 self.router_creator.clean()
 298             except:
 299                 pass
 300
 301         if self.network_creator:
 302             try:
 303                 self.network_creator.clean()
 304             except:
 305                 pass
 306
 307         if self.image_creator and not self.image_creator.image_settings.exists:
 308             try:
 309                 self.image_creator.clean()
 310             except:
 311                 pass
 312
 313         if os.path.isfile(self.test_file_local_path):
 314             os.remove(self.test_file_local_path)
 315
 316         # super(self.__class__, self).__clean__()
 317
 318     def test_derive_vm_inst_settings(self):
 319         """
 320         Validates the utility function settings_utils#create_vm_inst_settings
 321         returns an acceptable VmInstanceSettings object
 322         """
 323         self.inst_creator.create(block=True)
 324
 325         server = nova_utils.get_server(
 326             self.nova, vm_inst_settings=self.inst_creator.instance_settings)
 327         derived_vm_settings = settings_utils.create_vm_inst_settings(
 328             self.nova, self.neutron, server)
 329         self.assertIsNotNone(derived_vm_settings)
 330         self.assertIsNotNone(derived_vm_settings.port_settings)
 331         self.assertIsNotNone(derived_vm_settings.floating_ip_settings)
 332
 333     def test_derive_image_settings(self):
 334         """
 335         Validates the utility function settings_utils#create_image_settings
 336         returns an acceptable ImageConfig object
 337         """
 338         self.inst_creator.create(block=True)
 339
 340         server = nova_utils.get_server(
 341             self.nova, vm_inst_settings=self.inst_creator.instance_settings)
 342         derived_image_settings = settings_utils.determine_image_config(
 343             self.glance, server, [self.image_creator.image_settings])
 344         self.assertIsNotNone(derived_image_settings)
 345         self.assertEqual(self.image_creator.image_settings.name,
 346                          derived_image_settings.name)
 347
 348
 349 class SettingsUtilsUnitTests(unittest.TestCase):
 350     """
 351     Exercises the settings_utils.py functions around volumes
 352     """
 353
 354     def test_vol_settings_from_vol(self):
 355         volume = Volume(
 356             name='vol-name', volume_id='vol-id', description='desc', size=99,
 357             vol_type='vol-type', availability_zone='zone1', multi_attach=True)
 358         settings = settings_utils.create_volume_settings(volume)
 359         self.assertEqual(volume.name, settings.name)
 360         self.assertEqual(volume.description, settings.description)
 361         self.assertEqual(volume.size, settings.size)
 362         self.assertEqual(volume.type, settings.type_name)
 363         self.assertEqual(volume.availability_zone, settings.availability_zone)
 364         self.assertEqual(volume.multi_attach, settings.multi_attach)
 365
 366     def test_vol_type_settings_from_vol(self):
 367         encryption = VolumeTypeEncryption(
 368             volume_encryption_id='vol-encrypt-id', volume_type_id='vol-typ-id',
 369             control_location='front-end', provider='FooClass', cipher='1',
 370             key_size=1)
 371         qos_spec = QoSSpec(name='qos-spec-name', spec_id='qos-spec-id',
 372                            consumer=Consumer.back_end)
 373         volume_type = VolumeType(
 374             name='vol-type-name', volume_type_id='vol-type-id', public=True,
 375             encryption=encryption, qos_spec=qos_spec)
 376
 377         settings = settings_utils.create_volume_type_settings(volume_type)
 378         self.assertEqual(volume_type.name, settings.name)
 379         self.assertEqual(volume_type.public, settings.public)
 380
 381         encrypt_settings = settings.encryption
 382         self.assertIsNotNone(encrypt_settings)
 383         self.assertEqual(encryption.control_location,
 384                          encrypt_settings.control_location.value)
 385         self.assertEqual(encryption.cipher, encrypt_settings.cipher)
 386         self.assertEqual(encryption.key_size, encrypt_settings.key_size)
 387
 388         self.assertEqual(qos_spec.name, settings.qos_spec_name)
 389
 390     def test_flavor_settings_from_flavor(self):
 391         flavor = Flavor(
 392             name='flavor-name', flavor_id='flavor-id', ram=99, disk=101,
 393             vcpus=9, ephemeral=3, swap=5, rxtx_factor=7, is_public=False)
 394         settings = settings_utils.create_flavor_config(flavor)
 395         self.assertEqual(flavor.name, settings.name)
 396         self.assertEqual(flavor.id, settings.flavor_id)
 397         self.assertEqual(flavor.ram, settings.ram)
 398         self.assertEqual(flavor.disk, settings.disk)
 399         self.assertEqual(flavor.vcpus, settings.vcpus)
 400         self.assertEqual(flavor.ephemeral, settings.ephemeral)
 401         self.assertEqual(flavor.swap, settings.swap)
 402         self.assertEqual(flavor.rxtx_factor, settings.rxtx_factor)
 403         self.assertEqual(flavor.is_public, settings.is_public)