diff options
Diffstat (limited to 'snaps/openstack')
| -rw-r--r-- | snaps/openstack/create_instance.py | 5 | ||||
| -rw-r--r-- | snaps/openstack/create_network.py | 110 | ||||
| -rw-r--r-- | snaps/openstack/tests/create_instance_tests.py | 255 | ||||
| -rw-r--r-- | snaps/openstack/tests/create_network_tests.py | 166 | ||||
| -rw-r--r-- | snaps/openstack/tests/create_stack_tests.py | 4 | ||||
| -rw-r--r-- | snaps/openstack/utils/neutron_utils.py | 96 | ||||
| -rw-r--r-- | snaps/openstack/utils/tests/neutron_utils_tests.py | 389 |
7 files changed, 800 insertions, 225 deletions
diff --git a/snaps/openstack/create_instance.py b/snaps/openstack/create_instance.py index 60ebf39..706e6a6 100644 --- a/snaps/openstack/create_instance.py +++ b/snaps/openstack/create_instance.py @@ -16,7 +16,7 @@ import logging import time from neutronclient.common.exceptions import PortNotFoundClient -from novaclient.exceptions import NotFound +from novaclient.exceptions import NotFound, BadRequest from snaps.openstack.create_network import PortSettings from snaps.openstack.openstack_creator import OpenStackComputeObject @@ -373,6 +373,9 @@ class OpenStackVmInstance(OpenStackComputeObject): 'Added floating IP %s to port IP %s on instance %s', floating_ip.ip, ip, self.instance_settings.name) return + except BadRequest as bre: + logger.error('Cannot add floating IP [%s]', bre) + raise except Exception as e: logger.debug( 'Retry adding floating IP to instance. Last attempt ' diff --git a/snaps/openstack/create_network.py b/snaps/openstack/create_network.py index bf873f2..d639c2b 100644 --- a/snaps/openstack/create_network.py +++ b/snaps/openstack/create_network.py @@ -14,7 +14,8 @@ # limitations under the License. import logging -from neutronclient.common.exceptions import NotFound +import enum +from neutronclient.common.exceptions import NetworkNotFoundClient from snaps.openstack.openstack_creator import OpenStackNetworkObject from snaps.openstack.utils import keystone_utils, neutron_utils @@ -41,7 +42,6 @@ class OpenStackNetwork(OpenStackNetworkObject): # Attributes instantiated on create() self.__network = None - self.__subnets = list() def initialize(self): """ @@ -54,15 +54,6 @@ class OpenStackNetwork(OpenStackNetworkObject): self._neutron, network_settings=self.network_settings, project_id=self.network_settings.get_project_id(self._os_creds)) - if self.__network: - for subnet_setting in self.network_settings.subnet_settings: - sub_inst = neutron_utils.get_subnet( - self._neutron, subnet_settings=subnet_setting) - if sub_inst: - self.__subnets.append(sub_inst) - logger.debug( - "Subnet '%s' created successfully" % sub_inst.id) - return self.__network def create(self): @@ -77,19 +68,7 @@ class OpenStackNetwork(OpenStackNetworkObject): self.__network = neutron_utils.create_network( self._neutron, self._os_creds, self.network_settings) logger.debug( - "Network '%s' created successfully" % self.__network.id) - - for subnet_setting in self.network_settings.subnet_settings: - sub_inst = neutron_utils.get_subnet( - self._neutron, subnet_settings=subnet_setting) - if not sub_inst: - sub_inst = neutron_utils.create_subnet( - self._neutron, subnet_setting, self._os_creds, - self.__network) - if sub_inst: - self.__subnets.append(sub_inst) - logger.debug( - "Subnet '%s' created successfully" % sub_inst.id) + 'Network [%s] created successfully' % self.__network.id) return self.__network @@ -97,23 +76,11 @@ class OpenStackNetwork(OpenStackNetworkObject): """ Removes and deletes all items created in reverse order. """ - for subnet in self.__subnets: - try: - logger.info( - 'Deleting subnet with name ' + subnet.name) - neutron_utils.delete_subnet(self._neutron, subnet) - except NotFound as e: - logger.warning( - 'Error deleting subnet with message - ' + str(e)) - pass - self.__subnets = list() - if self.__network: try: neutron_utils.delete_network(self._neutron, self.__network) - except NotFound: + except NetworkNotFoundClient: pass - self.__network = None def get_network(self): @@ -123,13 +90,6 @@ class OpenStackNetwork(OpenStackNetworkObject): """ return self.__network - def get_subnets(self): - """ - Returns the OpenStack subnet objects - :return: - """ - return self.__subnets - class NetworkSettings: """ @@ -189,7 +149,7 @@ class NetworkSettings: self.subnet_settings = list() subnet_settings = kwargs.get('subnets') if not subnet_settings: - subnet_settings = kwargs.get('subnet_settings') + subnet_settings = kwargs.get('subnet_settings', list()) if subnet_settings: for subnet_config in subnet_settings: if isinstance(subnet_config, SubnetSettings): @@ -262,6 +222,15 @@ class NetworkSettingsError(Exception): """ +class IPv6Mode(enum.Enum): + """ + A rule's direction + """ + slaac = 'slaac' + stateful = 'dhcpv6-stateful' + stateless = 'dhcpv6-stateless' + + class SubnetSettings: """ Class representing a subnet configuration @@ -301,10 +270,10 @@ class SubnetSettings: ] :param destination: The destination for static route (optional) :param nexthop: The next hop for the destination (optional) - :param ipv6_ra_mode: A valid value is dhcpv6-stateful, - dhcpv6-stateless, or slaac (optional) - :param ipv6_address_mode: A valid value is dhcpv6-stateful, - dhcpv6-stateless, or slaac (optional) + :param ipv6_ra_mode: an instance of the IPv6Mode enum + (optional when enable_dhcp is True) + :param ipv6_address_mode: an instance of the IPv6Mode enum + (optional when enable_dhcp is True) :raise: SubnetSettingsError when config does not have or cidr values are None """ @@ -322,16 +291,19 @@ class SubnetSettings: self.gateway_ip = kwargs.get('gateway_ip') self.enable_dhcp = kwargs.get('enable_dhcp') - if kwargs.get('dns_nameservers'): + if 'dns_nameservers' in kwargs: self.dns_nameservers = kwargs.get('dns_nameservers') else: - self.dns_nameservers = ['8.8.8.8'] + if self.ip_version == 4: + self.dns_nameservers = ['8.8.8.8'] + else: + self.dns_nameservers = list() self.host_routes = kwargs.get('host_routes') self.destination = kwargs.get('destination') self.nexthop = kwargs.get('nexthop') - self.ipv6_ra_mode = kwargs.get('ipv6_ra_mode') - self.ipv6_address_mode = kwargs.get('ipv6_address_mode') + self.ipv6_ra_mode = map_mode(kwargs.get('ipv6_ra_mode')) + self.ipv6_address_mode = map_mode(kwargs.get('ipv6_address_mode')) if not self.name or not self.cidr: raise SubnetSettingsError('Name and cidr required for subnets') @@ -383,12 +355,40 @@ class SubnetSettings: if self.nexthop: out['nexthop'] = self.nexthop if self.ipv6_ra_mode: - out['ipv6_ra_mode'] = self.ipv6_ra_mode + out['ipv6_ra_mode'] = self.ipv6_ra_mode.value if self.ipv6_address_mode: - out['ipv6_address_mode'] = self.ipv6_address_mode + out['ipv6_address_mode'] = self.ipv6_address_mode.value return out +def map_mode(mode): + """ + Takes a the direction value maps it to the Direction enum. When None return + None + :param mode: the mode value + :return: the IPv6Mode enum object + :raise: SubnetSettingsError if value is invalid + """ + if not mode: + return None + if isinstance(mode, IPv6Mode): + return mode + else: + mode_str = str(mode) + if mode_str == 'slaac': + return IPv6Mode.slaac + elif mode_str == 'dhcpv6-stateful': + return IPv6Mode.stateful + elif mode_str == 'stateful': + return IPv6Mode.stateful + elif mode_str == 'dhcpv6-stateless': + return IPv6Mode.stateless + elif mode_str == 'stateless': + return IPv6Mode.stateless + else: + raise SubnetSettingsError('Invalid mode - ' + mode_str) + + class SubnetSettingsError(Exception): """ Exception to be thrown when subnet settings attributes are incorrect diff --git a/snaps/openstack/tests/create_instance_tests.py b/snaps/openstack/tests/create_instance_tests.py index f5793d1..cd4e4da 100644 --- a/snaps/openstack/tests/create_instance_tests.py +++ b/snaps/openstack/tests/create_instance_tests.py @@ -21,6 +21,7 @@ import uuid import os from neutronclient.common.exceptions import InvalidIpForSubnetClient +from novaclient.exceptions import BadRequest from snaps import file_utils from snaps.openstack import create_network, create_router @@ -31,7 +32,7 @@ from snaps.openstack.create_instance import ( VmInstanceSettingsError, FloatingIpSettingsError) from snaps.openstack.create_keypairs import OpenStackKeypair, KeypairSettings from snaps.openstack.create_network import ( - OpenStackNetwork, PortSettings, NetworkSettings) + OpenStackNetwork, PortSettings, NetworkSettings, SubnetSettings) from snaps.openstack.create_router import OpenStackRouter, RouterSettings from snaps.openstack.create_security_group import ( SecurityGroupSettings, OpenStackSecurityGroup, SecurityGroupRuleSettings, @@ -614,12 +615,6 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase): 'Unexpected exception cleaning keypair with message - %s', e) - if os.path.isfile(self.keypair_pub_filepath): - os.remove(self.keypair_pub_filepath) - - if os.path.isfile(self.keypair_priv_filepath): - os.remove(self.keypair_priv_filepath) - if self.flavor_creator: try: self.flavor_creator.clean() @@ -687,7 +682,7 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase): self.image_creator.image_settings, keypair_settings=self.keypair_creator.keypair_settings) self.inst_creators.append(inst_creator) - vm_inst = inst_creator.create() + vm_inst = inst_creator.create(block=True) self.assertEqual(ip_1, inst_creator.get_port_ip(self.port_1_name)) self.assertTrue(inst_creator.vm_active(block=True)) @@ -706,6 +701,7 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase): name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings], + security_group_names=[self.sec_grp_creator.sec_grp_settings.name], floating_ip_settings=[FloatingIpSettings( name=self.floating_ip_name, port_name=self.port_1_name, router_name=self.pub_net_config.router_settings.name)]) @@ -723,8 +719,6 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase): ip = inst_creator.get_port_ip(port_settings.name) self.assertTrue(check_dhcp_lease(inst_creator, ip)) - inst_creator.add_security_group( - self.sec_grp_creator.get_security_group()) self.assertEqual(vm_inst.id, inst_creator.get_vm_inst().id) self.assertTrue(validate_ssh_client(inst_creator)) @@ -742,6 +736,7 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase): name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings], + security_group_names=[self.sec_grp_creator.sec_grp_settings.name], floating_ip_settings=[FloatingIpSettings( name=self.floating_ip_name, port_name=self.port_1_name, router_name=self.pub_net_config.router_settings.name)]) @@ -761,8 +756,6 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase): ip = inst_creator.get_port_ip(port_settings.name) self.assertTrue(check_dhcp_lease(inst_creator, ip)) - inst_creator.add_security_group( - self.sec_grp_creator.get_security_group()) self.assertEqual(vm_inst.id, inst_creator.get_vm_inst().id) self.assertTrue(validate_ssh_client(inst_creator)) @@ -780,6 +773,7 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase): name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=[port_settings], + security_group_names=[self.sec_grp_creator.sec_grp_settings.name], floating_ip_settings=[FloatingIpSettings( name=self.floating_ip_name, port_name=self.port_1_name, router_name=self.pub_net_config.router_settings.name)]) @@ -799,8 +793,6 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase): ip = inst_creator.get_port_ip(port_settings.name) self.assertTrue(check_dhcp_lease(inst_creator, ip)) - inst_creator.add_security_group( - self.sec_grp_creator.get_security_group()) self.assertEqual(vm_inst.id, inst_creator.get_vm_inst().id) self.assertTrue(validate_ssh_client(inst_creator)) @@ -813,6 +805,230 @@ class CreateInstanceSingleNetworkTests(OSIntegrationTestCase): self.assertTrue(validate_ssh_client(inst_creator2)) +class CreateInstanceIPv6NetworkTests(OSIntegrationTestCase): + """ + Test for the CreateInstance class with a single NIC/Port with Floating IPs + """ + + def setUp(self): + """ + Instantiates the CreateImage object that is responsible for downloading + and creating an OS image file within OpenStack + """ + super(self.__class__, self).__start__() + + self.nova = nova_utils.nova_client(self.os_creds) + self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) + self.keypair_priv_filepath = 'tmp/' + self.guid + self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub' + self.keypair_name = self.guid + '-kp' + self.vm_inst_name = self.guid + '-inst' + self.port1_name = self.guid + 'port1' + self.port2_name = self.guid + 'port2' + + # Initialize for tearDown() + self.image_creator = None + self.network_creator = None + self.router_creator = None + self.flavor_creator = None + self.keypair_creator = None + self.sec_grp_creator = None + self.inst_creator = None + + os_image_settings = openstack_tests.cirros_image_settings( + name=self.guid + '-image', image_metadata=self.image_metadata) + try: + self.image_creator = OpenStackImage( + self.os_creds, os_image_settings) + self.image_creator.create() + + self.flavor_creator = OpenStackFlavor( + self.admin_os_creds, + FlavorSettings( + name=self.guid + '-flavor-name', ram=256, disk=10, vcpus=2, + metadata=self.flavor_metadata)) + self.flavor_creator.create() + + self.keypair_creator = OpenStackKeypair( + self.os_creds, KeypairSettings( + name=self.keypair_name, + public_filepath=self.keypair_pub_filepath, + private_filepath=self.keypair_priv_filepath)) + self.keypair_creator.create() + + sec_grp_name = self.guid + '-sec-grp' + rule1 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name, + direction=Direction.ingress, + protocol=Protocol.icmp) + rule2 = SecurityGroupRuleSettings(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, + SecurityGroupSettings(name=sec_grp_name, + rule_settings=[rule1, rule2])) + self.sec_grp_creator.create() + except Exception as e: + self.tearDown() + raise e + + def tearDown(self): + """ + Cleans the created object + """ + if self.inst_creator: + try: + self.inst_creator.clean() + except Exception as e: + logger.error( + 'Unexpected exception cleaning VM instance with message ' + '- %s', e) + + if self.keypair_creator: + try: + self.keypair_creator.clean() + except Exception as e: + logger.error( + 'Unexpected exception cleaning keypair with message - %s', + e) + + if self.flavor_creator: + try: + self.flavor_creator.clean() + except Exception as e: + logger.error( + 'Unexpected exception cleaning flavor with message - %s', + e) + + if self.sec_grp_creator: + try: + self.sec_grp_creator.clean() + except Exception as e: + logger.error( + 'Unexpected exception cleaning security group with message' + ' - %s', e) + + if self.router_creator: + try: + self.router_creator.clean() + except Exception as e: + logger.error( + 'Unexpected exception cleaning router with message - %s', + e) + + if self.network_creator: + try: + self.network_creator.clean() + except Exception as e: + logger.error( + 'Unexpected exception cleaning network with message - %s', + e) + + if self.image_creator and not self.image_creator.image_settings.exists: + try: + self.image_creator.clean() + except Exception as e: + logger.error( + 'Unexpected exception cleaning image with message - %s', e) + + super(self.__class__, self).__clean__() + + def test_v4fip_v6overlay(self): + """ + Tests the ability to assign an IPv4 floating IP to an IPv6 overlay + network when the external network does not have an IPv6 subnet. + """ + subnet_settings = SubnetSettings( + name=self.guid + '-subnet', cidr='1:1:0:0:0:0:0:0/64', + ip_version=6) + network_settings = NetworkSettings( + name=self.guid + '-net', subnet_settings=[subnet_settings]) + router_settings = RouterSettings( + name=self.guid + '-router', external_gateway=self.ext_net_name, + internal_subnets=[subnet_settings.name]) + + # Create Network + self.network_creator = OpenStackNetwork( + self.os_creds, network_settings) + self.network_creator.create() + + # Create Router + self.router_creator = OpenStackRouter( + self.os_creds, router_settings) + self.router_creator.create() + + port_settings = PortSettings( + name=self.port1_name, network_name=network_settings.name) + + instance_settings = VmInstanceSettings( + name=self.vm_inst_name, + flavor=self.flavor_creator.flavor_settings.name, + port_settings=[port_settings], + security_group_names=[self.sec_grp_creator.sec_grp_settings.name], + floating_ip_settings=[FloatingIpSettings( + name='fip1', port_name=self.port1_name, + router_name=router_settings.name)]) + + self.inst_creator = OpenStackVmInstance( + self.os_creds, instance_settings, + self.image_creator.image_settings, + keypair_settings=self.keypair_creator.keypair_settings) + + with self.assertRaises(BadRequest): + self.inst_creator.create(block=True) + + def test_fip_v4and6_overlay(self): + """ + Tests the ability to assign an IPv4 floating IP to an IPv6 overlay + network when the external network does not have an IPv6 subnet. + """ + subnet4_settings = SubnetSettings( + name=self.guid + '-subnet4', cidr='10.0.1.0/24', + ip_version=4) + subnet6_settings = SubnetSettings( + name=self.guid + '-subnet6', cidr='1:1:0:0:0:0:0:0/64', + ip_version=6) + network_settings = NetworkSettings( + name=self.guid + '-net', + subnet_settings=[subnet4_settings, subnet6_settings]) + router_settings = RouterSettings( + name=self.guid + '-router', external_gateway=self.ext_net_name, + internal_subnets=[subnet4_settings.name]) + + # Create Network + self.network_creator = OpenStackNetwork( + self.os_creds, network_settings) + self.network_creator.create() + + # Create Router + self.router_creator = OpenStackRouter( + self.os_creds, router_settings) + self.router_creator.create() + + port_settings = PortSettings( + name=self.port1_name, network_name=network_settings.name) + + instance_settings = VmInstanceSettings( + name=self.vm_inst_name, + flavor=self.flavor_creator.flavor_settings.name, + port_settings=[port_settings], + security_group_names=[self.sec_grp_creator.sec_grp_settings.name], + floating_ip_settings=[FloatingIpSettings( + name='fip1', port_name=self.port1_name, + router_name=router_settings.name)]) + + self.inst_creator = OpenStackVmInstance( + self.os_creds, instance_settings, + self.image_creator.image_settings, + keypair_settings=self.keypair_creator.keypair_settings) + + self.inst_creator.create(block=True) + ssh_client = self.inst_creator.ssh_client() + self.assertIsNotNone(ssh_client) + + class CreateInstancePortManipulationTests(OSIntegrationTestCase): """ Test for the CreateInstance class with a single NIC/Port where mac and IP @@ -1368,12 +1584,6 @@ class CreateInstancePubPrivNetTests(OSIntegrationTestCase): 'Unexpected exception cleaning keypair with message - %s', e) - if os.path.isfile(self.keypair_pub_filepath): - os.remove(self.keypair_pub_filepath) - - if os.path.isfile(self.keypair_priv_filepath): - os.remove(self.keypair_priv_filepath) - if self.flavor_creator: try: self.flavor_creator.clean() @@ -1442,6 +1652,7 @@ class CreateInstancePubPrivNetTests(OSIntegrationTestCase): name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=ports_settings, + security_group_names=[self.sec_grp_creator.sec_grp_settings.name], floating_ip_settings=[FloatingIpSettings( name=self.floating_ip_name, port_name=self.port_1_name, router_name=self.pub_net_config.router_settings.name)]) @@ -1461,10 +1672,6 @@ class CreateInstancePubPrivNetTests(OSIntegrationTestCase): ip = self.inst_creator.get_port_ip(ports_settings[0].name) self.assertTrue(check_dhcp_lease(self.inst_creator, ip)) - # Add security group to VM - self.inst_creator.add_security_group( - self.sec_grp_creator.get_security_group()) - # Effectively blocks until VM's ssh port has been opened self.assertTrue(self.inst_creator.vm_ssh_active(block=True)) diff --git a/snaps/openstack/tests/create_network_tests.py b/snaps/openstack/tests/create_network_tests.py index 49ad6ab..3d0a102 100644 --- a/snaps/openstack/tests/create_network_tests.py +++ b/snaps/openstack/tests/create_network_tests.py @@ -16,14 +16,12 @@ import unittest import uuid from snaps.openstack import create_router -from snaps.openstack.create_network import (OpenStackNetwork, NetworkSettings, - SubnetSettings, PortSettings, - NetworkSettingsError, - SubnetSettingsError, - PortSettingsError) +from snaps.openstack.create_network import ( + OpenStackNetwork, NetworkSettings, SubnetSettings, PortSettings, + NetworkSettingsError, SubnetSettingsError, PortSettingsError, IPv6Mode) from snaps.openstack.tests import openstack_tests -from snaps.openstack.tests.os_source_file_test import (OSIntegrationTestCase, - OSComponentTestCase) +from snaps.openstack.tests.os_source_file_test import ( + OSIntegrationTestCase, OSComponentTestCase) from snaps.openstack.utils import neutron_utils from snaps.openstack.utils.tests import neutron_utils_tests @@ -161,7 +159,7 @@ class SubnetSettingsUnitTests(unittest.TestCase): self.assertIsNone(settings.ipv6_ra_mode) self.assertIsNone(settings.ipv6_address_mode) - def test_all(self): + def test_all_string_enums(self): host_routes = {'destination': '0.0.0.0/0', 'nexthop': '123.456.78.9'} settings = SubnetSettings(name='foo', cidr='10.0.0.0/24', ip_version=6, project_name='bar-project', @@ -187,8 +185,37 @@ class SubnetSettingsUnitTests(unittest.TestCase): self.assertEqual(host_routes, settings.host_routes[0]) self.assertEqual('dest', settings.destination) self.assertEqual('hop', settings.nexthop) - self.assertEqual('dhcpv6-stateful', settings.ipv6_ra_mode) - self.assertEqual('slaac', settings.ipv6_address_mode) + self.assertEqual(IPv6Mode.stateful, settings.ipv6_ra_mode) + self.assertEqual(IPv6Mode.slaac, settings.ipv6_address_mode) + + def test_all_type_enums(self): + host_routes = {'destination': '0.0.0.0/0', 'nexthop': '123.456.78.9'} + settings = SubnetSettings(name='foo', cidr='10.0.0.0/24', ip_version=6, + project_name='bar-project', + start='10.0.0.2', end='10.0.0.101', + gateway_ip='10.0.0.1', enable_dhcp=False, + dns_nameservers=['8.8.8.8'], + host_routes=[host_routes], + destination='dest', + nexthop='hop', + ipv6_ra_mode=IPv6Mode.stateful, + ipv6_address_mode=IPv6Mode.slaac) + self.assertEqual('foo', settings.name) + self.assertEqual('10.0.0.0/24', settings.cidr) + self.assertEqual(6, settings.ip_version) + self.assertEqual('bar-project', settings.project_name) + self.assertEqual('10.0.0.2', settings.start) + self.assertEqual('10.0.0.101', settings.end) + self.assertEqual('10.0.0.1', settings.gateway_ip) + self.assertEqual(False, settings.enable_dhcp) + self.assertEqual(1, len(settings.dns_nameservers)) + self.assertEqual('8.8.8.8', settings.dns_nameservers[0]) + self.assertEqual(1, len(settings.host_routes)) + self.assertEqual(host_routes, settings.host_routes[0]) + self.assertEqual('dest', settings.destination) + self.assertEqual('hop', settings.nexthop) + self.assertEqual(IPv6Mode.stateful, settings.ipv6_ra_mode) + self.assertEqual(IPv6Mode.slaac, settings.ipv6_address_mode) def test_config_all(self): host_routes = {'destination': '0.0.0.0/0', 'nexthop': '123.456.78.9'} @@ -199,7 +226,7 @@ class SubnetSettingsUnitTests(unittest.TestCase): 'gateway_ip': '10.0.0.1', 'enable_dhcp': False, 'dns_nameservers': ['8.8.8.8'], 'host_routes': [host_routes], 'destination': 'dest', 'nexthop': 'hop', - 'ipv6_ra_mode': 'dhcpv6-stateful', + 'ipv6_ra_mode': 'dhcpv6-stateless', 'ipv6_address_mode': 'slaac'}) self.assertEqual('foo', settings.name) self.assertEqual('10.0.0.0/24', settings.cidr) @@ -215,8 +242,8 @@ class SubnetSettingsUnitTests(unittest.TestCase): self.assertEqual(host_routes, settings.host_routes[0]) self.assertEqual('dest', settings.destination) self.assertEqual('hop', settings.nexthop) - self.assertEqual('dhcpv6-stateful', settings.ipv6_ra_mode) - self.assertEqual('slaac', settings.ipv6_address_mode) + self.assertEqual(IPv6Mode.stateless, settings.ipv6_ra_mode) + self.assertEqual(IPv6Mode.slaac, settings.ipv6_address_mode) class PortSettingsUnitTests(unittest.TestCase): @@ -325,7 +352,7 @@ class PortSettingsUnitTests(unittest.TestCase): class CreateNetworkSuccessTests(OSIntegrationTestCase): """ - Test for the CreateNework class defined in create_nework.py + Test for the CreateNetwork class defined in create_nework.py """ def setUp(self): @@ -344,7 +371,6 @@ class CreateNetworkSuccessTests(OSIntegrationTestCase): # Initialize for cleanup self.net_creator = None self.router_creator = None - self.neutron = neutron_utils.neutron_client(self.os_creds) def tearDown(self): """ @@ -362,7 +388,7 @@ class CreateNetworkSuccessTests(OSIntegrationTestCase): """ Tests the creation of an OpenStack network without a router. """ - # Create Nework + # Create Network self.net_creator = OpenStackNetwork(self.os_creds, self.net_config.network_settings) self.net_creator.create() @@ -381,7 +407,7 @@ class CreateNetworkSuccessTests(OSIntegrationTestCase): """ Tests the creation of an OpenStack network, it's deletion, then cleanup """ - # Create Nework + # Create Network self.net_creator = OpenStackNetwork(self.os_creds, self.net_config.network_settings) self.net_creator.create() @@ -429,14 +455,14 @@ class CreateNetworkSuccessTests(OSIntegrationTestCase): neutron_utils_tests.validate_interface_router( self.router_creator.get_internal_router_interface(), self.router_creator.get_router(), - self.net_creator.get_subnets()[0]) + self.net_creator.get_network().subnets[0]) def test_create_networks_same_name(self): """ Tests the creation of an OpenStack network and ensures that the OpenStackNetwork object will not create a second. """ - # Create Nework + # Create Network self.net_creator = OpenStackNetwork(self.os_creds, self.net_config.network_settings) self.net_creator.create() @@ -509,9 +535,108 @@ class CreateNetworkSuccessTests(OSIntegrationTestCase): self.router_creator.get_router().id, retrieved_router.id) +class CreateNetworkIPv6Tests(OSIntegrationTestCase): + """ + Test for the CreateNetwork class defined in create_nework.py when + """ + + def setUp(self): + """ + Sets up object for test + """ + super(self.__class__, self).__start__() + + self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) + self.neutron = neutron_utils.neutron_client(self.os_creds) + + # Initialize for cleanup + self.net_creator = None + + def tearDown(self): + """ + Cleans the network + """ + if self.net_creator: + self.net_creator.clean() + + super(self.__class__, self).__clean__() + + def test_create_network_one_ipv6_subnet(self): + """ + Tests the creation of an OpenStack network without a router. + """ + # Create Network + subnet_settings = SubnetSettings( + name=self.guid + '-subnet', cidr='1:1:0:0:0:0:0:0/64', + ip_version=6) + network_settings = NetworkSettings( + name=self.guid + '-net', subnet_settings=[subnet_settings]) + + self.net_creator = OpenStackNetwork(self.os_creds, network_settings) + self.net_creator.create() + + # Validate network was created + self.assertTrue(neutron_utils_tests.validate_network( + self.neutron, self.net_creator.network_settings.name, True)) + + network = self.net_creator.get_network() + self.assertEqual(1, len(network.subnets)) + subnet = network.subnets[0] + + self.assertEqual(network.id, subnet.network_id) + self.assertEqual(subnet_settings.name, subnet.name) + self.assertEqual('1:1::/64', subnet.cidr) + self.assertEqual(6, subnet.ip_version) + self.assertEqual(0, len(subnet.dns_nameservers)) + + def test_create_network_ipv4_ipv6_subnet(self): + """ + Tests the creation of an OpenStack network without a router. + """ + # Create Network + subnet4_settings = SubnetSettings( + name=self.guid + '-subnet4', cidr='10.0.1.0/24', ip_version=4) + subnet6_settings = SubnetSettings( + name=self.guid + '-subnet6', cidr='1:1:0:0:0:0:0:0/64', + ip_version=6) + + network_settings = NetworkSettings( + name=self.guid + '-net', + subnet_settings=[subnet4_settings, subnet6_settings]) + + self.net_creator = OpenStackNetwork(self.os_creds, network_settings) + self.net_creator.create() + + # Validate network was created + network = self.net_creator.get_network() + self.assertEqual(2, len(network.subnets)) + + subnet4 = None + subnet6 = None + for subnet in network.subnets: + if subnet.name == subnet4_settings.name: + subnet4 = subnet + if subnet.name == subnet6_settings.name: + subnet6 = subnet + + # Validate IPv4 subnet + self.assertEqual(network.id, subnet4.network_id) + self.assertEqual(subnet4_settings.name, subnet4.name) + self.assertEqual(subnet4_settings.cidr, subnet4.cidr) + self.assertEqual(4, subnet4.ip_version) + self.assertEqual(1, len(subnet4.dns_nameservers)) + + # Validate IPv6 subnet + self.assertEqual(network.id, subnet6.network_id) + self.assertEqual(subnet6_settings.name, subnet6.name) + self.assertEqual('1:1::/64', subnet6.cidr) + self.assertEqual(6, subnet6.ip_version) + self.assertEqual(0, len(subnet6.dns_nameservers)) + + class CreateNetworkTypeTests(OSComponentTestCase): """ - Test for the CreateNework class defined in create_nework.py for testing + Test for the CreateNetwork class defined in create_nework.py for testing creating networks of different types """ @@ -527,7 +652,6 @@ class CreateNetworkTypeTests(OSComponentTestCase): # Initialize for cleanup self.net_creator = None - self.neutron = neutron_utils.neutron_client(self.os_creds) def tearDown(self): """ diff --git a/snaps/openstack/tests/create_stack_tests.py b/snaps/openstack/tests/create_stack_tests.py index 690f3c8..dbe5e3a 100644 --- a/snaps/openstack/tests/create_stack_tests.py +++ b/snaps/openstack/tests/create_stack_tests.py @@ -354,8 +354,8 @@ class CreateStackSuccessTests(OSIntegrationTestCase): self.assertIsNotNone(neutron_utils.get_network_by_id( neutron, net_creators[0].get_network().id)) - self.assertEqual(1, len(net_creators[0].get_subnets())) - subnet = net_creators[0].get_subnets()[0] + self.assertEqual(1, len(net_creators[0].get_network().subnets)) + subnet = net_creators[0].get_network().subnets[0] subnet_by_name = neutron_utils.get_subnet( neutron, subnet_name=subnet.name) self.assertEqual(subnet, subnet_by_name) diff --git a/snaps/openstack/utils/neutron_utils.py b/snaps/openstack/utils/neutron_utils.py index cce53a6..9ca9aba 100644 --- a/snaps/openstack/utils/neutron_utils.py +++ b/snaps/openstack/utils/neutron_utils.py @@ -53,15 +53,33 @@ def create_network(neutron, os_creds, network_settings): :param network_settings: A dictionary containing the network configuration and is responsible for creating the network request JSON body - :return: a SNAPS-OO Network domain object + :return: a SNAPS-OO Network domain object if found else None """ - if neutron and network_settings: - logger.info('Creating network with name ' + network_settings.name) - json_body = network_settings.dict_for_neutron(os_creds) - os_network = neutron.create_network(body=json_body) - return Network(**os_network['network']) - else: - raise NeutronException('Failded to create network') + logger.info('Creating network with name ' + network_settings.name) + json_body = network_settings.dict_for_neutron(os_creds) + os_network = neutron.create_network(body=json_body) + + if os_network: + network = get_network_by_id(neutron, os_network['network']['id']) + + subnets = list() + for subnet_settings in network_settings.subnet_settings: + try: + subnets.append( + create_subnet(neutron, subnet_settings, os_creds, network)) + except: + logger.error( + 'Unexpected error creating subnet [%s] for network [%s]', + subnet_settings.name, network.name) + + for subnet in subnets: + delete_subnet(neutron, subnet) + + delete_network(neutron, network) + + raise + + return get_network_by_id(neutron, network.id) def delete_network(neutron, network): @@ -71,6 +89,14 @@ def delete_network(neutron, network): :param network: a SNAPS-OO Network domain object """ if neutron and network: + if network.subnets: + for subnet in network.subnets: + logger.info('Deleting subnet with name ' + subnet.name) + try: + delete_subnet(neutron, subnet) + except NotFound: + pass + logger.info('Deleting network with name ' + network.name) neutron.delete_network(network.id) @@ -100,12 +126,13 @@ def get_network(neutron, network_settings=None, network_name=None, networks = neutron.list_networks(**net_filter) for network, netInsts in networks.items(): for inst in netInsts: - return Network(**inst) + return __map_network(neutron, inst) -def get_network_by_id(neutron, network_id): +def __get_os_network_by_id(neutron, network_id): """ - Returns the network object (dictionary) with the given ID else None + Returns the OpenStack network object (dictionary) with the given ID else + None :param neutron: the client :param network_id: the id of the network to retrieve :return: a SNAPS-OO Network domain object @@ -113,18 +140,42 @@ def get_network_by_id(neutron, network_id): networks = neutron.list_networks(**{'id': network_id}) for network in networks['networks']: if network['id'] == network_id: - return Network(**network) + return network + + +def get_network_by_id(neutron, network_id): + """ + Returns the SNAPS Network domain object for the given ID else None + :param neutron: the client + :param network_id: the id of the network to retrieve + :return: a SNAPS-OO Network domain object + """ + os_network = __get_os_network_by_id(neutron, network_id) + if os_network: + return __map_network(neutron, os_network) + + +def __map_network(neutron, os_network): + """ + Returns the network object (dictionary) with the given ID else None + :param neutron: the client + :param os_network: the OpenStack Network dict + :return: a SNAPS-OO Network domain object + """ + subnets = get_subnets_by_network_id(neutron, os_network['id']) + os_network['subnets'] = subnets + return Network(**os_network) -def create_subnet(neutron, subnet_settings, os_creds, network=None): +def create_subnet(neutron, subnet_settings, os_creds, network): """ Creates a network subnet for OpenStack :param neutron: the client - :param network: the network object :param subnet_settings: A dictionary containing the subnet configuration and is responsible for creating the subnet request JSON body :param os_creds: the OpenStack credentials + :param network: the network object :return: a SNAPS-OO Subnet domain object """ if neutron and network and subnet_settings: @@ -207,9 +258,19 @@ def get_subnets_by_network(neutron, network): :param network: the SNAPS-OO Network domain object :return: a list of Subnet objects """ + return get_subnets_by_network_id(neutron, network.id) + + +def get_subnets_by_network_id(neutron, network_id): + """ + Returns a list of SNAPS-OO Subnet domain objects + :param neutron: the OpenStack neutron client + :param network_id: the subnet's ID + :return: a list of Subnet objects + """ out = list() - os_subnets = neutron.list_subnets(network_id=network.id) + os_subnets = neutron.list_subnets(network_id=network_id) for os_subnet in os_subnets['subnets']: out.append(Subnet(**os_subnet)) @@ -301,7 +362,8 @@ def __map_router(neutron, os_router): port_subnets = list() # Order by create date - sorted_ports = sorted(device_ports, key=lambda dev_port: dev_port['created_at']) + sorted_ports = sorted( + device_ports, key=lambda dev_port: dev_port['created_at']) for port in sorted_ports: subnets = list() @@ -649,7 +711,7 @@ def get_external_networks(neutron): out = list() for network in neutron.list_networks( **{'router:external': True})['networks']: - out.append(Network(**network)) + out.append(__map_network(neutron, network)) return out diff --git a/snaps/openstack/utils/tests/neutron_utils_tests.py b/snaps/openstack/utils/tests/neutron_utils_tests.py index 0726920..9022578 100644 --- a/snaps/openstack/utils/tests/neutron_utils_tests.py +++ b/snaps/openstack/utils/tests/neutron_utils_tests.py @@ -14,6 +14,8 @@ # limitations under the License. import uuid +from neutronclient.common.exceptions import NotFound, BadRequest + from snaps.openstack import create_router from snaps.openstack.create_network import NetworkSettings, SubnetSettings, \ PortSettings @@ -102,7 +104,7 @@ class NeutronUtilsNetworkTests(OSComponentTestCase): def test_create_network(self): """ - Tests the neutron_utils.create_neutron_net() function + Tests the neutron_utils.create_network() function """ self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) @@ -110,10 +112,12 @@ class NeutronUtilsNetworkTests(OSComponentTestCase): self.network.name) self.assertTrue(validate_network( self.neutron, self.net_config.network_settings.name, True)) + self.assertEqual(len(self.net_config.network_settings.subnet_settings), + len(self.network.subnets)) def test_create_network_empty_name(self): """ - Tests the neutron_utils.create_neutron_net() function with an empty + Tests the neutron_utils.create_network() function with an empty network name """ with self.assertRaises(Exception): @@ -123,7 +127,7 @@ class NeutronUtilsNetworkTests(OSComponentTestCase): def test_create_network_null_name(self): """ - Tests the neutron_utils.create_neutron_net() function when the network + Tests the neutron_utils.create_network() function when the network name is None """ with self.assertRaises(Exception): @@ -142,7 +146,6 @@ class NeutronUtilsSubnetTests(OSComponentTestCase): self.port_name = str(guid) + '-port' self.neutron = neutron_utils.neutron_client(self.os_creds) self.network = None - self.subnet = None self.net_config = openstack_tests.get_pub_net_config( net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet', external_net=self.ext_net_name) @@ -151,11 +154,6 @@ class NeutronUtilsSubnetTests(OSComponentTestCase): """ Cleans the remote OpenStack objects """ - if self.subnet: - try: - neutron_utils.delete_subnet(self.neutron, self.subnet) - except: - pass if self.network: try: neutron_utils.delete_network(self.neutron, self.network) @@ -164,7 +162,7 @@ class NeutronUtilsSubnetTests(OSComponentTestCase): def test_create_subnet(self): """ - Tests the neutron_utils.create_neutron_net() function + Tests the neutron_utils.create_network() function """ self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) @@ -174,20 +172,18 @@ class NeutronUtilsSubnetTests(OSComponentTestCase): self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] - self.subnet = neutron_utils.create_subnet( - self.neutron, subnet_setting, self.os_creds, network=self.network) self.assertTrue(validate_subnet( self.neutron, subnet_setting.name, subnet_setting.cidr, True)) subnet_query1 = neutron_utils.get_subnet( self.neutron, subnet_name=subnet_setting.name) - self.assertEqual(self.subnet, subnet_query1) + self.assertEqual(self.network.subnets[0], subnet_query1) subnet_query2 = neutron_utils.get_subnets_by_network(self.neutron, self.network) self.assertIsNotNone(subnet_query2) self.assertEqual(1, len(subnet_query2)) - self.assertEqual(self.subnet, subnet_query2[0]) + self.assertEqual(self.network.subnets[0], subnet_query2[0]) def test_create_subnet_null_name(self): """ @@ -206,7 +202,7 @@ class NeutronUtilsSubnetTests(OSComponentTestCase): def test_create_subnet_empty_name(self): """ - Tests the neutron_utils.create_neutron_net() function with an empty + Tests the neutron_utils.create_network() function with an empty name """ self.network = neutron_utils.create_network( @@ -217,8 +213,6 @@ class NeutronUtilsSubnetTests(OSComponentTestCase): self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] - self.subnet = neutron_utils.create_subnet( - self.neutron, subnet_setting, self.os_creds, network=self.network) self.assertTrue(validate_subnet( self.neutron, subnet_setting.name, subnet_setting.cidr, True)) self.assertFalse(validate_subnet( @@ -226,49 +220,265 @@ class NeutronUtilsSubnetTests(OSComponentTestCase): subnet_query1 = neutron_utils.get_subnet( self.neutron, subnet_name=subnet_setting.name) - self.assertEqual(self.subnet, subnet_query1) + self.assertEqual(self.network.subnets[0], subnet_query1) subnet_query2 = neutron_utils.get_subnets_by_network(self.neutron, self.network) self.assertIsNotNone(subnet_query2) self.assertEqual(1, len(subnet_query2)) - self.assertEqual(self.subnet, subnet_query2[0]) + self.assertEqual(self.network.subnets[0], subnet_query2[0]) def test_create_subnet_null_cidr(self): """ Tests the neutron_utils.create_neutron_subnet() function for an Exception when the subnet CIDR value is None """ - self.network = neutron_utils.create_network( - self.neutron, self.os_creds, self.net_config.network_settings) - self.assertEqual(self.net_config.network_settings.name, - self.network.name) - self.assertTrue(validate_network( - self.neutron, self.net_config.network_settings.name, True)) - + self.net_config.network_settings.subnet_settings[0].cidr = None with self.assertRaises(Exception): - sub_sets = SubnetSettings( - cidr=None, name=self.net_config.subnet_name) - neutron_utils.create_subnet( - self.neutron, sub_sets, self.os_creds, network=self.network) + self.network = neutron_utils.create_network( + self.neutron, self.os_creds, self.net_config.network_settings) def test_create_subnet_empty_cidr(self): """ Tests the neutron_utils.create_neutron_subnet() function for an Exception when the subnet CIDR value is empty """ + self.net_config.network_settings.subnet_settings[0].cidr = '' + with self.assertRaises(Exception): + self.network = neutron_utils.create_network( + self.neutron, self.os_creds, self.net_config.network_settings) + + +class NeutronUtilsIPv6Tests(OSComponentTestCase): + """ + Test for creating IPv6 networks with subnets via neutron_utils.py + """ + + def setUp(self): + self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) + self.neutron = neutron_utils.neutron_client(self.os_creds) + self.network = None + + def tearDown(self): + """ + Cleans the remote OpenStack objects + """ + if self.network: + try: + neutron_utils.delete_network(self.neutron, self.network) + except: + pass + + def test_create_network_slaac(self): + """ + Tests the neutron_utils.create_network() with an IPv6 subnet where DHCP + is True and IPv6 modes are slaac + """ + sub_setting = SubnetSettings( + name=self.guid + '-subnet', cidr='1:1:0:0:0:0:0:0/64', + ip_version=6, dns_nameservers=['2620:0:ccc:0:0:0:0:2'], + gateway_ip='1:1:0:0:0:0:0:1', start='1:1::ff', end='1:1::ffff', + enable_dhcp=True, ipv6_ra_mode='slaac', ipv6_address_mode='slaac') + self.network_settings = NetworkSettings( + name=self.guid + '-net', subnet_settings=[sub_setting]) + self.network = neutron_utils.create_network( - self.neutron, self.os_creds, self.net_config.network_settings) - self.assertEqual(self.net_config.network_settings.name, - self.network.name) - self.assertTrue(validate_network( - self.neutron, self.net_config.network_settings.name, True)) + self.neutron, self.os_creds, self.network_settings) + self.assertEqual(self.network_settings.name, self.network.name) + + subnet_settings = self.network_settings.subnet_settings[0] + self.assertEqual(1, len(self.network.subnets)) + subnet = self.network.subnets[0] + + self.assertEqual(self.network.id, subnet.network_id) + self.assertEqual(subnet_settings.name, subnet.name) + self.assertEqual(subnet_settings.start, subnet.start) + self.assertEqual(subnet_settings.end, subnet.end) + self.assertEqual('1:1::/64', subnet.cidr) + self.assertEqual(6, subnet.ip_version) + self.assertEqual(1, len(subnet.dns_nameservers)) + self.assertEqual( + sub_setting.dns_nameservers[0], subnet.dns_nameservers[0]) + self.assertTrue(subnet.enable_dhcp) + self.assertEqual( + subnet_settings.ipv6_ra_mode.value, subnet.ipv6_ra_mode) + self.assertEqual( + subnet_settings.ipv6_address_mode.value, subnet.ipv6_address_mode) - with self.assertRaises(Exception): - sub_sets = SubnetSettings( - cidr='', name=self.net_config.subnet_name) - neutron_utils.create_subnet(self.neutron, sub_sets, self.os_creds, - network=self.network) + def test_create_network_stateful(self): + """ + Tests the neutron_utils.create_network() with an IPv6 subnet where DHCP + is True and IPv6 modes are stateful + """ + sub_setting = SubnetSettings( + name=self.guid + '-subnet', cidr='1:1:0:0:0:0:0:0/64', + ip_version=6, dns_nameservers=['2620:0:ccc:0:0:0:0:2'], + gateway_ip='1:1:0:0:0:0:0:1', start='1:1::ff', end='1:1::ffff', + enable_dhcp=True, ipv6_ra_mode='dhcpv6-stateful', + ipv6_address_mode='dhcpv6-stateful') + self.network_settings = NetworkSettings( + name=self.guid + '-net', subnet_settings=[sub_setting]) + + self.network = neutron_utils.create_network( + self.neutron, self.os_creds, self.network_settings) + + self.assertEqual(self.network_settings.name, self.network.name) + + subnet_settings = self.network_settings.subnet_settings[0] + self.assertEqual(1, len(self.network.subnets)) + subnet = self.network.subnets[0] + + self.assertEqual(self.network.id, subnet.network_id) + self.assertEqual(subnet_settings.name, subnet.name) + self.assertEqual(subnet_settings.start, subnet.start) + self.assertEqual(subnet_settings.end, subnet.end) + self.assertEqual('1:1::/64', subnet.cidr) + self.assertEqual(6, subnet.ip_version) + self.assertEqual(1, len(subnet.dns_nameservers)) + self.assertEqual( + sub_setting.dns_nameservers[0], subnet.dns_nameservers[0]) + self.assertTrue(subnet.enable_dhcp) + self.assertEqual( + subnet_settings.ipv6_ra_mode.value, subnet.ipv6_ra_mode) + self.assertEqual( + subnet_settings.ipv6_address_mode.value, subnet.ipv6_address_mode) + + def test_create_network_stateless(self): + """ + Tests the neutron_utils.create_network() when DHCP is enabled and + the RA and address modes are both 'slaac' + """ + sub_setting = SubnetSettings( + name=self.guid + '-subnet', cidr='1:1:0:0:0:0:0:0/64', + ip_version=6, dns_nameservers=['2620:0:ccc:0:0:0:0:2'], + gateway_ip='1:1:0:0:0:0:0:1', start='1:1::ff', end='1:1::ffff', + enable_dhcp=True, ipv6_ra_mode='dhcpv6-stateless', + ipv6_address_mode='dhcpv6-stateless') + self.network_settings = NetworkSettings( + name=self.guid + '-net', subnet_settings=[sub_setting]) + + self.network = neutron_utils.create_network( + self.neutron, self.os_creds, self.network_settings) + + self.assertEqual(self.network_settings.name, self.network.name) + + subnet_settings = self.network_settings.subnet_settings[0] + self.assertEqual(1, len(self.network.subnets)) + subnet = self.network.subnets[0] + + self.assertEqual(self.network.id, subnet.network_id) + self.assertEqual(subnet_settings.name, subnet.name) + self.assertEqual(subnet_settings.start, subnet.start) + self.assertEqual(subnet_settings.end, subnet.end) + self.assertEqual('1:1::/64', subnet.cidr) + self.assertEqual(6, subnet.ip_version) + self.assertEqual(1, len(subnet.dns_nameservers)) + self.assertEqual( + sub_setting.dns_nameservers[0], subnet.dns_nameservers[0]) + self.assertTrue(subnet.enable_dhcp) + self.assertEqual( + subnet_settings.ipv6_ra_mode.value, subnet.ipv6_ra_mode) + self.assertEqual( + subnet_settings.ipv6_address_mode.value, subnet.ipv6_address_mode) + + def test_create_network_no_dhcp_slaac(self): + """ + Tests the neutron_utils.create_network() for a BadRequest when + DHCP is not enabled and the RA and address modes are both 'slaac' + """ + sub_setting = SubnetSettings( + name=self.guid + '-subnet', cidr='1:1:0:0:0:0:0:0/64', + ip_version=6, dns_nameservers=['2620:0:ccc:0:0:0:0:2'], + gateway_ip='1:1:0:0:0:0:0:1', start='1:1::ff', end='1:1::ffff', + enable_dhcp=False, ipv6_ra_mode='slaac', ipv6_address_mode='slaac') + self.network_settings = NetworkSettings( + name=self.guid + '-net', subnet_settings=[sub_setting]) + + with self.assertRaises(BadRequest): + self.network = neutron_utils.create_network( + self.neutron, self.os_creds, self.network_settings) + + def test_create_network_invalid_start_ip(self): + """ + Tests the neutron_utils.create_network() that contains one IPv6 subnet + with an invalid start IP to ensure Neutron assigns it the smallest IP + possible + """ + sub_setting = SubnetSettings( + name=self.guid + '-subnet', cidr='1:1::/48', ip_version=6, + start='foo') + self.network_settings = NetworkSettings( + name=self.guid + '-net', subnet_settings=[sub_setting]) + + self.network = neutron_utils.create_network( + self.neutron, self.os_creds, self.network_settings) + + self.assertEqual('1:1::2', self.network.subnets[0].start) + self.assertEqual( + '1:1:0:ffff:ffff:ffff:ffff:ffff', self.network.subnets[0].end) + + def test_create_network_invalid_end_ip(self): + """ + Tests the neutron_utils.create_network() that contains one IPv6 subnet + with an invalid end IP to ensure Neutron assigns it the largest IP + possible + """ + sub_setting = SubnetSettings( + name=self.guid + '-subnet', cidr='1:1::/48', ip_version=6, + end='bar') + self.network_settings = NetworkSettings( + name=self.guid + '-net', subnet_settings=[sub_setting]) + + self.network = neutron_utils.create_network( + self.neutron, self.os_creds, self.network_settings) + + self.assertEqual('1:1::2', self.network.subnets[0].start) + self.assertEqual( + '1:1:0:ffff:ffff:ffff:ffff:ffff', self.network.subnets[0].end) + + def test_create_network_with_bad_cidr(self): + """ + Tests the neutron_utils.create_network() for a BadRequest when + the subnet CIDR is invalid + """ + sub_setting = SubnetSettings( + name=self.guid + '-subnet', cidr='1:1:1:/48', ip_version=6) + self.network_settings = NetworkSettings( + name=self.guid + '-net', subnet_settings=[sub_setting]) + + with self.assertRaises(BadRequest): + self.network = neutron_utils.create_network( + self.neutron, self.os_creds, self.network_settings) + + def test_create_network_invalid_gateway_ip(self): + """ + Tests the neutron_utils.create_network() for a BadRequest when + the subnet gateway IP is invalid + """ + sub_setting = SubnetSettings( + name=self.guid + '-subnet', cidr='1:1::/48', ip_version=6, + gateway_ip='1:2::1') + self.network_settings = NetworkSettings( + name=self.guid + '-net', subnet_settings=[sub_setting]) + + with self.assertRaises(BadRequest): + self.network = neutron_utils.create_network( + self.neutron, self.os_creds, self.network_settings) + + def test_create_network_with_bad_dns(self): + """ + Tests the neutron_utils.create_network() for a BadRequest when + the DNS IP is invalid + """ + sub_setting = SubnetSettings( + name=self.guid + '-subnet', cidr='1:1::/48', ip_version=6, + dns_nameservers=['foo']) + self.network_settings = NetworkSettings( + name=self.guid + '-net', subnet_settings=[sub_setting]) + + with self.assertRaises(BadRequest): + self.network = neutron_utils.create_network( + self.neutron, self.os_creds, self.network_settings) class NeutronUtilsRouterTests(OSComponentTestCase): @@ -281,7 +491,6 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.port_name = str(guid) + '-port' self.neutron = neutron_utils.neutron_client(self.os_creds) self.network = None - self.subnet = None self.port = None self.router = None self.interface_router = None @@ -294,8 +503,8 @@ class NeutronUtilsRouterTests(OSComponentTestCase): Cleans the remote OpenStack objects """ if self.interface_router: - neutron_utils.remove_interface_router(self.neutron, self.router, - self.subnet) + neutron_utils.remove_interface_router( + self.neutron, self.router, self.network.subnets[0]) if self.router: try: @@ -310,22 +519,12 @@ class NeutronUtilsRouterTests(OSComponentTestCase): except: pass - if self.subnet: - try: - neutron_utils.delete_subnet(self.neutron, self.subnet) - except: - pass - if self.network: - try: - neutron_utils.delete_network(self.neutron, self.network) - except: - pass + neutron_utils.delete_network(self.neutron, self.network) def test_create_router_simple(self): """ - Tests the neutron_utils.create_neutron_net() function when an external - gateway is requested + Tests the neutron_utils.create_router() """ self.router = neutron_utils.create_router( self.neutron, self.os_creds, self.net_config.router_settings) @@ -334,8 +533,7 @@ class NeutronUtilsRouterTests(OSComponentTestCase): def test_create_router_with_public_interface(self): """ - Tests the neutron_utils.create_neutron_net() function when an external - gateway is requested + Tests the neutron_utils.create_router() function with a pubic interface """ subnet_setting = self.net_config.network_settings.subnet_settings[0] self.net_config = openstack_tests.OSNetworkConfig( @@ -351,30 +549,7 @@ class NeutronUtilsRouterTests(OSComponentTestCase): ext_net = neutron_utils.get_network( self.neutron, network_name=self.ext_net_name) - self.assertEqual( - self.router.external_network_id, ext_net.id) - - def test_create_router_empty_name(self): - """ - Tests the neutron_utils.create_neutron_net() function - """ - with self.assertRaises(Exception): - this_router_settings = create_router.RouterSettings(name='') - self.router = neutron_utils.create_router(self.neutron, - self.os_creds, - this_router_settings) - - def test_create_router_null_name(self): - """ - Tests the neutron_utils.create_neutron_subnet() function when the - subnet CIDR value is None - """ - with self.assertRaises(Exception): - this_router_settings = create_router.RouterSettings() - self.router = neutron_utils.create_router(self.neutron, - self.os_creds, - this_router_settings) - validate_router(self.neutron, None, True) + self.assertEqual(self.router.external_network_id, ext_net.id) def test_add_interface_router(self): """ @@ -388,9 +563,6 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] - self.subnet = neutron_utils.create_subnet( - self.neutron, subnet_setting, - self.os_creds, self.network) self.assertTrue(validate_subnet( self.neutron, subnet_setting.name, subnet_setting.cidr, True)) @@ -400,9 +572,9 @@ class NeutronUtilsRouterTests(OSComponentTestCase): True) self.interface_router = neutron_utils.add_interface_router( - self.neutron, self.router, self.subnet) + self.neutron, self.router, self.network.subnets[0]) validate_interface_router(self.interface_router, self.router, - self.subnet) + self.network.subnets[0]) def test_add_interface_router_null_router(self): """ @@ -417,15 +589,12 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] - self.subnet = neutron_utils.create_subnet( - self.neutron, subnet_setting, - self.os_creds, self.network) self.assertTrue(validate_subnet( self.neutron, subnet_setting.name, subnet_setting.cidr, True)) with self.assertRaises(NeutronException): self.interface_router = neutron_utils.add_interface_router( - self.neutron, self.router, self.subnet) + self.neutron, self.router, self.network.subnets[0]) def test_add_interface_router_null_subnet(self): """ @@ -446,7 +615,31 @@ class NeutronUtilsRouterTests(OSComponentTestCase): with self.assertRaises(NeutronException): self.interface_router = neutron_utils.add_interface_router( - self.neutron, self.router, self.subnet) + self.neutron, self.router, None) + + def test_add_interface_router_missing_subnet(self): + """ + Tests the neutron_utils.add_interface_router() function for an + Exception when the subnet object has been deleted + """ + self.network = neutron_utils.create_network( + self.neutron, self.os_creds, self.net_config.network_settings) + self.assertEqual(self.net_config.network_settings.name, + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) + + self.router = neutron_utils.create_router( + self.neutron, self.os_creds, self.net_config.router_settings) + validate_router(self.neutron, self.net_config.router_settings.name, + True) + + for subnet in self.network.subnets: + neutron_utils.delete_subnet(self.neutron, subnet) + + with self.assertRaises(NotFound): + self.interface_router = neutron_utils.add_interface_router( + self.neutron, self.router, self.network.subnets[0]) def test_create_port(self): """ @@ -460,8 +653,6 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] - self.subnet = neutron_utils.create_subnet( - self.neutron, subnet_setting, self.os_creds, self.network) self.assertTrue(validate_subnet( self.neutron, subnet_setting.name, subnet_setting.cidr, True)) @@ -486,8 +677,6 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] - self.subnet = neutron_utils.create_subnet( - self.neutron, subnet_setting, self.os_creds, self.network) self.assertTrue(validate_subnet(self.neutron, subnet_setting.name, subnet_setting.cidr, True)) @@ -512,9 +701,6 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] - self.subnet = neutron_utils.create_subnet( - self.neutron, subnet_setting, - self.os_creds, self.network) self.assertTrue(validate_subnet( self.neutron, subnet_setting.name, subnet_setting.cidr, True)) @@ -559,9 +745,6 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] - self.subnet = neutron_utils.create_subnet( - self.neutron, subnet_setting, - self.os_creds, self.network) self.assertTrue(validate_subnet( self.neutron, subnet_setting.name, subnet_setting.cidr, True)) @@ -588,8 +771,6 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] - self.subnet = neutron_utils.create_subnet( - self.neutron, subnet_setting, self.os_creds, self.network) self.assertTrue(validate_subnet( self.neutron, subnet_setting.name, subnet_setting.cidr, True)) @@ -616,8 +797,6 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] - self.subnet = neutron_utils.create_subnet( - self.neutron, subnet_setting, self.os_creds, self.network) self.assertTrue(validate_subnet( self.neutron, subnet_setting.name, subnet_setting.cidr, True)) |