| test_create_router_invalid_gateway_name| 2 | Ensures that an exception is raised when attempting to |
| | | create a router to an external network that does not exist|
+----------------------------------------+---------------+-----------------------------------------------------------+
+| test_create_router_admin_ports | 2 | Ensures that an exception is raised when attempting to |
+| | | create a router with ports to networks owned by another |
+| | | project |
++----------------------------------------+---------------+-----------------------------------------------------------+
+
+create_router_tests.py - CreateMultipleRouterTests
+--------------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Neutron API | Description |
++=======================================+===============+===========================================================+
+| test_router_same_name_diff_proj | 2 | Ensures that a router with the same name can be created |
+| | | against different projects |
++---------------------------------------+---------------+-----------------------------------------------------------+
+| test_router_create_by_admin_to | 2 | Ensures that a router can be created by the admin user |
+| _different_project | | to another project and that a creator with the credentials|
+| | | to the other project will not create a new router with |
+| | | the same name |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
+create_router_tests.py - CreateRouterSecurityGroupTests
+-------------------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Neutron API | Description |
++=======================================+===============+===========================================================+
+| test_create_router_secure_port | 2 | Ensures that a router's port can have a security group |
+| | | applied to it |
++---------------------------------------+---------------+-----------------------------------------------------------+
create_qos_tests.py - CreateQoSTests
------------------------------------
project_name=self.project_name)
if not network:
raise PortConfigError(
- 'Cannot locate network with name - ' + self.network_name)
+ 'Cannot locate network with name - ' + self.network_name
+ + ' in project - ' + str(self.project_name))
out['network_id'] = network.id
sec_grp_ids = list()
for sec_grp_name in self.security_groups:
sec_grp = neutron_utils.get_security_group(
- neutron, sec_grp_name=sec_grp_name,
+ neutron, keystone, sec_grp_name=sec_grp_name,
project_name=self.project_name)
if sec_grp:
sec_grp_ids.append(sec_grp.id)
:return: the external network name or None
"""
router = neutron_utils.get_router(
- self.__neutron, router_name=router_name)
+ self.__neutron, self.__keystone, router_name=router_name,
+ project_name=self._os_creds.project_name)
if router and router.external_network_id:
network = neutron_utils.get_network_by_id(
self.__neutron, router.external_network_id)
if not router_settings:
raise RouterCreationError('router_settings is required')
+ self.__keystone = None
+
self.router_settings = router_settings
# Attributes instantiated on create()
"""
super(self.__class__, self).initialize()
+ self.__keystone = keystone_utils.keystone_client(self._os_creds)
+
try:
self.__router = neutron_utils.get_router(
- self._neutron, router_settings=self.router_settings)
+ self._neutron, self.__keystone,
+ router_settings=self.router_settings,
+ project_name=self._os_creds.project_name)
except Unauthorized as e:
logger.warn('Unable to lookup router with name %s - %s',
self.router_settings.name, e)
raise RouterCreationError(
'Subnet not found with name ' + internal_subnet_name)
- keystone = keystone_utils.keystone_client(self._os_creds)
for port_setting in self.router_settings.port_settings:
port = neutron_utils.get_port(
- self._neutron, keystone, port_settings=port_setting,
+ self._neutron, self.__keystone, port_settings=port_setting,
project_name=self._os_creds.project_name)
if port:
self.__ports.append(port)
raise RouterCreationError(
'Subnet not found with name ' + internal_subnet_name)
- keystone = keystone_utils.keystone_client(self._os_creds)
for port_setting in self.router_settings.port_settings:
port = neutron_utils.get_port(
- self._neutron, keystone, port_settings=port_setting,
+ self._neutron, self.__keystone, port_settings=port_setting,
project_name=self._os_creds.project_name)
logger.info(
'Retrieved port %s for router - %s', port_setting.name,
# Validate routers
neutron_utils_tests.validate_router(
- self.neutron, self.router_creator.router_settings.name, True)
+ self.neutron, self.keystone,
+ self.router_creator.router_settings.name,
+ self.os_creds.project_name, True)
neutron_utils_tests.validate_interface_router(
self.router_creator.get_internal_router_interface(),
self.router_creator.create()
retrieved_router = neutron_utils.get_router(
- self.neutron, router_settings=self.router_creator.router_settings)
+ self.neutron, self.keystone,
+ router_settings=self.router_creator.router_settings,
+ project_name=self.os_creds.project_name)
self.assertEqual(
self.router_creator.get_router().id, retrieved_router.id)
self.router_creator.create()
retrieved_router = neutron_utils.get_router(
- self.neutron, router_settings=self.router_creator.router_settings)
+ self.neutron, self.keystone,
+ router_settings=self.router_creator.router_settings,
+ project_name=self.os_creds.project_name)
self.assertEqual(
self.router_creator.get_router().id, retrieved_router.id)
import unittest
import uuid
-from snaps.config.network import PortConfig, NetworkConfig
+from snaps.config.network import PortConfig, NetworkConfig, PortConfigError
from snaps.config.router import RouterConfigError, RouterConfig
+from snaps.config.security_group import SecurityGroupConfig
from snaps.openstack import create_network
from snaps.openstack import create_router
from snaps.openstack.create_network import OpenStackNetwork
-from snaps.openstack.create_router import RouterSettings
+from snaps.openstack.create_router import RouterSettings, OpenStackRouter
+from snaps.openstack.create_security_group import OpenStackSecurityGroup
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
from snaps.openstack.utils import neutron_utils, settings_utils
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)
+ self.router_creator = create_router.OpenStackRouter(
+ self.os_creds, router_settings)
self.router_creator.create()
- router = neutron_utils.get_router(self.neutron,
- router_settings=router_settings)
+ router = neutron_utils.get_router(
+ self.neutron, self.keystone, router_settings=router_settings,
+ project_name=self.os_creds.project_name)
self.assertIsNotNone(router)
self.assertEqual(self.router_creator.get_router(), router)
self.router_creator.create()
router = neutron_utils.get_router(
- self.neutron, router_settings=router_settings)
+ self.neutron, self.keystone, router_settings=router_settings,
+ project_name=self.os_creds.project_name)
self.assertIsNotNone(router)
self.assertEqual(self.router_creator.get_router().id, router.id)
self.router_creator.create()
router = neutron_utils.get_router(
- self.neutron, router_settings=router_settings)
+ self.neutron, self.keystone, router_settings=router_settings,
+ project_name=self.os_creds.project_name)
self.assertIsNotNone(router)
self.assertEqual(self.router_creator.get_router().id, router.id)
created_router = self.router_creator.create()
self.assertIsNotNone(created_router)
retrieved_router = neutron_utils.get_router(
- self.neutron, router_settings=self.router_settings)
+ self.neutron, self.keystone, router_settings=self.router_settings,
+ project_name=self.os_creds.project_name)
self.assertIsNotNone(retrieved_router)
neutron_utils.delete_router(self.neutron, created_router)
retrieved_router = neutron_utils.get_router(
- self.neutron, router_settings=self.router_settings)
+ self.neutron, self.keystone, router_settings=self.router_settings,
+ project_name=self.os_creds.project_name)
self.assertIsNone(retrieved_router)
# Should not raise an exception
router_settings)
self.router_creator.create()
- router = neutron_utils.get_router(self.neutron,
- router_settings=router_settings)
+ router = neutron_utils.get_router(
+ self.neutron, self.keystone, router_settings=router_settings,
+ project_name=self.os_creds.project_name)
self.assertIsNotNone(router)
self.assertEqual(self.router_creator.get_router(), router)
self.router_creator.create()
router = neutron_utils.get_router(
- self.neutron, router_settings=router_settings)
+ self.neutron, self.keystone, router_settings=router_settings,
+ project_name=self.os_creds.project_name)
self.assertIsNotNone(router)
self.assertEqual(self.router_creator.get_router(), router)
self.router_creator.create()
router = neutron_utils.get_router(
- self.neutron, router_settings=router_settings)
+ self.neutron, self.keystone, router_settings=router_settings,
+ project_name=self.os_creds.project_name)
self.assertEqual(router, self.router_creator.get_router())
self.router_creator.create()
router = neutron_utils.get_router(
- self.neutron, router_settings=router_settings)
+ self.neutron, self.keystone, router_settings=router_settings,
+ project_name=self.os_creds.project_name)
self.assertEquals(router, self.router_creator.get_router())
super(self.__class__, self).__start__()
self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ self.network_creator = None
self.router_creator = None
def tearDown(self):
if self.router_creator:
self.router_creator.clean()
+ if self.network_creator:
+ self.network_creator.clean()
+
super(self.__class__, self).__clean__()
def test_create_router_noname(self):
self.router_creator = create_router.OpenStackRouter(
self.os_creds, router_settings)
self.router_creator.create()
+
+ def test_create_router_admin_ports(self):
+ """
+ Test creation of a router with ports to subnets owned by the admin
+ project
+ """
+ network_settings = NetworkConfig(
+ name=self.guid + '-pub-net1',
+ subnet_settings=[
+ create_network.SubnetConfig(
+ cidr=cidr1, name=self.guid + '-pub-subnet1',
+ gateway_ip=static_gateway_ip1)])
+ self.network_creator = OpenStackNetwork(
+ self.admin_os_creds, network_settings)
+ self.network_creator.create()
+
+ port_settings = [
+ create_network.PortConfig(
+ 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 = RouterConfig(
+ 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)
+
+ with self.assertRaises(PortConfigError):
+ self.router_creator.create()
+
+
+class CreateMultipleRouterTests(OSIntegrationTestCase):
+ """
+ Test for the OpenStackRouter class and how it interacts with routers
+ groups within other projects with the same name
+ """
+
+ def setUp(self):
+ """
+ Initializes objects used for router testing
+ """
+ super(self.__class__, self).__start__()
+
+ self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ self.admin_router_creator = None
+ self.proj_router_creator = None
+ self.neutron = neutron_utils.neutron_client(self.os_creds)
+
+ network_settings = NetworkConfig(
+ name=self.guid + '-pub-net', shared=True,
+ subnet_settings=[
+ create_network.SubnetConfig(
+ cidr=cidr1, name=self.guid + '-pub-subnet',
+ gateway_ip=static_gateway_ip1)])
+
+ self.network_creator = OpenStackNetwork(
+ self.admin_os_creds, network_settings)
+ self.network_creator.create()
+
+ def tearDown(self):
+ """
+ Cleans the remote OpenStack objects used for router testing
+ """
+ if self.admin_router_creator:
+ self.admin_router_creator.clean()
+
+ if self.proj_router_creator:
+ self.proj_router_creator.clean()
+
+ if self.network_creator:
+ self.network_creator.clean()
+
+ super(self.__class__, self).__clean__()
+
+ def test_router_same_name_diff_proj(self):
+ """
+ Tests the creation of an OpenStackNetwork with the same name
+ within a different project/tenant when not configured but implied by
+ the OSCreds.
+ """
+ # Create Router
+
+ router_config = RouterConfig(name=self.guid + '-router')
+ self.admin_router_creator = OpenStackRouter(
+ self.admin_os_creds, router_config)
+ self.admin_router_creator.create()
+
+ self.proj_router_creator = OpenStackRouter(
+ self.os_creds, router_config)
+ self.proj_router_creator.create()
+
+ self.assertNotEqual(
+ self.admin_router_creator.get_router().id,
+ self.proj_router_creator.get_router().id)
+
+ admin_creator2 = OpenStackRouter(
+ self.admin_os_creds, router_config)
+ admin_creator2.create()
+ self.assertEqual(
+ self.admin_router_creator.get_router(),
+ admin_creator2.get_router())
+
+ proj_creator2 = OpenStackRouter(self.os_creds, router_config)
+ proj_creator2.create()
+ self.assertEqual(self.proj_router_creator.get_router(),
+ proj_creator2.get_router())
+
+ def test_router_create_by_admin_to_different_project(self):
+ """
+ Tests the creation of an OpenStackRouter by the admin user and
+ initialize again with tenant credentials.
+ """
+ # Create Network
+
+ admin_router_config = RouterConfig(
+ name=self.guid + '-router',
+ project_name=self.os_creds.project_name)
+
+ self.admin_router_creator = OpenStackRouter(
+ self.admin_os_creds, admin_router_config)
+ self.admin_router_creator.create()
+
+ proj_router_config = RouterConfig(
+ name=self.guid + '-router',
+ project_name=self.os_creds.project_name)
+
+ self.proj_router_creator = OpenStackRouter(
+ self.os_creds, proj_router_config)
+ self.proj_router_creator.create()
+
+ self.assertEqual(
+ self.admin_router_creator.get_router().id,
+ self.proj_router_creator.get_router().id)
+
+
+class CreateRouterSecurityGroupTests(OSIntegrationTestCase):
+ """
+ Class for testing routers with ports containing security groups
+ """
+
+ def setUp(self):
+ """
+ Initializes objects used for router testing
+ """
+ super(self.__class__, self).__start__()
+
+ self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ self.router_creator = None
+ self.network_creator = None
+
+ self.sec_grp_creator = OpenStackSecurityGroup(
+ self.os_creds, SecurityGroupConfig(name=self.guid + '-sec_grp'))
+ self.sec_grp_creator.create()
+
+ self.neutron = neutron_utils.neutron_client(self.os_creds)
+
+ def tearDown(self):
+ """
+ Cleans the remote OpenStack objects used for router testing
+ """
+ if self.router_creator:
+ self.router_creator.clean()
+
+ if self.network_creator:
+ self.network_creator.clean()
+
+ if self.sec_grp_creator:
+ self.sec_grp_creator.clean()
+
+ super(self.__class__, self).__clean__()
+
+ def test_create_router_secure_port(self):
+ """
+ Test creation of a router with a port that has a security group.
+ """
+ network_settings = NetworkConfig(
+ name=self.guid + '-pub-net1',
+ subnet_settings=[
+ create_network.SubnetConfig(
+ cidr=cidr1, name=self.guid + '-pub-subnet1')])
+ self.network_creator = OpenStackNetwork(
+ self.os_creds, network_settings)
+ self.network_creator.create()
+
+ port_settings = [
+ create_network.PortConfig(
+ name=self.guid + '-port1',
+ ip_addrs=[{
+ 'subnet_name': network_settings.subnet_settings[0].name,
+ 'ip': static_gateway_ip1}],
+ network_name=network_settings.name,
+ security_groups=[self.sec_grp_creator.sec_grp_settings.name])]
+
+ router_settings = RouterConfig(
+ 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)
+ self.router_creator.create()
return __map_router(neutron, router['router'])
-def get_router(neutron, router_settings=None, router_name=None):
+def get_router(neutron, keystone, router_settings=None, router_name=None,
+ project_name=None):
"""
Returns the first router object (dictionary) found the given the settings
values if not None, else finds the first with the value of the router_name
parameter, else None
- :param neutron: the client
+ :param neutron: the Neutron client
+ :param keystone: the Keystone client
:param router_settings: the RouterConfig object
:param router_name: the name of the network to retrieve
+ :param project_name: the name of the router's project
:return: a SNAPS-OO Router domain object
"""
router_filter = dict()
else:
return None
- routers = neutron.list_routers(**router_filter)
-
- for routerInst in routers['routers']:
- return __map_router(neutron, routerInst)
-
- return None
+ os_routers = neutron.list_routers(**router_filter)
+ for os_router in os_routers['routers']:
+ if project_name:
+ project = keystone_utils.get_project_by_id(
+ keystone, os_router['project_id'])
+ if project and project.name == project_name:
+ return __map_router(neutron, os_router)
def __map_router(neutron, os_router):
if self.router:
try:
neutron_utils.delete_router(self.neutron, self.router)
- validate_router(self.neutron, self.router.name, False)
+ validate_router(
+ self.neutron, self.keystone, self.router.name,
+ self.os_creds.project_name, False)
except:
pass
"""
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)
+ validate_router(
+ self.neutron, self.keystone, self.net_config.router_settings.name,
+ self.os_creds.project_name, True)
def test_create_router_with_public_interface(self):
"""
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)
+ self.neutron, self.keystone, self.net_config.router_settings.name,
+ self.os_creds.project_name, True)
ext_net = neutron_utils.get_network(
self.neutron, self.keystone, network_name=self.ext_net_name)
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)
+ validate_router(
+ self.neutron, self.keystone, self.net_config.router_settings.name,
+ self.os_creds.project_name, True)
self.interface_router = neutron_utils.add_interface_router(
self.neutron, self.router, self.network.subnets[0])
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)
+ validate_router(
+ self.neutron, self.keystone, self.net_config.router_settings.name,
+ self.os_creds.project_name, True)
with self.assertRaises(NeutronException):
self.interface_router = neutron_utils.add_interface_router(
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)
+ validate_router(
+ self.neutron, self.keystone, self.net_config.router_settings.name,
+ self.os_creds.project_name, True)
for subnet in self.network.subnets:
neutron_utils.delete_subnet(self.neutron, subnet)
return False
-def validate_router(neutron, name, exists):
+def validate_router(neutron, keystone, name, project_name, exists):
"""
Returns true if a router for a given name DOES NOT exist if the exists
parameter is false conversely true. Returns false if a router for a given
name DOES exist if the exists parameter is true conversely false.
:param neutron: The neutron client
+ :param keystone: The keystone client
:param name: The expected router name
+ :param project_name: The name of the project in which the router should
+ exist
:param exists: Whether or not the network name should exist or not
:return: True/False
"""
- router = neutron_utils.get_router(neutron, router_name=name)
+ router = neutron_utils.get_router(
+ neutron, keystone, router_name=name, project_name=project_name)
if exists and router:
return True
return False
from snaps.openstack.tests.create_network_tests import (
CreateNetworkSuccessTests, NetworkSettingsUnitTests, PortSettingsUnitTests,
SubnetSettingsUnitTests, CreateNetworkTypeTests, CreateNetworkIPv6Tests,
- CreateNetworkGatewayTests)
+ CreateMultipleNetworkTests, CreateNetworkGatewayTests)
from snaps.openstack.tests.create_project_tests import (
CreateProjectSuccessTests, ProjectSettingsUnitTests,
CreateProjectUserTests)
QoSSettingsUnitTests, CreateQoSTests)
from snaps.openstack.tests.create_router_tests import (
CreateRouterSuccessTests, CreateRouterNegativeTests,
- RouterSettingsUnitTests)
+ RouterSettingsUnitTests, CreateMultipleRouterTests,
+ CreateRouterSecurityGroupTests)
from snaps.openstack.tests.create_security_group_tests import (
CreateSecurityGroupTests, SecurityGroupRuleSettingsUnitTests,
SecurityGroupSettingsUnitTests, CreateMultipleSecurityGroupTests)
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(
+ CreateMultipleNetworkTests, 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,
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(
+ CreateMultipleRouterTests, 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(
+ CreateRouterSecurityGroupTests, 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(
CreateQoSTests, os_creds=os_creds,
ext_net_name=ext_net_name, use_keystone=use_keystone,
Code Review / snaps.git / commitdiff