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