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.security_group import (
  26     SecurityGroupRuleConfig, Direction, Protocol, SecurityGroupConfig)
  27 from snaps.config.vm_inst import VmInstanceConfig, FloatingIpConfig
  28 from snaps.domain.flavor import Flavor
  29 from snaps.domain.volume import (
  30     Volume, VolumeType, VolumeTypeEncryption, QoSSpec)
  31 from snaps.openstack import (
  32     create_image, create_network, create_router, create_flavor,
  33     create_keypairs, create_instance)
  34 from snaps.openstack.create_qos import Consumer
  35 from snaps.openstack.create_network import OpenStackNetwork
  36 from snaps.openstack.create_security_group import OpenStackSecurityGroup
  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 = SecurityGroupRuleConfig(
 222                 sec_grp_name=sec_grp_name, direction=Direction.ingress,
 223                 protocol=Protocol.icmp)
 224             rule2 = SecurityGroupRuleConfig(
 225                 sec_grp_name=sec_grp_name, direction=Direction.ingress,
 226                 protocol=Protocol.tcp, port_range_min=22, port_range_max=22)
 227             self.sec_grp_creator = OpenStackSecurityGroup(
 228                 self.os_creds,
 229                 SecurityGroupConfig(
 230                     name=sec_grp_name, rule_settings=[rule1, rule2]))
 231             self.sec_grp_creator.create()
 232
 233             # Create instance
 234             ports_settings = list()
 235             ports_settings.append(
 236                 PortConfig(
 237                     name=self.port_1_name,
 238                     network_name=self.pub_net_config.network_settings.name))
 239
 240             instance_settings = VmInstanceConfig(
 241                 name=self.vm_inst_name,
 242                 flavor=self.flavor_creator.flavor_settings.name,
 243                 port_settings=ports_settings,
 244                 floating_ip_settings=[FloatingIpConfig(
 245                     name=self.floating_ip_name, port_name=self.port_1_name,
 246                     router_name=self.pub_net_config.router_settings.name)])
 247
 248             self.inst_creator = create_instance.OpenStackVmInstance(
 249                 self.os_creds, instance_settings,
 250                 self.image_creator.image_settings,
 251                 keypair_settings=self.keypair_creator.keypair_settings)
 252         except:
 253             self.tearDown()
 254             raise
 255
 256     def tearDown(self):
 257         """
 258         Cleans the created objects
 259         """
 260         if self.inst_creator:
 261             try:
 262                 self.inst_creator.clean()
 263             except:
 264                 pass
 265
 266         if self.sec_grp_creator:
 267             try:
 268                 self.sec_grp_creator.clean()
 269             except:
 270                 pass
 271
 272         if self.keypair_creator:
 273             try:
 274                 self.keypair_creator.clean()
 275             except:
 276                 pass
 277
 278         if self.flavor_creator:
 279             try:
 280                 self.flavor_creator.clean()
 281             except:
 282                 pass
 283
 284         if os.path.isfile(self.keypair_pub_filepath):
 285             try:
 286                 os.remove(self.keypair_pub_filepath)
 287             except:
 288                 pass
 289
 290         if os.path.isfile(self.keypair_priv_filepath):
 291             try:
 292                 os.remove(self.keypair_priv_filepath)
 293             except:
 294                 pass
 295
 296         if self.router_creator:
 297             try:
 298                 self.router_creator.clean()
 299             except:
 300                 pass
 301
 302         if self.network_creator:
 303             try:
 304                 self.network_creator.clean()
 305             except:
 306                 pass
 307
 308         if self.image_creator and not self.image_creator.image_settings.exists:
 309             try:
 310                 self.image_creator.clean()
 311             except:
 312                 pass
 313
 314         if os.path.isfile(self.test_file_local_path):
 315             os.remove(self.test_file_local_path)
 316
 317         # super(self.__class__, self).__clean__()
 318
 319     def test_derive_vm_inst_config(self):
 320         """
 321         Validates the utility function settings_utils#create_vm_inst_config
 322         returns an acceptable VmInstanceConfig object
 323         """
 324         self.inst_creator.create(block=True)
 325
 326         server = nova_utils.get_server(
 327             self.nova, self.neutron,
 328             vm_inst_settings=self.inst_creator.instance_settings)
 329         derived_vm_settings = settings_utils.create_vm_inst_config(
 330             self.nova, self.neutron, server, self.project_id)
 331         self.assertIsNotNone(derived_vm_settings)
 332         self.assertIsNotNone(derived_vm_settings.port_settings)
 333         self.assertIsNotNone(derived_vm_settings.floating_ip_settings)
 334
 335     def test_derive_image_settings(self):
 336         """
 337         Validates the utility function settings_utils#create_image_settings
 338         returns an acceptable ImageConfig object
 339         """
 340         self.inst_creator.create(block=True)
 341
 342         server = nova_utils.get_server(
 343             self.nova, self.neutron,
 344             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)