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