# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs") # and others. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at: # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import logging import unittest import os import uuid from snaps.config.network import SubnetConfig, NetworkConfig, PortConfig from snaps.config.flavor import FlavorConfig from snaps.config.keypair import KeypairConfig from snaps.config.qos import Consumer from snaps.config.security_group import ( SecurityGroupRuleConfig, Direction, Protocol, SecurityGroupConfig) from snaps.config.vm_inst import VmInstanceConfig, FloatingIpConfig from snaps.domain.flavor import Flavor from snaps.domain.volume import ( Volume, VolumeType, VolumeTypeEncryption, QoSSpec) from snaps.openstack import ( create_image, create_network, create_router, create_flavor, create_keypairs, create_instance) from snaps.openstack.create_qos import Consumer from snaps.openstack.create_network import OpenStackNetwork from snaps.openstack.create_security_group import OpenStackSecurityGroup from snaps.openstack.tests import openstack_tests from snaps.openstack.tests.os_source_file_test import OSComponentTestCase from snaps.openstack.utils import ( neutron_utils, settings_utils, nova_utils, glance_utils) __author__ = 'spisarski' logger = logging.getLogger('nova_utils_tests') class SettingsUtilsNetworkingTests(OSComponentTestCase): """ Tests the ability to reverse engineer NetworkConfig objects from existing networks deployed to OpenStack """ def setUp(self): """ Instantiates OpenStack instances that cannot be spawned by Heat """ guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) self.network_name = guid + '-net' self.subnet_name = guid + '-subnet' self.net_creator = None self.neutron = neutron_utils.neutron_client(self.os_creds) def tearDown(self): """ Cleans the image and downloaded image file """ if self.net_creator: try: self.net_creator.clean() except: pass def test_derive_net_settings_no_subnet(self): """ Validates the utility function settings_utils#create_network_config returns an acceptable NetworkConfig object and ensures that the new settings object will not cause the new OpenStackNetwork instance to create another network """ net_settings = NetworkConfig(name=self.network_name) self.net_creator = OpenStackNetwork(self.os_creds, net_settings) network = self.net_creator.create() derived_settings = settings_utils.create_network_config( self.neutron, network) self.assertIsNotNone(derived_settings) self.assertEqual(net_settings.name, derived_settings.name) self.assertEqual(net_settings.admin_state_up, derived_settings.admin_state_up) self.assertEqual(net_settings.external, derived_settings.external) self.assertEqual(len(net_settings.subnet_settings), len(derived_settings.subnet_settings)) net_creator = OpenStackNetwork(self.os_creds, derived_settings) derived_network = net_creator.create() self.assertEqual(network, derived_network) def test_derive_net_settings_two_subnets(self): """ Validates the utility function settings_utils#create_network_config returns an acceptable NetworkConfig object """ subnet_settings = list() subnet_settings.append(SubnetConfig(name='sub1', cidr='10.0.0.0/24')) subnet_settings.append(SubnetConfig(name='sub2', cidr='10.0.1.0/24')) net_settings = NetworkConfig( name=self.network_name, subnet_settings=subnet_settings) self.net_creator = OpenStackNetwork(self.os_creds, net_settings) network = self.net_creator.create() derived_settings = settings_utils.create_network_config( self.neutron, network) self.assertIsNotNone(derived_settings) self.assertEqual(net_settings.name, derived_settings.name) self.assertEqual(net_settings.admin_state_up, derived_settings.admin_state_up) self.assertEqual(net_settings.external, derived_settings.external) self.assertEqual(len(net_settings.subnet_settings), len(derived_settings.subnet_settings)) # Validate the first subnet orig_sub1 = net_settings.subnet_settings[0] found = False for derived_sub in derived_settings.subnet_settings: if orig_sub1.name == derived_sub.name: self.assertEqual(orig_sub1.cidr, derived_sub.cidr) found = True self.assertTrue(found) # Validate the second subnet orig_sub2 = net_settings.subnet_settings[1] found = False for derived_sub in derived_settings.subnet_settings: if orig_sub2.name == derived_sub.name: self.assertEqual(orig_sub2.cidr, derived_sub.cidr) self.assertEqual(orig_sub2.ip_version, derived_sub.ip_version) found = True self.assertTrue(found) class SettingsUtilsVmInstTests(OSComponentTestCase): """ Tests the ability to reverse engineer VmInstanceConfig objects from existing VMs/servers deployed to OpenStack """ def setUp(self): """ Instantiates the CreateImage object that is responsible for downloading and creating an OS image file within OpenStack """ self.nova = nova_utils.nova_client(self.os_creds) self.glance = glance_utils.glance_client(self.os_creds) self.neutron = neutron_utils.neutron_client(self.os_creds) guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) self.keypair_priv_filepath = 'tmp/' + guid self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub' self.keypair_name = guid + '-kp' self.vm_inst_name = guid + '-inst' self.test_file_local_path = 'tmp/' + guid + '-hello.txt' self.port_1_name = guid + '-port-1' self.port_2_name = guid + '-port-2' self.floating_ip_name = guid + 'fip1' # Setup members to cleanup just in case they don't get created self.inst_creator = None self.keypair_creator = None self.sec_grp_creator = None self.flavor_creator = None self.router_creator = None self.network_creator = None self.image_creator = None try: # Create Image os_image_settings = openstack_tests.cirros_image_settings( name=guid + '-' + '-image', image_metadata=self.image_metadata) self.image_creator = create_image.OpenStackImage(self.os_creds, os_image_settings) self.image_creator.create() # First network is public self.pub_net_config = openstack_tests.get_pub_net_config( net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet', router_name=guid + '-pub-router', external_net=self.ext_net_name) self.network_creator = create_network.OpenStackNetwork( self.os_creds, self.pub_net_config.network_settings) self.network_creator.create() # Create routers self.router_creator = create_router.OpenStackRouter( self.os_creds, self.pub_net_config.router_settings) self.router_creator.create() # Create Flavor self.flavor_creator = create_flavor.OpenStackFlavor( self.os_creds, FlavorConfig( name=guid + '-flavor-name', ram=256, disk=1, vcpus=1)) self.flavor_creator.create() # Create Key/Pair self.keypair_creator = create_keypairs.OpenStackKeypair( self.os_creds, KeypairConfig( name=self.keypair_name, public_filepath=self.keypair_pub_filepath, private_filepath=self.keypair_priv_filepath)) self.keypair_creator.create() # Create Security Group sec_grp_name = guid + '-sec-grp' rule1 = SecurityGroupRuleConfig( sec_grp_name=sec_grp_name, direction=Direction.ingress, protocol=Protocol.icmp) rule2 = SecurityGroupRuleConfig( sec_grp_name=sec_grp_name, direction=Direction.ingress, protocol=Protocol.tcp, port_range_min=22, port_range_max=22) self.sec_grp_creator = OpenStackSecurityGroup( self.os_creds, SecurityGroupConfig( name=sec_grp_name, rule_settings=[rule1, rule2])) self.sec_grp_creator.create() # Create instance ports_settings = list() ports_settings.append( PortConfig( name=self.port_1_name, network_name=self.pub_net_config.network_settings.name)) instance_settings = VmInstanceConfig( name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=ports_settings, floating_ip_settings=[FloatingIpConfig( name=self.floating_ip_name, port_name=self.port_1_name, router_name=self.pub_net_config.router_settings.name)]) self.inst_creator = create_instance.OpenStackVmInstance( self.os_creds, instance_settings, self.image_creator.image_settings, keypair_settings=self.keypair_creator.keypair_settings) except: self.tearDown() raise def tearDown(self): """ Cleans the created objects """ if self.inst_creator: try: self.inst_creator.clean() except: pass if self.sec_grp_creator: try: self.sec_grp_creator.clean() except: pass if self.keypair_creator: try: self.keypair_creator.clean() except: pass if self.flavor_creator: try: self.flavor_creator.clean() except: pass if os.path.isfile(self.keypair_pub_filepath): try: os.remove(self.keypair_pub_filepath) except: pass if os.path.isfile(self.keypair_priv_filepath): try: os.remove(self.keypair_priv_filepath) except: pass if self.router_creator: try: self.router_creator.clean() except: pass if self.network_creator: try: self.network_creator.clean() except: pass if self.image_creator and not self.image_creator.image_settings.exists: try: self.image_creator.clean() except: pass if os.path.isfile(self.test_file_local_path): os.remove(self.test_file_local_path) # super(self.__class__, self).__clean__() def test_derive_vm_inst_config(self): """ Validates the utility function settings_utils#create_vm_inst_config returns an acceptable VmInstanceConfig object """ self.inst_creator.create(block=True) server = nova_utils.get_server( self.nova, vm_inst_settings=self.inst_creator.instance_settings) derived_vm_settings = settings_utils.create_vm_inst_config( self.nova, self.neutron, server) self.assertIsNotNone(derived_vm_settings) self.assertIsNotNone(derived_vm_settings.port_settings) self.assertIsNotNone(derived_vm_settings.floating_ip_settings) def test_derive_image_settings(self): """ Validates the utility function settings_utils#create_image_settings returns an acceptable ImageConfig object """ self.inst_creator.create(block=True) server = nova_utils.get_server( self.nova, vm_inst_settings=self.inst_creator.instance_settings) derived_image_settings = settings_utils.determine_image_config( self.glance, server, [self.image_creator.image_settings]) self.assertIsNotNone(derived_image_settings) self.assertEqual(self.image_creator.image_settings.name, derived_image_settings.name) class SettingsUtilsUnitTests(unittest.TestCase): """ Exercises the settings_utils.py functions around volumes """ def test_vol_settings_from_vol(self): volume = Volume( name='vol-name', volume_id='vol-id', description='desc', size=99, vol_type='vol-type', availability_zone='zone1', multi_attach=True) settings = settings_utils.create_volume_config(volume) self.assertEqual(volume.name, settings.name) self.assertEqual(volume.description, settings.description) self.assertEqual(volume.size, settings.size) self.assertEqual(volume.type, settings.type_name) self.assertEqual(volume.availability_zone, settings.availability_zone) self.assertEqual(volume.multi_attach, settings.multi_attach) def test_vol_type_settings_from_vol(self): encryption = VolumeTypeEncryption( volume_encryption_id='vol-encrypt-id', volume_type_id='vol-typ-id', control_location='front-end', provider='FooClass', cipher='1', key_size=1) qos_spec = QoSSpec(name='qos-spec-name', spec_id='qos-spec-id', consumer=Consumer.back_end) volume_type = VolumeType( name='vol-type-name', volume_type_id='vol-type-id', public=True, encryption=encryption, qos_spec=qos_spec) settings = settings_utils.create_volume_type_config(volume_type) self.assertEqual(volume_type.name, settings.name) self.assertEqual(volume_type.public, settings.public) encrypt_settings = settings.encryption self.assertIsNotNone(encrypt_settings) self.assertEqual(encryption.control_location, encrypt_settings.control_location.value) self.assertEqual(encryption.cipher, encrypt_settings.cipher) self.assertEqual(encryption.key_size, encrypt_settings.key_size) self.assertEqual(qos_spec.name, settings.qos_spec_name) def test_flavor_settings_from_flavor(self): flavor = Flavor( name='flavor-name', flavor_id='flavor-id', ram=99, disk=101, vcpus=9, ephemeral=3, swap=5, rxtx_factor=7, is_public=False) settings = settings_utils.create_flavor_config(flavor) self.assertEqual(flavor.name, settings.name) self.assertEqual(flavor.id, settings.flavor_id) self.assertEqual(flavor.ram, settings.ram) self.assertEqual(flavor.disk, settings.disk) self.assertEqual(flavor.vcpus, settings.vcpus) self.assertEqual(flavor.ephemeral, settings.ephemeral) self.assertEqual(flavor.swap, settings.swap) self.assertEqual(flavor.rxtx_factor, settings.rxtx_factor) self.assertEqual(flavor.is_public, settings.is_public)