import logging
from neutronclient.common.exceptions import NotFound
-
from snaps.openstack.create_network import PortSettings
from snaps.openstack.utils import neutron_utils, keystone_utils
"""
Constructor - all parameters are required
:param os_creds: The credentials to connect with OpenStack
- :param router_settings: The settings used to create a router object (must be an instance of the
- RouterSettings class)
+ :param router_settings: The settings used to create a router object
+ (must be an instance of the RouterSettings
+ class)
"""
self.__os_creds = os_creds
self.__internal_subnets = list()
self.__internal_router_interface = None
- # Dict where the port object is the key and any newly created router interfaces are the value
+ # Dict where the port object is the key and any newly created router
+ # interfaces are the value
self.__ports = list()
def create(self, cleanup=False):
"""
self.__neutron = neutron_utils.neutron_client(self.__os_creds)
- logger.debug('Creating Router with name - ' + self.router_settings.name)
+ logger.debug(
+ 'Creating Router with name - ' + self.router_settings.name)
existing = False
- router_inst = neutron_utils.get_router_by_name(self.__neutron, self.router_settings.name)
+ router_inst = neutron_utils.get_router_by_name(
+ self.__neutron, self.router_settings.name)
if router_inst:
self.__router = router_inst
existing = True
else:
if not cleanup:
- self.__router = neutron_utils.create_router(self.__neutron, self.__os_creds, self.router_settings)
+ self.__router = neutron_utils.create_router(
+ self.__neutron, self.__os_creds, self.router_settings)
for internal_subnet_name in self.router_settings.internal_subnets:
- internal_subnet = neutron_utils.get_subnet_by_name(self.__neutron, internal_subnet_name)
+ internal_subnet = neutron_utils.get_subnet_by_name(
+ self.__neutron, internal_subnet_name)
if internal_subnet:
self.__internal_subnets.append(internal_subnet)
if internal_subnet and not cleanup and not existing:
self.__internal_router_interface = neutron_utils.add_interface_router(
self.__neutron, self.__router, subnet=internal_subnet)
else:
- raise Exception('Subnet not found with name ' + internal_subnet_name)
+ raise Exception(
+ 'Subnet not found with name ' + internal_subnet_name)
for port_setting in self.router_settings.port_settings:
- port = neutron_utils.get_port_by_name(self.__neutron, port_setting.name)
- logger.info('Retrieved port ' + port_setting.name + ' for router - ' + self.router_settings.name)
+ port = neutron_utils.get_port_by_name(self.__neutron,
+ port_setting.name)
+ logger.info(
+ 'Retrieved port %s for router - %s', port_setting.name,
+ self.router_settings.name)
if port:
self.__ports.append(port)
if not port and not cleanup and not existing:
- port = neutron_utils.create_port(self.__neutron, self.__os_creds, port_setting)
+ port = neutron_utils.create_port(self.__neutron,
+ self.__os_creds, port_setting)
if port:
- logger.info('Created port ' + port_setting.name + ' for router - ' + self.router_settings.name)
+ logger.info(
+ 'Created port %s for router - %s', port_setting.name,
+ self.router_settings.name)
self.__ports.append(port)
- neutron_utils.add_interface_router(self.__neutron, self.__router, port=port)
+ neutron_utils.add_interface_router(self.__neutron,
+ self.__router,
+ port=port)
else:
- raise Exception('Error creating port with name - ' + port_setting.name)
+ raise Exception(
+ 'Error creating port with name - ' + port_setting.name)
return self.__router
Removes and deletes all items created in reverse order.
"""
for port in self.__ports:
- logger.info('Removing router interface from router ' + self.router_settings.name +
- ' and port ' + port['port']['name'])
+ logger.info(
+ 'Removing router interface from router %s and port %s',
+ self.router_settings.name, port['port']['name'])
try:
- neutron_utils.remove_interface_router(self.__neutron, self.__router, port=port)
+ neutron_utils.remove_interface_router(self.__neutron,
+ self.__router, port=port)
except NotFound:
pass
self.__ports = list()
for internal_subnet in self.__internal_subnets:
- logger.info('Removing router interface from router ' + self.router_settings.name +
- ' and subnet ' + internal_subnet['subnet']['name'])
+ logger.info(
+ 'Removing router interface from router %s and subnet %s',
+ self.router_settings.name, internal_subnet['subnet']['name'])
try:
- neutron_utils.remove_interface_router(self.__neutron, self.__router, subnet=internal_subnet)
+ neutron_utils.remove_interface_router(self.__neutron,
+ self.__router,
+ subnet=internal_subnet)
except NotFound:
pass
self.__internal_subnets = list()
Class representing a router configuration
"""
- def __init__(self, config=None, name=None, project_name=None, external_gateway=None,
- admin_state_up=True, external_fixed_ips=None, internal_subnets=list(),
- port_settings=list()):
+ def __init__(self, **kwargs):
"""
Constructor - all parameters are optional
- :param config: Should be a dict object containing the configuration settings using the attribute names below
- as each member's the key and overrides any of the other parameters.
:param name: The router name.
- :param project_name: The name of the project who owns the network. Only administrative users can specify a
- project ID other than their own. You cannot change this value through authorization
- policies.
+ :param project_name: The name of the project who owns the network. Only
+ administrative users can specify a project ID
+ other than their own. You cannot change this value
+ through authorization policies.
:param external_gateway: Name of the external network to which to route
- :param admin_state_up: The administrative status of the router. True = up / False = down (default True)
- :param external_fixed_ips: Dictionary containing the IP address parameters.
- :param internal_subnets: List of subnet names to which to connect this router for Floating IP purposes
+ :param admin_state_up: The administrative status of the router.
+ True = up / False = down (default True)
+ :param external_fixed_ips: Dictionary containing the IP address
+ parameters.
+ :param internal_subnets: List of subnet names to which to connect this
+ router for Floating IP purposes
:param port_settings: List of PortSettings objects
:return:
"""
- if config:
- self.name = config.get('name')
- self.project_name = config.get('project_name')
- self.external_gateway = config.get('external_gateway')
-
- self.admin_state_up = config.get('admin_state_up')
- self.enable_snat = config.get('enable_snat')
- self.external_fixed_ips = config.get('external_fixed_ips')
- if config.get('internal_subnets'):
- self.internal_subnets = config['internal_subnets']
- else:
- self.internal_subnets = internal_subnets
-
- self.port_settings = list()
- if config.get('interfaces'):
- interfaces = config['interfaces']
- for interface in interfaces:
- if interface.get('port'):
- self.port_settings.append(PortSettings(config=interface['port']))
+ self.name = kwargs.get('name')
+ self.project_name = kwargs.get('project_name')
+ self.external_gateway = kwargs.get('external_gateway')
+
+ self.admin_state_up = kwargs.get('admin_state_up')
+ self.enable_snat = kwargs.get('enable_snat')
+ self.external_fixed_ips = kwargs.get('external_fixed_ips')
+ if kwargs.get('internal_subnets'):
+ self.internal_subnets = kwargs['internal_subnets']
else:
- self.name = name
- self.project_name = project_name
- self.external_gateway = external_gateway
- self.admin_state_up = admin_state_up
- self.external_fixed_ips = external_fixed_ips
- self.internal_subnets = internal_subnets
- self.port_settings = port_settings
+ self.internal_subnets = list()
+
+ self.port_settings = list()
+ if kwargs.get('interfaces'):
+ interfaces = kwargs['interfaces']
+ for interface in interfaces:
+ if interface.get('port'):
+ self.port_settings.append(
+ PortSettings(**interface['port']))
if not self.name:
raise Exception('Name is required')
def dict_for_neutron(self, neutron, os_creds):
"""
Returns a dictionary object representing this object.
- This is meant to be converted into JSON designed for use by the Neutron API
+ This is meant to be converted into JSON designed for use by the Neutron
+ API
TODO - expand automated testing to exercise all parameters
- :param neutron: The neutron client to retrieve external network information if necessary
+ :param neutron: The neutron client to retrieve external network
+ information if necessary
:param os_creds: The OpenStack credentials
:return: the dictionary object
"""
if project_id:
out['project_id'] = project_id
else:
- raise Exception('Could not find project ID for project named - ' + self.project_name)
+ raise Exception(
+ 'Could not find project ID for project named - ' +
+ self.project_name)
if self.admin_state_up is not None:
out['admin_state_up'] = self.admin_state_up
if self.external_gateway:
- ext_net = neutron_utils.get_network(neutron, self.external_gateway, project_id)
+ ext_net = neutron_utils.get_network(neutron, self.external_gateway,
+ project_id)
if ext_net:
ext_gw['network_id'] = ext_net['network']['id']
out['external_gateway_info'] = ext_gw
else:
- raise Exception('Could not find the external network named - ' + self.external_gateway)
+ raise Exception(
+ 'Could not find the external network named - ' +
+ self.external_gateway)
# TODO: Enable SNAT option for Router
# TODO: Add external_fixed_ips Tests
from snaps.openstack import create_router
from snaps.openstack.create_network import NetworkSettings
from snaps.openstack.create_network import OpenStackNetwork
-from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
from snaps.openstack.create_router import RouterSettings
+from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
from snaps.openstack.utils import neutron_utils
__author__ = 'mmakati'
class CreateRouterSuccessTests(OSIntegrationTestCase):
"""
- Class for testing routers with various positive scenarios expected to succeed
+ Class for testing routers with various positive scenarios expected to
+ succeed
"""
def setUp(self):
"""
Test creation of a most basic router with minimal options.
"""
- router_settings = RouterSettings(name=self.guid + '-pub-router', external_gateway=self.ext_net_name)
+ router_settings = RouterSettings(name=self.guid + '-pub-router',
+ external_gateway=self.ext_net_name)
- self.router_creator = create_router.OpenStackRouter(self.os_creds, router_settings)
+ self.router_creator = create_router.OpenStackRouter(self.os_creds,
+ router_settings)
self.router_creator.create()
- router = neutron_utils.get_router_by_name(self.neutron, router_settings.name)
+ router = neutron_utils.get_router_by_name(self.neutron,
+ router_settings.name)
self.assertIsNotNone(router)
- self.assertTrue(verify_router_attributes(router, self.router_creator, ext_gateway=self.ext_net_name))
+ self.assertTrue(verify_router_attributes(
+ router, self.router_creator, ext_gateway=self.ext_net_name))
def test_create_delete_router(self):
"""
- Test that clean() will not raise an exception if the router is deleted by another process.
+ Test that clean() will not raise an exception if the router is deleted
+ by another process.
"""
- self.router_settings = RouterSettings(name=self.guid + '-pub-router', external_gateway=self.ext_net_name)
+ self.router_settings = RouterSettings(
+ name=self.guid + '-pub-router', external_gateway=self.ext_net_name)
- self.router_creator = create_router.OpenStackRouter(self.os_creds, self.router_settings)
+ self.router_creator = create_router.OpenStackRouter(
+ self.os_creds, self.router_settings)
created_router = self.router_creator.create()
self.assertIsNotNone(created_router)
- retrieved_router = neutron_utils.get_router_by_name(self.neutron, self.router_settings.name)
+ retrieved_router = neutron_utils.get_router_by_name(
+ self.neutron, self.router_settings.name)
self.assertIsNotNone(retrieved_router)
neutron_utils.delete_router(self.neutron, created_router)
- retrieved_router = neutron_utils.get_router_by_name(self.neutron, self.router_settings.name)
+ retrieved_router = neutron_utils.get_router_by_name(
+ self.neutron, self.router_settings.name)
self.assertIsNone(retrieved_router)
# Should not raise an exception
"""
Test creation of a basic router with admin state down.
"""
- router_settings = RouterSettings(name=self.guid + '-pub-router', admin_state_up=False)
+ router_settings = RouterSettings(name=self.guid + '-pub-router',
+ admin_state_up=False)
- self.router_creator = create_router.OpenStackRouter(self.os_creds, router_settings)
+ self.router_creator = create_router.OpenStackRouter(self.os_creds,
+ router_settings)
self.router_creator.create()
- router = neutron_utils.get_router_by_name(self.neutron, router_settings.name)
+ router = neutron_utils.get_router_by_name(self.neutron,
+ router_settings.name)
self.assertIsNotNone(router)
- self.assertTrue(verify_router_attributes(router, self.router_creator, admin_state=False))
+ self.assertTrue(verify_router_attributes(router, self.router_creator,
+ admin_state=False))
def test_create_router_admin_state_True(self):
"""
Test creation of a basic router with admin state Up.
"""
- router_settings = RouterSettings(name=self.guid + '-pub-router', admin_state_up=True)
+ router_settings = RouterSettings(name=self.guid + '-pub-router',
+ admin_state_up=True)
- self.router_creator = create_router.OpenStackRouter(self.os_creds, router_settings)
+ self.router_creator = create_router.OpenStackRouter(self.os_creds,
+ router_settings)
self.router_creator.create()
- router = neutron_utils.get_router_by_name(self.neutron, router_settings.name)
+ router = neutron_utils.get_router_by_name(self.neutron,
+ router_settings.name)
self.assertIsNotNone(router)
- self.assertTrue(verify_router_attributes(router, self.router_creator, admin_state=True))
+ self.assertTrue(verify_router_attributes(router, self.router_creator,
+ admin_state=True))
def test_create_router_private_network(self):
"""
- Test creation of a router connected with two private networks and no external gateway
+ Test creation of a router connected with two private networks and no
+ external gateway
"""
- network_settings1 = NetworkSettings(name=self.guid + '-pub-net1',
- subnet_settings=[
- create_network.SubnetSettings(cidr=cidr1,
- name=self.guid + '-pub-subnet1',
- gateway_ip=static_gateway_ip1)])
- network_settings2 = NetworkSettings(name=self.guid + '-pub-net2',
- subnet_settings=[
- create_network.SubnetSettings(cidr=cidr2,
- name=self.guid + '-pub-subnet2',
- gateway_ip=static_gateway_ip2)])
-
- self.network_creator1 = OpenStackNetwork(self.os_creds, network_settings1)
- self.network_creator2 = OpenStackNetwork(self.os_creds, network_settings2)
+ network_settings1 = NetworkSettings(
+ name=self.guid + '-pub-net1',
+ subnet_settings=[
+ create_network.SubnetSettings(
+ cidr=cidr1, name=self.guid + '-pub-subnet1',
+ gateway_ip=static_gateway_ip1)])
+ network_settings2 = NetworkSettings(
+ name=self.guid + '-pub-net2',
+ subnet_settings=[
+ create_network.SubnetSettings(
+ cidr=cidr2, name=self.guid + '-pub-subnet2',
+ gateway_ip=static_gateway_ip2)])
+
+ self.network_creator1 = OpenStackNetwork(self.os_creds,
+ network_settings1)
+ self.network_creator2 = OpenStackNetwork(self.os_creds,
+ network_settings2)
self.network_creator1.create()
self.network_creator2.create()
- port_settings = [create_network.PortSettings(name=self.guid + '-port1', ip_addrs=[
- {'subnet_name': network_settings1.subnet_settings[0].name, 'ip': static_gateway_ip1}],
- network_name=network_settings1.name)
- , create_network.PortSettings(name=self.guid + '-port2', ip_addrs=[
- {'subnet_name': network_settings2.subnet_settings[0].name, 'ip': static_gateway_ip2}],
+ port_settings = [
+ create_network.PortSettings(name=self.guid + '-port1', ip_addrs=[
+ {'subnet_name': network_settings1.subnet_settings[0].name,
+ 'ip': static_gateway_ip1}],
+ network_name=network_settings1.name),
+ create_network.PortSettings(name=self.guid + '-port2', ip_addrs=[
+ {'subnet_name': network_settings2.subnet_settings[0].name,
+ 'ip': static_gateway_ip2}],
network_name=network_settings2.name)]
- router_settings = RouterSettings(name=self.guid + '-pub-router', port_settings=port_settings)
- self.router_creator = create_router.OpenStackRouter(self.os_creds, router_settings)
+ router_settings = RouterSettings(name=self.guid + '-pub-router',
+ port_settings=port_settings)
+ self.router_creator = create_router.OpenStackRouter(self.os_creds,
+ router_settings)
self.router_creator.create()
- router = neutron_utils.get_router_by_name(self.neutron, router_settings.name)
+ router = neutron_utils.get_router_by_name(self.neutron,
+ router_settings.name)
self.assertTrue(verify_router_attributes(router, self.router_creator))
def test_create_router_external_network(self):
"""
- Test creation of a router connected to an external network and a private network.
+ Test creation of a router connected to an external network and a
+ private network.
"""
- network_settings = NetworkSettings(name=self.guid + '-pub-net1',
- subnet_settings=[
- create_network.SubnetSettings(cidr=cidr1,
- name=self.guid + '-pub-subnet1',
- gateway_ip=static_gateway_ip1)])
- self.network_creator1 = OpenStackNetwork(self.os_creds, network_settings)
+ network_settings = NetworkSettings(
+ name=self.guid + '-pub-net1',
+ subnet_settings=[
+ create_network.SubnetSettings(
+ cidr=cidr1, name=self.guid + '-pub-subnet1',
+ gateway_ip=static_gateway_ip1)])
+ self.network_creator1 = OpenStackNetwork(self.os_creds,
+ network_settings)
self.network_creator1.create()
- port_settings = [create_network.PortSettings(name=self.guid + '-port1', ip_addrs=[
- {'subnet_name': network_settings.subnet_settings[0].name, 'ip': static_gateway_ip1}],
- network_name=network_settings.name)]
+ port_settings = [
+ create_network.PortSettings(name=self.guid + '-port1', ip_addrs=[
+ {'subnet_name': network_settings.subnet_settings[0].name,
+ 'ip': static_gateway_ip1}],
+ network_name=network_settings.name)]
router_settings = RouterSettings(name=self.guid + '-pub-router',
- external_gateway=self.ext_net_name, port_settings=port_settings)
- self.router_creator = create_router.OpenStackRouter(self.os_creds, router_settings)
+ external_gateway=self.ext_net_name,
+ port_settings=port_settings)
+ self.router_creator = create_router.OpenStackRouter(self.os_creds,
+ router_settings)
self.router_creator.create()
- router = neutron_utils.get_router_by_name(self.neutron, router_settings.name)
+ router = neutron_utils.get_router_by_name(self.neutron,
+ router_settings.name)
- self.assertTrue(verify_router_attributes(router, self.router_creator, ext_gateway=self.ext_net_name))
+ self.assertTrue(verify_router_attributes(
+ router, self.router_creator, ext_gateway=self.ext_net_name))
class CreateRouterNegativeTests(OSIntegrationTestCase):
Test creating a router without a name.
"""
with self.assertRaises(Exception):
- router_settings = RouterSettings(name=None, external_gateway=self.ext_net_name)
- self.router_creator = create_router.OpenStackRouter(self.os_creds, router_settings)
+ router_settings = RouterSettings(
+ name=None, external_gateway=self.ext_net_name)
+ self.router_creator = create_router.OpenStackRouter(
+ self.os_creds, router_settings)
self.router_creator.create()
def test_create_router_invalid_gateway_name(self):
Test creating a router without a valid network gateway name.
"""
with self.assertRaises(Exception):
- router_settings = RouterSettings(name=self.guid + '-pub-router', external_gateway="Invalid_name")
- self.router_creator = create_router.OpenStackRouter(self.os_creds, router_settings)
+ router_settings = RouterSettings(name=self.guid + '-pub-router',
+ external_gateway="Invalid_name")
+ self.router_creator = create_router.OpenStackRouter(
+ self.os_creds, router_settings)
self.router_creator.create()
-def verify_router_attributes(router_operational, router_creator, admin_state=True, ext_gateway=None):
+def verify_router_attributes(router_operational, router_creator,
+ admin_state=True, ext_gateway=None):
"""
- Helper function to validate the attributes of router created with the one operational
- :param router_operational: Operational Router object returned from neutron utils
- :param router_creator: router_creator object returned from creating a router in the router test functions
+ Helper function to validate the attributes of router created with the one
+ operational
+ :param router_operational: Operational Router object returned from neutron
+ utils
+ :param router_creator: router_creator object returned from creating a
+ router in the router test functions
:param admin_state: True if router is expected to be Up, else False
- :param snat: True is enable_snat is True, else False
:param ext_gateway: None if router is not connected to external gateway
:return:
"""
return False
elif not router_creator:
return False
- elif not (router_operational['router']['name'] == router_creator.router_settings.name):
+ elif not (router_operational['router'][
+ 'name'] == router_creator.router_settings.name):
return False
elif not (router_operational['router']['id'] == router['router']['id']):
return False
return False
elif not (admin_state == router_operational['router']['admin_state_up']):
return False
- elif (ext_gateway is None) and (router_operational['router']['external_gateway_info'] is not None):
+ elif (ext_gateway is None) and \
+ (router_operational['router']['external_gateway_info'] is not None):
return False
elif ext_gateway is not None:
if router_operational['router']['external_gateway_info'] is None:
Code Review / snaps.git / commitdiff