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