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, keystone_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.keystone = keystone_utils.keystone_client(self.os_creds)
 159         self.glance = glance_utils.glance_client(self.os_creds)
 160         self.neutron = neutron_utils.neutron_client(self.os_creds)
 161
 162         guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 163         self.keypair_priv_filepath = 'tmp/' + guid
 164         self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub'
 165         self.keypair_name = guid + '-kp'
 166         self.vm_inst_name = guid + '-inst'
 167         self.test_file_local_path = 'tmp/' + guid + '-hello.txt'
 168         self.port_1_name = guid + '-port-1'
 169         self.port_2_name = guid + '-port-2'
 170         self.floating_ip_name = guid + 'fip1'
 171
 172         # Setup members to cleanup just in case they don't get created
 173         self.inst_creator = None
 174         self.keypair_creator = None
 175         self.sec_grp_creator = None
 176         self.flavor_creator = None
 177         self.router_creator = None
 178         self.network_creator = None
 179         self.image_creator = None
 180
 181         try:
 182             # Create Image
 183             os_image_settings = openstack_tests.cirros_image_settings(
 184                 name=guid + '-' + '-image',
 185                 image_metadata=self.image_metadata)
 186             self.image_creator = create_image.OpenStackImage(self.os_creds,
 187                                                              os_image_settings)
 188             self.image_creator.create()
 189
 190             # First network is public
 191             self.pub_net_config = openstack_tests.get_pub_net_config(
 192                 net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
 193                 router_name=guid + '-pub-router',
 194                 external_net=self.ext_net_name)
 195
 196             self.network_creator = create_network.OpenStackNetwork(
 197                 self.os_creds, self.pub_net_config.network_settings)
 198             self.network_creator.create()
 199
 200             # Create routers
 201             self.router_creator = create_router.OpenStackRouter(
 202                 self.os_creds, self.pub_net_config.router_settings)
 203             self.router_creator.create()
 204
 205             # Create Flavor
 206             self.flavor_creator = create_flavor.OpenStackFlavor(
 207                 self.os_creds,
 208                 FlavorConfig(
 209                     name=guid + '-flavor-name', ram=256, disk=1, vcpus=1))
 210             self.flavor_creator.create()
 211
 212             # Create Key/Pair
 213             self.keypair_creator = create_keypairs.OpenStackKeypair(
 214                 self.os_creds, KeypairConfig(
 215                     name=self.keypair_name,
 216                     public_filepath=self.keypair_pub_filepath,
 217                     private_filepath=self.keypair_priv_filepath))
 218             self.keypair_creator.create()
 219
 220             # Create Security Group
 221             sec_grp_name = guid + '-sec-grp'
 222             rule1 = SecurityGroupRuleConfig(
 223                 sec_grp_name=sec_grp_name, direction=Direction.ingress,
 224                 protocol=Protocol.icmp)
 225             rule2 = SecurityGroupRuleConfig(
 226                 sec_grp_name=sec_grp_name, direction=Direction.ingress,
 227                 protocol=Protocol.tcp, port_range_min=22, port_range_max=22)
 228             self.sec_grp_creator = OpenStackSecurityGroup(
 229                 self.os_creds,
 230                 SecurityGroupConfig(
 231                     name=sec_grp_name, rule_settings=[rule1, rule2]))
 232             self.sec_grp_creator.create()
 233
 234             # Create instance
 235             ports_settings = list()
 236             ports_settings.append(
 237                 PortConfig(
 238                     name=self.port_1_name,
 239                     network_name=self.pub_net_config.network_settings.name))
 240
 241             instance_settings = VmInstanceConfig(
 242                 name=self.vm_inst_name,
 243                 flavor=self.flavor_creator.flavor_settings.name,
 244                 port_settings=ports_settings,
 245                 floating_ip_settings=[FloatingIpConfig(
 246                     name=self.floating_ip_name, port_name=self.port_1_name,
 247                     router_name=self.pub_net_config.router_settings.name)])
 248
 249             self.inst_creator = create_instance.OpenStackVmInstance(
 250                 self.os_creds, instance_settings,
 251                 self.image_creator.image_settings,
 252                 keypair_settings=self.keypair_creator.keypair_settings)
 253         except:
 254             self.tearDown()
 255             raise
 256
 257     def tearDown(self):
 258         """
 259         Cleans the created objects
 260         """
 261         if self.inst_creator:
 262             try:
 263                 self.inst_creator.clean()
 264             except:
 265                 pass
 266
 267         if self.sec_grp_creator:
 268             try:
 269                 self.sec_grp_creator.clean()
 270             except:
 271                 pass
 272
 273         if self.keypair_creator:
 274             try:
 275                 self.keypair_creator.clean()
 276             except:
 277                 pass
 278
 279         if self.flavor_creator:
 280             try:
 281                 self.flavor_creator.clean()
 282             except:
 283                 pass
 284
 285         if os.path.isfile(self.keypair_pub_filepath):
 286             try:
 287                 os.remove(self.keypair_pub_filepath)
 288             except:
 289                 pass
 290
 291         if os.path.isfile(self.keypair_priv_filepath):
 292             try:
 293                 os.remove(self.keypair_priv_filepath)
 294             except:
 295                 pass
 296
 297         if self.router_creator:
 298             try:
 299                 self.router_creator.clean()
 300             except:
 301                 pass
 302
 303         if self.network_creator:
 304             try:
 305                 self.network_creator.clean()
 306             except:
 307                 pass
 308
 309         if self.image_creator and not self.image_creator.image_settings.exists:
 310             try:
 311                 self.image_creator.clean()
 312             except:
 313                 pass
 314
 315         if os.path.isfile(self.test_file_local_path):
 316             os.remove(self.test_file_local_path)
 317
 318         # super(self.__class__, self).__clean__()
 319
 320     def test_derive_vm_inst_config(self):
 321         """
 322         Validates the utility function settings_utils#create_vm_inst_config
 323         returns an acceptable VmInstanceConfig object
 324         """
 325         self.inst_creator.create(block=True)
 326
 327         server = nova_utils.get_server(
 328             self.nova, self.neutron, self.keystone,
 329             vm_inst_settings=self.inst_creator.instance_settings)
 330         derived_vm_settings = settings_utils.create_vm_inst_config(
 331             self.nova, self.keystone, self.neutron, server,
 332             self.os_creds.project_name)
 333         self.assertIsNotNone(derived_vm_settings)
 334         self.assertIsNotNone(derived_vm_settings.port_settings)
 335         self.assertIsNotNone(derived_vm_settings.floating_ip_settings)
 336
 337     def test_derive_image_settings(self):
 338         """
 339         Validates the utility function settings_utils#create_image_settings
 340         returns an acceptable ImageConfig object
 341         """
 342         self.inst_creator.create(block=True)
 343
 344         server = nova_utils.get_server(
 345             self.nova, self.neutron, self.keystone,
 346             vm_inst_settings=self.inst_creator.instance_settings)
 347         derived_image_settings = settings_utils.determine_image_config(
 348             self.glance, server, [self.image_creator.image_settings])
 349         self.assertIsNotNone(derived_image_settings)
 350         self.assertEqual(self.image_creator.image_settings.name,
 351                          derived_image_settings.name)
 352
 353
 354 class SettingsUtilsUnitTests(unittest.TestCase):
 355     """
 356     Exercises the settings_utils.py functions around volumes
 357     """
 358
 359     def test_vol_settings_from_vol(self):
 360         volume = Volume(
 361             name='vol-name', volume_id='vol-id', project_id='proj-id',
 362             description='desc', size=99, vol_type='vol-type',
 363             availability_zone='zone1', multi_attach=True)
 364         settings = settings_utils.create_volume_config(volume)
 365         self.assertEqual(volume.name, settings.name)
 366         self.assertEqual(volume.description, settings.description)
 367         self.assertEqual(volume.size, settings.size)
 368         self.assertEqual(volume.type, settings.type_name)
 369         self.assertEqual(volume.availability_zone, settings.availability_zone)
 370         self.assertEqual(volume.multi_attach, settings.multi_attach)
 371
 372     def test_vol_type_settings_from_vol(self):
 373         encryption = VolumeTypeEncryption(
 374             volume_encryption_id='vol-encrypt-id', volume_type_id='vol-typ-id',
 375             control_location='front-end', provider='FooClass', cipher='1',
 376             key_size=1)
 377         qos_spec = QoSSpec(name='qos-spec-name', spec_id='qos-spec-id',
 378                            consumer=Consumer.back_end)
 379         volume_type = VolumeType(
 380             name='vol-type-name', volume_type_id='vol-type-id', public=True,
 381             encryption=encryption, qos_spec=qos_spec)
 382
 383         settings = settings_utils.create_volume_type_config(volume_type)
 384         self.assertEqual(volume_type.name, settings.name)
 385         self.assertEqual(volume_type.public, settings.public)
 386
 387         encrypt_settings = settings.encryption
 388         self.assertIsNotNone(encrypt_settings)
 389         self.assertEqual(encryption.control_location,
 390                          encrypt_settings.control_location.value)
 391         self.assertEqual(encryption.cipher, encrypt_settings.cipher)
 392         self.assertEqual(encryption.key_size, encrypt_settings.key_size)
 393
 394         self.assertEqual(qos_spec.name, settings.qos_spec_name)
 395
 396     def test_flavor_settings_from_flavor(self):
 397         flavor = Flavor(
 398             name='flavor-name', flavor_id='flavor-id', ram=99, disk=101,
 399             vcpus=9, ephemeral=3, swap=5, rxtx_factor=7, is_public=False)
 400         settings = settings_utils.create_flavor_config(flavor)
 401         self.assertEqual(flavor.name, settings.name)
 402         self.assertEqual(flavor.id, settings.flavor_id)
 403         self.assertEqual(flavor.ram, settings.ram)
 404         self.assertEqual(flavor.disk, settings.disk)
 405         self.assertEqual(flavor.vcpus, settings.vcpus)
 406         self.assertEqual(flavor.ephemeral, settings.ephemeral)
 407         self.assertEqual(flavor.swap, settings.swap)
 408         self.assertEqual(flavor.rxtx_factor, settings.rxtx_factor)
 409         self.assertEqual(flavor.is_public, settings.is_public)