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         # super(self.__class__, self).__start__()
 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                 create_flavor.FlavorSettings(name=guid + '-flavor-name',
 206                                              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, create_keypairs.KeypairSettings(
 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 ImageSettings 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_settings(
 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_settings(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)