+---------------------------------------+---------------+-----------------------------------------------------------+
| Test Name | Neutron API | Description |
+=======================================+===============+===========================================================+
-| test_create_subnet | 2 | Ensures neutron_utils.create_subnet() can properly create |
+| test_create_subnet | 2 | Ensures neutron_utils.create_network() can properly create|
| | | an OpenStack subnet object |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_create_subnet_null_name | 2 | Ensures neutron_utils.create_subnet() raises an exception |
+| test_create_subnet_null_name | 2 | Ensures neutron_utils.create_network() raises an exception|
| | | when the subnet name is None |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_create_subnet_empty_name | 2 | Ensures neutron_utils.create_subnet() raises an exception |
+| test_create_subnet_empty_name | 2 | Ensures neutron_utils.create_network() raises an exception|
| | | when the subnet name is an empty string |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_create_subnet_null_cidr | 2 | Ensures neutron_utils.create_subnet() raises an exception |
+| test_create_subnet_null_cidr | 2 | Ensures neutron_utils.create_network() raises an exception|
| | | when the subnet CIDR is None |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_create_subnet_empty_cidr | 2 | Ensures neutron_utils.create_subnet() raises an exception |
+| test_create_subnet_empty_cidr | 2 | Ensures neutron_utils.create_network() raises an exception|
| | | when the subnet CIDR is an empty string |
+---------------------------------------+---------------+-----------------------------------------------------------+
+neutron_utils_tests.py - NeutronUtilsIPv6Tests
+----------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Neutron API | Description |
++=======================================+===============+===========================================================+
+| test_create_network_slaac | 2 | Ensures neutron_utils.create_network() can properly create|
+| | | an OpenStack network with an IPv6 subnet when DHCP is True|
+| | | and modes are 'slaac' |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_network_stateful | 2 | Ensures neutron_utils.create_network() can properly create|
+| | | an OpenStack network with an IPv6 subnet when DHCP is True|
+| | | and modes are 'stateful' |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_network_stateless | 2 | Ensures neutron_utils.create_network() can properly create|
+| | | an OpenStack network with an IPv6 subnet when DHCP is True|
+| | | and modes are 'stateless' |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_network_no_dhcp_slaac | 2 | Ensures neutron_utils.create_network() raises a BadRequest|
+| | | exception when deploying the network with an IPv6 subnet |
+| | | when DHCP is False and modes are 'slaac' |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_network_invalid_start_ip | 2 | Ensures neutron_utils.create_network() sets the start IP |
+| | | address to the minimum value when the start configuration |
+| | | parameter is some garbage value |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_network_invalid_end_ip | 2 | Ensures neutron_utils.create_network() sets the end IP |
+| | | address to the maximum value when the end configuration |
+| | | parameter is some garbage value |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_network_with_bad_cidr | 2 | Ensures neutron_utils.create_network() raises a BadRequest|
+| | | exception when the IPv6 CIDR is incorrect |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_network_invalid_gateway_ip| 2 | Ensures neutron_utils.create_network() raises a BadRequest|
+| | | exception when the IPv6 gateway IP does not match the CIDR|
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_network_with_bad_dns | 2 | Ensures neutron_utils.create_network() raises a BadRequest|
+| | | exception when the IPv6 DNS IP address is not a valid IPv6|
+| | | address |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
neutron_utils_tests.py - NeutronUtilsRouterTests
------------------------------------------------
| face | | an OpenStack router object with an interface to the |
| | | external network |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_create_router_empty_name | 2 | Ensures neutron_utils.create_router() raises an exception |
-| | | when the name is an empty string |
-+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_create_router_null_name | 2 | Ensures neutron_utils.create_router() raises an exception |
-| | | when the name is None |
-+---------------------------------------+---------------+-----------------------------------------------------------+
| test_add_interface_router | 2 | Ensures neutron_utils.add_interface_router() properly adds|
| | | an interface to another subnet |
+---------------------------------------+---------------+-----------------------------------------------------------+
| test_add_interface_router_null_subnet | 2 | Ensures neutron_utils.add_interface_router() raises an |
| | | exception when the subnet object is None |
+---------------------------------------+---------------+-----------------------------------------------------------+
+| test_add_interface_router_missing_sub | 2 | Ensures neutron_utils.add_interface_router() raises an |
+| net | | exception when the subnet object had been deleted |
++---------------------------------------+---------------+-----------------------------------------------------------+
| test_create_port | 2 | Ensures neutron_utils.create_port() can properly create an|
| | | OpenStack port object |
+---------------------------------------+---------------+-----------------------------------------------------------+
| | | 'admin' project ID |
+---------------------------------------+---------------+-----------------------------------------------------------+
+create_network_tests.py - CreateNetworkIPv6Tests
+------------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Neutron API | Description |
++=======================================+===============+===========================================================+
+| test_create_network_one_ipv6_subnet | 2 | Ensures that a network can be created with an IPv6 subnet |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_network_ipv4_ipv6_subnet | 2 | Ensures that a network can be created with an IPv4 and |
+| | | IPv6 subnet |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
create_router_tests.py - CreateRouterSuccessTests
-------------------------------------------------
| | Neutron 2 | delete it when using a 3-part image |
+-----------------------------------------------------+---------------+-----------------------------------------------------------+
-create_instance_tests.py - CreateInstancePubPrivNetTests
---------------------------------------------------------
+create_instance_tests.py - CreateInstancePubPrivNetTests (Staging)
+------------------------------------------------------------------
+---------------------------------------+---------------+-----------------------------------------------------------+
| Test Name | API Versions | Description |
| | Neutron 2 | NIC configured via SSH/Ansible after startup |
+---------------------------------------+---------------+-----------------------------------------------------------+
+create_instance_tests.py - CreateInstanceIPv6NetworkTests (Staging)
+-------------------------------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | API Versions | Description |
++=======================================+===============+===========================================================+
+| test_v4fip_v6overlay | Nova 2 | Expects a BadRequest exception to be raised when |
+| | Neutron 2 | attempting to add an IPv4 floating IP to a VM with an IPv6|
+| | | port |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_fip_v4and6_overlay | Nova 2 | Connects to a VM via a floating IP joined to a port that |
+| | Neutron 2 | has been confiured with both IPv4 and IPv6 addresses |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
create_instance_tests.py - InstanceSecurityGroupTests
-----------------------------------------------------
def __init__(self, **kwargs):
"""
Constructor
+ :param name: the network's name
+ :param id: the network's ID
+ :param admin_state_up: T/F - network is up when True
+ :param shared: T/F - network can be shared amongst other project's
+ :param external: T/F - network is deemed to be external
+ :param type: vlan, vxlan, flat, etc.
+ :param subnets: list of Subnet objects
"""
self.name = kwargs.get('name')
self.id = kwargs.get('id')
self.shared = kwargs.get('shared')
self.external = kwargs.get('router:external', kwargs.get('external'))
self.type = kwargs.get('provider:network_type', kwargs.get('type'))
+ self.subnets = kwargs.get('subnets', list())
def __eq__(self, other):
return (self.name == other.name and self.id == other.id and
self.admin_state_up == other.admin_state_up and
self.shared == other.shared and
- self.external == other.external and self.type == other.type)
+ self.external == other.external and
+ self.type == other.type and
+ self.subnets == other.subnets)
class Subnet:
def __init__(self, **kwargs):
"""
Constructor
+ :param name: the network's name
+ :param id: the subnet's ID
+ :param network_id: the network's ID
+ :param cidr: the CIDR
+ :param ip_version: the IP version
+ :param gateway_ip: the IP of the gateway
+ :param enable_dhcp: T/F if DHCP is enabled
+ :param dns_nameservers: list of DNS server IPs
+ :param host_routes: routes as returned in a dict by Neutron
+ :param ipv6_ra_mode: IPv6 RA Mode
+ :param ipv6_address_mode: IPv6 Address Mode
+ :param start: start IP address pool
+ :param end: end IP address pool
"""
self.name = kwargs.get('name')
self.id = kwargs.get('id')
"""
def test_construction_kwargs_1(self):
+ subnet = Subnet(
+ **{'name': 'foo', 'id': 'bar', 'network_id': 'foo-bar'})
network = Network(
**{'name': 'foo', 'id': 'bar', 'provider:network_type': 'flat',
'admin_state_up': False, 'shared': True,
- 'router:external': False})
+ 'router:external': False, 'subnets': [subnet]})
self.assertEqual('foo', network.name)
self.assertEqual('bar', network.id)
self.assertEqual('flat', network.type)
self.assertFalse(network.admin_state_up)
self.assertFalse(network.external)
self.assertTrue(network.shared)
+ self.assertEqual([subnet], network.subnets)
def test_construction_kwargs_2(self):
+ subnet = Subnet(
+ **{'name': 'foo', 'id': 'bar', 'network_id': 'foo-bar'})
network = Network(
**{'name': 'foo', 'id': 'bar', 'type': 'flat',
- 'admin_state_up': False, 'shared': True,
- 'external': False})
+ 'admin_state_up': False, 'shared': True, 'external': False,
+ 'subnets': [subnet]})
self.assertEqual('foo', network.name)
self.assertEqual('bar', network.id)
self.assertEqual('flat', network.type)
self.assertFalse(network.admin_state_up)
self.assertFalse(network.external)
self.assertTrue(network.shared)
+ self.assertEqual([subnet], network.subnets)
def test_construction_named(self):
+ subnet = Subnet(
+ **{'name': 'foo', 'id': 'bar', 'network_id': 'foo-bar'})
network = Network(
name='foo', id='bar', type='flat', admin_state_up=False,
- shared=True, external=False)
+ shared=True, external=False, subnets=[subnet])
self.assertEqual('foo', network.name)
self.assertEqual('bar', network.id)
self.assertEqual('flat', network.type)
self.assertFalse(network.admin_state_up)
self.assertFalse(network.external)
self.assertTrue(network.shared)
+ self.assertEqual([subnet], network.subnets)
class SubnetObjectTests(unittest.TestCase):
self.assertIsNone(router.external_network_id)
def test_ext_gateway_named(self):
- ext_gateway = {'network_id': '123',
- 'external_fixed_ips': ['456', '789']}
router = Router(
external_fixed_ips=['456', '789'], external_network_id='123',
admin_state_up='yes', tenant_id='1234', status='hello', id='id',
self.assertEqual(['456', '789'], router.external_fixed_ips)
self.assertEqual('123', router.external_network_id)
-
def test_ext_net_ips_named(self):
ext_gateway = {'network_id': '123',
'external_fixed_ips': ['456', '789']}
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
'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 '
# 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
# Attributes instantiated on create()
self.__network = None
- self.__subnets = list()
def initialize(self):
"""
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):
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
"""
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):
"""
return self.__network
- def get_subnets(self):
- """
- Returns the OpenStack subnet objects
- :return:
- """
- return self.__subnets
-
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):
"""
+class IPv6Mode(enum.Enum):
+ """
+ A rule's direction
+ """
+ slaac = 'slaac'
+ stateful = 'dhcpv6-stateful'
+ stateless = 'dhcpv6-stateless'
+
+
class SubnetSettings:
"""
Class representing a subnet configuration
]
: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
"""
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')
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
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
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,
'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()
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))
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)])
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))
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)])
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))
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)])
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))
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
'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()
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)])
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))
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
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',
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'}
'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)
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):
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):
# Initialize for cleanup
self.net_creator = None
self.router_creator = None
- self.neutron = neutron_utils.neutron_client(self.os_creds)
def tearDown(self):
"""
"""
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()
"""
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()
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()
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
"""
# Initialize for cleanup
self.net_creator = None
- self.neutron = neutron_utils.neutron_client(self.os_creds)
def tearDown(self):
"""
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)
: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):
: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)
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
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:
: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))
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()
out = list()
for network in neutron.list_networks(
**{'router:external': True})['networks']:
- out.append(Network(**network))
+ out.append(__map_network(neutron, network))
return out
# 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
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)
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):
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):
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)
"""
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)
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)
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):
"""
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(
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(
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):
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
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:
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)
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(
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):
"""
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))
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):
"""
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):
"""
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):
"""
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))
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))
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))
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))
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))
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))
VmInstanceSettingsUnitTests, CreateInstancePortManipulationTests,
SimpleHealthCheck, CreateInstanceFromThreePartImage,
CreateInstanceMockOfflineTests, CreateInstanceTwoNetTests,
- CreateInstanceVolumeTests)
+ CreateInstanceVolumeTests, CreateInstanceIPv6NetworkTests)
from snaps.openstack.tests.create_keypairs_tests import (
CreateKeypairsTests, KeypairSettingsUnitTests, CreateKeypairsCleanupTests)
from snaps.openstack.tests.create_network_tests import (
CreateNetworkSuccessTests, NetworkSettingsUnitTests, PortSettingsUnitTests,
- SubnetSettingsUnitTests, CreateNetworkTypeTests)
+ SubnetSettingsUnitTests, CreateNetworkTypeTests, CreateNetworkIPv6Tests)
from snaps.openstack.tests.create_project_tests import (
CreateProjectSuccessTests, ProjectSettingsUnitTests,
CreateProjectUserTests)
from snaps.openstack.utils.tests.neutron_utils_tests import (
NeutronSmokeTests, NeutronUtilsNetworkTests, NeutronUtilsSubnetTests,
NeutronUtilsRouterTests, NeutronUtilsSecurityGroupTests,
- NeutronUtilsFloatingIpTests)
+ NeutronUtilsFloatingIpTests, NeutronUtilsIPv6Tests)
from snaps.openstack.utils.tests.nova_utils_tests import (
NovaSmokeTests, NovaUtilsKeypairTests, NovaUtilsFlavorTests,
NovaUtilsInstanceTests, NovaUtilsInstanceVolumeTests)
suite.addTest(OSComponentTestCase.parameterize(
NeutronUtilsSubnetTests, os_creds=os_creds, ext_net_name=ext_net_name,
log_level=log_level))
+ suite.addTest(OSComponentTestCase.parameterize(
+ NeutronUtilsIPv6Tests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
NeutronUtilsRouterTests, os_creds=os_creds, ext_net_name=ext_net_name,
log_level=log_level))
ext_net_name=ext_net_name, use_keystone=use_keystone,
flavor_metadata=flavor_metadata, image_metadata=image_metadata,
log_level=log_level))
+ suite.addTest(OSIntegrationTestCase.parameterize(
+ CreateNetworkIPv6Tests, os_creds=os_creds,
+ ext_net_name=ext_net_name, use_keystone=use_keystone,
+ flavor_metadata=flavor_metadata, image_metadata=image_metadata,
+ log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(
CreateRouterSuccessTests, os_creds=os_creds, ext_net_name=ext_net_name,
use_keystone=use_keystone,
:return: None as the tests will be adding to the 'suite' parameter object
"""
suite.addTest(OSComponentTestCase.parameterize(
- CreateNetworkTypeTests, os_creds=os_creds, ext_net_name=ext_net_name,
- log_level=log_level))
+ CreateNetworkTypeTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
CreateInstanceMockOfflineTests, os_creds=os_creds,
ext_net_name=ext_net_name, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(
- CreateInstancePubPrivNetTests, os_creds=os_creds,
- ext_net_name=ext_net_name, log_level=log_level))
+ CreateInstancePubPrivNetTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level))
+ suite.addTest(OSIntegrationTestCase.parameterize(
+ CreateInstanceIPv6NetworkTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
HeatUtilsVolumeTests, os_creds=os_creds,
ext_net_name=ext_net_name, log_level=log_level))
Code Review / snaps.git / commitdiff