- Router - snaps.openstack.create\_router.OpenStackRouter
- - snaps.openstack.create\_router.RouterSettings
+ - snaps.openstack.router.RouterConfig
- name - the router name (required)
- project\_name - the name of the project (optional - can only be
.. code:: python
- from snaps.openstack.create_router import RouterSettings, OpenStackRouter
+ from snaps.config.router import RouterConfig
+ from snaps.openstack.create_router import OpenStackRouter
- router_settings = RouterSettings(name='router-name', external_gateway='external')
+ router_settings = RouterConfig(name='router-name', external_gateway='external')
router_creator = OpenStackRouter(os_creds, router_settings)
router_creator.create()
Ensures that all required members are included when constructing a
Port domain object
+RouterConfigUnitTests
+---------------------
+
+Ensures that all required members are included when constructing a
+RouterConfig object
+
RouterSettingsUnitTests
-----------------------
Ensures that all required members are included when constructing a
-RouterSettings object
+deprecated RouterSettings object
RouterDomainObjectTests
-----------------------
from snaps.config.image import ImageConfig
from snaps.config.keypair import KeypairConfig
from snaps.config.project import ProjectConfig
+from snaps.config.router import RouterConfig
from snaps.config.user import UserConfig
from snaps.openstack.create_flavor import OpenStackFlavor
from snaps.openstack.create_image import OpenStackImage
PortSettings, NetworkSettings, OpenStackNetwork)
from snaps.openstack.create_project import OpenStackProject
from snaps.openstack.create_qos import QoSSettings, OpenStackQoS
-from snaps.openstack.create_router import RouterSettings, OpenStackRouter
+from snaps.openstack.create_router import OpenStackRouter
from snaps.openstack.create_security_group import (
OpenStackSecurityGroup, SecurityGroupSettings)
from snaps.openstack.create_user import OpenStackUser
# Create routers
creators.append(__create_instances(
- os_creds_dict, OpenStackRouter, RouterSettings,
+ os_creds_dict, OpenStackRouter, RouterConfig,
os_config.get('routers'), 'router', clean, users_dict))
# Create keypairs
--- /dev/null
+# 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
+ """
--- /dev/null
+# 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)
import logging
from neutronclient.common.exceptions import NotFound
-from snaps.openstack.create_network import PortSettings
+
+from snaps.config.router import RouterConfig
from snaps.openstack.openstack_creator import OpenStackNetworkObject
-from snaps.openstack.utils import neutron_utils, keystone_utils
+from snaps.openstack.utils import neutron_utils
__author__ = 'spisarski'
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
+ (must be an instance of the RouterConfig
class)
"""
super(self.__class__, self).__init__(os_creds)
"""
-class RouterSettings:
+class RouterSettings(RouterConfig):
"""
- Class representing a router configuration
+ Class to hold the configuration settings required for creating OpenStack
+ router objects
+ deprecated
"""
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 RouterSettingsError('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 RouterSettingsError(
- '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 RouterSettingsError(
- 'Could not find the external network named - ' +
- self.external_gateway)
-
- return {'router': out}
-
-
-class RouterSettingsError(Exception):
- """
- Exception to be thrown when router settings attributes are incorrect
- """
+ from warnings import warn
+ warn('Use snaps.config.router.RouterConfig instead',
+ DeprecationWarning)
+ super(self.__class__, self).__init__(**kwargs)
from novaclient.exceptions import BadRequest
from snaps import file_utils
+from snaps.config.router import RouterConfig
from snaps.config.keypair import KeypairConfig
from snaps.openstack import create_network, create_router
from snaps.config.flavor import FlavorConfig
from snaps.openstack.create_keypairs import OpenStackKeypair
from snaps.openstack.create_network import (
OpenStackNetwork, PortSettings, NetworkSettings, SubnetSettings)
-from snaps.openstack.create_router import OpenStackRouter, RouterSettings
+from snaps.openstack.create_router import OpenStackRouter
from snaps.openstack.create_security_group import (
SecurityGroupSettings, OpenStackSecurityGroup, SecurityGroupRuleSettings,
Direction, Protocol)
ip_version=6)
network_settings = NetworkSettings(
name=self.guid + '-net', subnet_settings=[subnet_settings])
- router_settings = RouterSettings(
+ router_settings = RouterConfig(
name=self.guid + '-router', external_gateway=self.ext_net_name,
internal_subnets=[subnet_settings.name])
network_settings = NetworkSettings(
name=self.guid + '-net',
subnet_settings=[subnet4_settings, subnet6_settings])
- router_settings = RouterSettings(
+ router_settings = RouterConfig(
name=self.guid + '-router', external_gateway=self.ext_net_name,
internal_subnets=[subnet4_settings.name])
network_name=self.net_config_2.name,
project_name=self.os_creds.project_name)]
- router_settings = RouterSettings(name=self.guid + '-pub-router',
- port_settings=port_settings)
+ router_settings = RouterConfig(
+ name=self.guid + '-pub-router', port_settings=port_settings)
self.router_creator = create_router.OpenStackRouter(
self.os_creds, router_settings)
self.router_creator.create()
import unittest
import uuid
+from snaps.config.router import RouterConfigError, RouterConfig
from snaps.openstack import create_network
from snaps.openstack import create_router
from snaps.openstack.create_network import (
NetworkSettings, PortSettings)
from snaps.openstack.create_network import OpenStackNetwork
-from snaps.openstack.create_router import (
- RouterSettings, RouterSettingsError)
+from snaps.openstack.create_router import RouterSettings
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
from snaps.openstack.utils import neutron_utils, settings_utils
"""
def test_no_params(self):
- with self.assertRaises(RouterSettingsError):
+ with self.assertRaises(RouterConfigError):
RouterSettings()
def test_empty_config(self):
- with self.assertRaises(RouterSettingsError):
+ with self.assertRaises(RouterConfigError):
RouterSettings(**dict())
def test_name_only(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 = RouterConfig(
+ name=self.guid + '-pub-router', external_gateway=self.ext_net_name)
self.router_creator = create_router.OpenStackRouter(self.os_creds,
router_settings)
Test creation of a most basic router with the admin user pointing
to the new project.
"""
- router_settings = RouterSettings(
+ router_settings = RouterConfig(
name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
project_name=self.os_creds.project_name)
Test creation of a most basic router with the new user pointing
to the admin project.
"""
- router_settings = RouterSettings(
+ router_settings = RouterConfig(
name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
project_name=self.admin_os_creds.project_name)
Test that clean() will not raise an exception if the router is deleted
by another process.
"""
- self.router_settings = RouterSettings(
+ self.router_settings = RouterConfig(
name=self.guid + '-pub-router', external_gateway=self.ext_net_name)
self.router_creator = create_router.OpenStackRouter(
"""
Test creation of a basic router with admin state down.
"""
- router_settings = RouterSettings(name=self.guid + '-pub-router',
- admin_state_up=False)
+ router_settings = RouterConfig(
+ name=self.guid + '-pub-router', admin_state_up=False)
self.router_creator = create_router.OpenStackRouter(self.os_creds,
router_settings)
"""
Test creation of a basic router with admin state Up.
"""
- router_settings = RouterSettings(
+ router_settings = RouterConfig(
name=self.guid + '-pub-router', admin_state_up=True)
self.router_creator = create_router.OpenStackRouter(
network_name=network_settings2.name,
project_name=self.os_creds.project_name)]
- router_settings = RouterSettings(name=self.guid + '-pub-router',
- port_settings=port_settings)
+ router_settings = RouterConfig(
+ name=self.guid + '-pub-router', port_settings=port_settings)
self.router_creator = create_router.OpenStackRouter(self.os_creds,
router_settings)
self.router_creator.create()
network_name=network_settings.name,
project_name=self.os_creds.project_name)]
- router_settings = RouterSettings(
+ router_settings = RouterConfig(
name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
port_settings=port_settings)
self.router_creator = create_router.OpenStackRouter(
def check_router_recreation(self, router, orig_settings):
"""
- Validates the derived RouterSettings with the original
+ Validates the derived RouterConfig with the original
:param router: the Router domain object to test
- :param orig_settings: the original RouterSettings object that was
+ :param orig_settings: the original RouterConfig object that was
responsible for creating the router
- :return: the derived RouterSettings object
+ :return: the derived RouterConfig object
"""
derived_settings = settings_utils.create_router_settings(
self.neutron, router)
"""
Test creating a router without a name.
"""
- with self.assertRaises(RouterSettingsError):
- router_settings = RouterSettings(
+ with self.assertRaises(RouterConfigError):
+ router_settings = RouterConfig(
name=None, external_gateway=self.ext_net_name)
self.router_creator = create_router.OpenStackRouter(
self.os_creds, router_settings)
"""
Test creating a router without a valid network gateway name.
"""
- with self.assertRaises(RouterSettingsError):
- router_settings = RouterSettings(name=self.guid + '-pub-router',
- external_gateway="Invalid_name")
+ with self.assertRaises(RouterConfigError):
+ router_settings = RouterConfig(
+ name=self.guid + '-pub-router',
+ external_gateway="Invalid_name")
self.router_creator = create_router.OpenStackRouter(
self.os_creds, router_settings)
self.router_creator.create()
import pkg_resources
from snaps import file_utils
from snaps.config.image import ImageConfig
+from snaps.config.router import RouterConfig
from snaps.openstack.create_network import NetworkSettings, SubnetSettings
-from snaps.openstack.create_router import RouterSettings
from snaps.openstack.os_credentials import OSCreds, ProxySettings
__author__ = 'spisarski'
if router_name:
if subnet_name:
- self.router_settings = RouterSettings(
+ self.router_settings = RouterConfig(
name=router_name, external_gateway=external_gateway,
internal_subnets=[subnet_name])
else:
- self.router_settings = RouterSettings(
+ self.router_settings = RouterConfig(
name=router_name, external_gateway=external_gateway)
"""
Creates a network on which the CMTSs can attach
:param os_creds: The OpenStack credentials object
- :param router_settings: The RouterSettings instance
+ :param router_settings: The RouterConfig instance
:param cleanup: Denotes whether or not this is being called for cleanup
:return: A reference to the network creator objects for each network from
which network elements such as the subnet, router, interface
values if not None, else finds the first with the value of the router_name
parameter, else None
:param neutron: the client
- :param router_settings: the RouterSettings object
+ :param router_settings: the RouterConfig object
:param router_name: the name of the network to retrieve
:return: a SNAPS-OO Router domain object
"""
from snaps import file_utils
from snaps.config.flavor import FlavorConfig
from snaps.config.keypair import KeypairConfig
+from snaps.config.router import RouterConfig
from snaps.openstack.create_instance import (
VmInstanceSettings, FloatingIpSettings)
from snaps.openstack.create_network import (
PortSettings, SubnetSettings, NetworkSettings)
from snaps.openstack.create_security_group import (
SecurityGroupSettings, SecurityGroupRuleSettings)
-from snaps.openstack.create_router import RouterSettings
from snaps.openstack.create_volume import VolumeSettings
from snaps.openstack.create_volume_type import (
VolumeTypeSettings, VolumeTypeEncryptionSettings, ControlLocation)
def create_router_settings(neutron, router):
"""
- Returns a RouterSettings object
+ Returns a RouterConfig object
:param neutron: the neutron client
:param router: a SNAPS-OO Router domain object
:return:
if port_setting.network_name != ext_net_name:
filtered_settings.append(port_setting)
- return RouterSettings(
+ return RouterConfig(
name=router.name, external_gateway=ext_net_name,
admin_state_up=router.admin_state_up,
port_settings=filtered_settings)
import logging
import unittest
+from snaps.config.tests.router_tests import RouterConfigUnitTests
from snaps.config.tests.user_tests import UserConfigUnitTests
from snaps.config.tests.project_tests import ProjectConfigUnitTests
from snaps.config.tests.keypair_tests import KeypairConfigUnitTests
PortSettingsUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
PortDomainObjectTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ RouterConfigUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
RouterSettingsUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
Code Review / snaps.git / commitdiff