diff options
Diffstat (limited to 'snaps/config')
-rw-r--r-- | snaps/config/router.py | 113 | ||||
-rw-r--r-- | snaps/config/tests/router_tests.py | 98 |
2 files changed, 211 insertions, 0 deletions
diff --git a/snaps/config/router.py b/snaps/config/router.py new file mode 100644 index 0000000..db26870 --- /dev/null +++ b/snaps/config/router.py @@ -0,0 +1,113 @@ +# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs") +# and others. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from snaps.openstack.create_network import PortSettings +from snaps.openstack.utils import neutron_utils, keystone_utils + + +class RouterConfig(object): + """ + Class representing a router configuration + """ + + def __init__(self, **kwargs): + """ + Constructor - all parameters are optional + :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 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 internal_subnets: List of subnet names to which to connect this + router for Floating IP purposes + :param port_settings: List of PortSettings objects + :return: + """ + 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', True) + self.enable_snat = kwargs.get('enable_snat') + if kwargs.get('internal_subnets'): + self.internal_subnets = kwargs['internal_subnets'] + else: + self.internal_subnets = list() + + self.port_settings = list() + if kwargs.get('interfaces', kwargs.get('port_settings')): + interfaces = kwargs.get('interfaces', kwargs.get('port_settings')) + for interface in interfaces: + if isinstance(interface, PortSettings): + self.port_settings.append(interface) + else: + self.port_settings.append( + PortSettings(**interface['port'])) + + if not self.name: + raise RouterConfigError('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 + + TODO - expand automated testing to exercise all parameters + :param neutron: The neutron client to retrieve external network + information if necessary + :param os_creds: The OpenStack credentials + :return: the dictionary object + """ + out = dict() + ext_gw = dict() + + if self.name: + out['name'] = self.name + if self.project_name: + keystone = keystone_utils.keystone_client(os_creds) + project = keystone_utils.get_project( + keystone=keystone, project_name=self.project_name) + project_id = None + if project: + project_id = project.id + if project_id: + out['tenant_id'] = project_id + else: + raise RouterConfigError( + '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, network_name=self.external_gateway) + if ext_net: + ext_gw['network_id'] = ext_net.id + out['external_gateway_info'] = ext_gw + else: + raise RouterConfigError( + 'Could not find the external network named - ' + + self.external_gateway) + + return {'router': out} + + +class RouterConfigError(Exception): + """ + Exception to be thrown when router settings attributes are incorrect + """ diff --git a/snaps/config/tests/router_tests.py b/snaps/config/tests/router_tests.py new file mode 100644 index 0000000..ffa227a --- /dev/null +++ b/snaps/config/tests/router_tests.py @@ -0,0 +1,98 @@ +# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs") +# and others. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at: +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +import unittest + +from snaps.config.router import RouterConfig, RouterConfigError +from snaps.openstack.create_network import PortSettings + + +class RouterConfigUnitTests(unittest.TestCase): + """ + Class for testing the RouterConfig class + """ + + def test_no_params(self): + with self.assertRaises(RouterConfigError): + RouterConfig() + + def test_empty_config(self): + with self.assertRaises(RouterConfigError): + RouterConfig(**dict()) + + def test_name_only(self): + settings = RouterConfig(name='foo') + self.assertEqual('foo', settings.name) + self.assertIsNone(settings.project_name) + self.assertIsNone(settings.external_gateway) + self.assertTrue(settings.admin_state_up) + self.assertIsNone(settings.enable_snat) + self.assertIsNotNone(settings.internal_subnets) + self.assertTrue(isinstance(settings.internal_subnets, list)) + self.assertEqual(0, len(settings.internal_subnets)) + self.assertIsNotNone(settings.port_settings) + self.assertTrue(isinstance(settings.port_settings, list)) + self.assertEqual(0, len(settings.port_settings)) + + def test_config_with_name_only(self): + settings = RouterConfig(**{'name': 'foo'}) + self.assertEqual('foo', settings.name) + self.assertIsNone(settings.project_name) + self.assertIsNone(settings.external_gateway) + self.assertTrue(settings.admin_state_up) + self.assertIsNone(settings.enable_snat) + self.assertIsNotNone(settings.internal_subnets) + self.assertTrue(isinstance(settings.internal_subnets, list)) + self.assertEqual(0, len(settings.internal_subnets)) + self.assertIsNotNone(settings.port_settings) + self.assertTrue(isinstance(settings.port_settings, list)) + self.assertEqual(0, len(settings.port_settings)) + + def test_all(self): + port_settings = PortSettings(name='foo', network_name='bar') + settings = RouterConfig( + name='foo', project_name='bar', external_gateway='foo_gateway', + admin_state_up=True, enable_snat=False, + internal_subnets=['10.0.0.1/24'], interfaces=[port_settings]) + self.assertEqual('foo', settings.name) + self.assertEqual('bar', settings.project_name) + self.assertEqual('foo_gateway', settings.external_gateway) + self.assertTrue(settings.admin_state_up) + self.assertFalse(settings.enable_snat) + self.assertIsNotNone(settings.internal_subnets) + self.assertTrue(isinstance(settings.internal_subnets, list)) + self.assertEqual(1, len(settings.internal_subnets)) + self.assertEqual(['10.0.0.1/24'], settings.internal_subnets) + self.assertEqual([port_settings], settings.port_settings) + + def test_config_all(self): + settings = RouterConfig( + **{'name': 'foo', 'project_name': 'bar', + 'external_gateway': 'foo_gateway', 'admin_state_up': True, + 'enable_snat': False, 'internal_subnets': ['10.0.0.1/24'], + 'interfaces': + [{'port': {'name': 'foo-port', + 'network_name': 'bar-net'}}]}) + self.assertEqual('foo', settings.name) + self.assertEqual('bar', settings.project_name) + self.assertEqual('foo_gateway', settings.external_gateway) + self.assertTrue(settings.admin_state_up) + self.assertFalse(settings.enable_snat) + self.assertIsNotNone(settings.internal_subnets) + self.assertTrue(isinstance(settings.internal_subnets, list)) + self.assertEqual(1, len(settings.internal_subnets)) + self.assertEqual(['10.0.0.1/24'], settings.internal_subnets) + self.assertEqual([PortSettings(**{'name': 'foo-port', + 'network_name': 'bar-net'})], + settings.port_settings) |