import unittest
import uuid
+from snaps.config.network import PortConfig, NetworkConfig
+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
+from snaps.openstack.utils import neutron_utils, settings_utils
__author__ = 'mmakati'
"""
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):
self.assertEqual('foo', settings.name)
self.assertIsNone(settings.project_name)
self.assertIsNone(settings.external_gateway)
- self.assertIsNone(settings.admin_state_up)
+ self.assertTrue(settings.admin_state_up)
self.assertIsNone(settings.enable_snat)
- self.assertIsNone(settings.external_fixed_ips)
self.assertIsNotNone(settings.internal_subnets)
self.assertTrue(isinstance(settings.internal_subnets, list))
self.assertEqual(0, len(settings.internal_subnets))
self.assertEqual('foo', settings.name)
self.assertIsNone(settings.project_name)
self.assertIsNone(settings.external_gateway)
- self.assertIsNone(settings.admin_state_up)
+ self.assertTrue(settings.admin_state_up)
self.assertIsNone(settings.enable_snat)
- self.assertIsNone(settings.external_fixed_ips)
self.assertIsNotNone(settings.internal_subnets)
self.assertTrue(isinstance(settings.internal_subnets, list))
self.assertEqual(0, len(settings.internal_subnets))
self.assertEqual(0, len(settings.port_settings))
def test_all(self):
- port_settings = PortSettings(name='foo', network_name='bar')
+ port_settings = PortConfig(name='foo', network_name='bar')
settings = RouterSettings(
name='foo', project_name='bar', external_gateway='foo_gateway',
- admin_state_up=True, enable_snat=False, external_fixed_ips=['ip1'],
+ 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.assertEqual(['ip1'], settings.external_fixed_ips)
self.assertIsNotNone(settings.internal_subnets)
self.assertTrue(isinstance(settings.internal_subnets, list))
self.assertEqual(1, len(settings.internal_subnets))
settings = RouterSettings(
**{'name': 'foo', 'project_name': 'bar',
'external_gateway': 'foo_gateway', 'admin_state_up': True,
- 'enable_snat': False, 'external_fixed_ips': ['ip1'],
- 'internal_subnets': ['10.0.0.1/24'],
+ 'enable_snat': False, 'internal_subnets': ['10.0.0.1/24'],
'interfaces':
[{'port': {'name': 'foo-port',
'network_name': 'bar-net'}}]})
self.assertEqual('foo_gateway', settings.external_gateway)
self.assertTrue(settings.admin_state_up)
self.assertFalse(settings.enable_snat)
- self.assertEqual(['ip1'], settings.external_fixed_ips)
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)
+ self.assertEqual(
+ [PortConfig(**{'name': 'foo-port', 'network_name': 'bar-net'})],
+ settings.port_settings)
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 = 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 = neutron_utils.get_router_by_name(self.neutron,
- router_settings.name)
+ router = neutron_utils.get_router(self.neutron,
+ router_settings=router_settings)
self.assertIsNotNone(router)
- self.assertTrue(verify_router_attributes(
- router, self.router_creator, ext_gateway=self.ext_net_name))
+ self.assertEqual(self.router_creator.get_router(), router)
+
+ self.check_router_recreation(router, router_settings)
def test_create_router_admin_user_to_new_project(self):
"""
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)
self.admin_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(
+ self.neutron, router_settings=router_settings)
self.assertIsNotNone(router)
- self.assertTrue(verify_router_attributes(
- router, self.router_creator, ext_gateway=self.ext_net_name))
+ self.assertEqual(self.router_creator.get_router().id, router.id)
+
+ self.check_router_recreation(router, router_settings)
- def test_create_router_new_user_to_admin_project(self):
+ def test_create_router_new_user_as_admin_project(self):
"""
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)
+ project_name=self.os_creds.project_name)
self.router_creator = create_router.OpenStackRouter(
- self.os_creds, router_settings)
+ self.admin_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(
+ self.neutron, router_settings=router_settings)
self.assertIsNotNone(router)
- self.assertTrue(verify_router_attributes(
- router, self.router_creator, ext_gateway=self.ext_net_name))
+ self.assertEqual(self.router_creator.get_router().id, router.id)
+
+ self.check_router_recreation(router, router_settings)
def test_create_delete_router(self):
"""
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(
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(
+ self.neutron, router_settings=self.router_settings)
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(
+ self.neutron, router_settings=self.router_settings)
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 = RouterConfig(
+ 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 = neutron_utils.get_router_by_name(self.neutron,
- router_settings.name)
+ router = neutron_utils.get_router(self.neutron,
+ router_settings=router_settings)
self.assertIsNotNone(router)
- self.assertTrue(verify_router_attributes(router, self.router_creator,
- admin_state=False))
+ self.assertEqual(self.router_creator.get_router(), router)
+
+ self.check_router_recreation(router, router_settings)
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 = RouterConfig(
+ 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(
+ self.neutron, router_settings=router_settings)
self.assertIsNotNone(router)
- self.assertTrue(verify_router_attributes(router, self.router_creator,
- admin_state=True))
+ self.assertEqual(self.router_creator.get_router(), router)
+
+ self.check_router_recreation(router, router_settings)
def test_create_router_private_network(self):
"""
Test creation of a router connected with two private networks and no
external gateway
"""
- network_settings1 = NetworkSettings(
+ network_settings1 = NetworkConfig(
name=self.guid + '-pub-net1',
subnet_settings=[
- create_network.SubnetSettings(
+ create_network.SubnetConfig(
cidr=cidr1, name=self.guid + '-pub-subnet1',
gateway_ip=static_gateway_ip1)])
- network_settings2 = NetworkSettings(
+ network_settings2 = NetworkConfig(
name=self.guid + '-pub-net2',
subnet_settings=[
- create_network.SubnetSettings(
+ create_network.SubnetConfig(
cidr=cidr2, name=self.guid + '-pub-subnet2',
gateway_ip=static_gateway_ip2)])
self.network_creator2.create()
port_settings = [
- create_network.PortSettings(
+ create_network.PortConfig(
name=self.guid + '-port1',
ip_addrs=[{
'subnet_name':
network_settings1.subnet_settings[0].name,
'ip': static_gateway_ip1
}],
- network_name=network_settings1.name,
- project_name=self.os_creds.project_name),
- create_network.PortSettings(
+ network_name=network_settings1.name),
+ create_network.PortConfig(
name=self.guid + '-port2',
ip_addrs=[{
'subnet_name': network_settings2.subnet_settings[0].name,
'ip': static_gateway_ip2
}],
- network_name=network_settings2.name,
- project_name=self.os_creds.project_name)]
+ 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 = 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()
- router = neutron_utils.get_router_by_name(self.neutron,
- router_settings.name)
+ router = neutron_utils.get_router(
+ self.neutron, router_settings=router_settings)
+
+ self.assertEqual(router, self.router_creator.get_router())
- self.assertTrue(verify_router_attributes(router, self.router_creator))
+ # Instantiate second identical creator to ensure a second router
+ # has not been created
+ router_creator2 = create_router.OpenStackRouter(
+ self.os_creds, router_settings)
+ router2 = router_creator2.create()
+ self.assertIsNotNone(self.router_creator.get_router(), router2)
+
+ self.check_router_recreation(router2, router_settings)
def test_create_router_external_network(self):
"""
Test creation of a router connected to an external network and a
private network.
"""
- network_settings = NetworkSettings(
+ network_settings = NetworkConfig(
name=self.guid + '-pub-net1',
subnet_settings=[
- create_network.SubnetSettings(
+ create_network.SubnetConfig(
cidr=cidr1, name=self.guid + '-pub-subnet1',
gateway_ip=static_gateway_ip1)])
self.network_creator1 = OpenStackNetwork(self.os_creds,
self.network_creator1.create()
port_settings = [
- create_network.PortSettings(
+ 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,
- project_name=self.os_creds.project_name)]
+ network_name=network_settings.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(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(
+ self.neutron, router_settings=router_settings)
- self.assertTrue(verify_router_attributes(
- router, self.router_creator, ext_gateway=self.ext_net_name))
+ self.assertEquals(router, self.router_creator.get_router())
+
+ self.check_router_recreation(router, router_settings)
+
+ def check_router_recreation(self, router, orig_settings):
+ """
+ Validates the derived RouterConfig with the original
+ :param router: the Router domain object to test
+ :param orig_settings: the original RouterConfig object that was
+ responsible for creating the router
+ :return: the derived RouterConfig object
+ """
+ derived_settings = settings_utils.create_router_config(
+ self.neutron, router)
+ self.assertIsNotNone(derived_settings)
+ self.assertEqual(
+ orig_settings.enable_snat, derived_settings.enable_snat)
+ self.assertEqual(orig_settings.external_gateway,
+ derived_settings.external_gateway)
+ self.assertEqual(orig_settings.name, derived_settings.name)
+ self.assertEqual(orig_settings.internal_subnets,
+ derived_settings.internal_subnets)
+
+ if orig_settings.external_gateway:
+ self.assertEqual(len(orig_settings.port_settings),
+ len(derived_settings.port_settings))
+ else:
+ self.assertEqual(len(orig_settings.port_settings),
+ len(derived_settings.port_settings))
+
+ if len(orig_settings.port_settings) > 0:
+ self.assertEqual(orig_settings.port_settings[0].name,
+ derived_settings.port_settings[0].name)
+
+ if len(orig_settings.port_settings) > 1:
+ self.assertEqual(orig_settings.port_settings[1].name,
+ derived_settings.port_settings[1].name)
+
+ return derived_settings
class CreateRouterNegativeTests(OSIntegrationTestCase):
"""
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()
-
-
-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 of type snaps.domain.Router
- :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 ext_gateway: None if router is not connected to external gateway
- :return:
- """
-
- router = router_creator.get_router()
-
- if not router_operational:
- return False
- elif not router_creator:
- return False
- elif not (router_operational.name == router_creator.router_settings.name):
- return False
- elif not (router_operational.id == router.id):
- return False
- elif not (router_operational.status == router.status):
- return False
- elif not (router_operational.tenant_id == router.tenant_id):
- return False
- elif not (admin_state == router_operational.admin_state_up):
- return False
- elif (ext_gateway is None) and \
- (router_operational.external_gateway_info is not None):
- return False
- elif ext_gateway is not None:
- if router_operational.external_gateway_info is None:
- return False
- return True
Code Review / snaps.git / blobdiff