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