From b41d8c0dfcdfa8f6610224652b6409e621b183f4 Mon Sep 17 00:00:00 2001 From: spisarski Date: Tue, 20 Jun 2017 10:32:13 -0600 Subject: Changes to RouterSettings constructors to use kwargs. And changed line lengths to 79 for pep8 JIRA: SNAPS-101 Change-Id: I2b63e4130644ad422aa1f81bcf2627e97d97c035 Signed-off-by: spisarski --- snaps/openstack/create_router.py | 145 ++++++++++++---------- snaps/openstack/tests/create_router_tests.py | 179 +++++++++++++++++---------- 2 files changed, 195 insertions(+), 129 deletions(-) diff --git a/snaps/openstack/create_router.py b/snaps/openstack/create_router.py index 0b56424..ffdf2e6 100644 --- a/snaps/openstack/create_router.py +++ b/snaps/openstack/create_router.py @@ -15,7 +15,6 @@ import logging from neutronclient.common.exceptions import NotFound - from snaps.openstack.create_network import PortSettings from snaps.openstack.utils import neutron_utils, keystone_utils @@ -33,8 +32,9 @@ class OpenStackRouter: """ 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 @@ -49,7 +49,8 @@ class OpenStackRouter: 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): @@ -60,18 +61,22 @@ class OpenStackRouter: """ 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: @@ -79,22 +84,32 @@ class OpenStackRouter: 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 @@ -103,19 +118,24 @@ class OpenStackRouter: 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() @@ -148,51 +168,43 @@ class RouterSettings: 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') @@ -200,10 +212,12 @@ class RouterSettings: 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 """ @@ -223,16 +237,21 @@ class RouterSettings: 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 diff --git a/snaps/openstack/tests/create_router_tests.py b/snaps/openstack/tests/create_router_tests.py index 3e22714..bd2588a 100644 --- a/snaps/openstack/tests/create_router_tests.py +++ b/snaps/openstack/tests/create_router_tests.py @@ -19,8 +19,8 @@ from snaps.openstack import create_network 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' @@ -33,7 +33,8 @@ static_gateway_ip2 = '10.200.202.1' 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): @@ -67,31 +68,40 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): """ 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 @@ -101,90 +111,118 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): """ 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): @@ -215,8 +253,10 @@ 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): @@ -224,18 +264,23 @@ class CreateRouterNegativeTests(OSIntegrationTestCase): 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: """ @@ -246,7 +291,8 @@ def verify_router_attributes(router_operational, router_creator, admin_state=Tru 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 @@ -256,7 +302,8 @@ def verify_router_attributes(router_operational, router_creator, admin_state=Tru 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: -- cgit 1.2.3-korg