X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=snaps%2Fopenstack%2Ftests%2Fcreate_router_tests.py;h=4e47f6de0c4f9b8a4ffe139463a33d3fc9608801;hb=c711acf8ae3e4ad6f746500747857bcc9fd6f7be;hp=6f37445fe85267d0bac42123c3ae0a11ba14e69c;hpb=463dc2cd59bb26150f15e57a0e94fafbbf8634d4;p=snaps.git diff --git a/snaps/openstack/tests/create_router_tests.py b/snaps/openstack/tests/create_router_tests.py index 6f37445..4e47f6d 100644 --- a/snaps/openstack/tests/create_router_tests.py +++ b/snaps/openstack/tests/create_router_tests.py @@ -15,15 +15,14 @@ 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' @@ -39,11 +38,11 @@ class RouterSettingsUnitTests(unittest.TestCase): """ 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): @@ -51,9 +50,8 @@ class RouterSettingsUnitTests(unittest.TestCase): 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)) @@ -66,9 +64,8 @@ class RouterSettingsUnitTests(unittest.TestCase): 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)) @@ -77,17 +74,16 @@ class RouterSettingsUnitTests(unittest.TestCase): 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)) @@ -98,8 +94,7 @@ class RouterSettingsUnitTests(unittest.TestCase): 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'}}]}) @@ -108,14 +103,13 @@ class RouterSettingsUnitTests(unittest.TestCase): 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): @@ -155,26 +149,27 @@ 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) @@ -182,53 +177,55 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): 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 @@ -238,53 +235,55 @@ 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 = 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)]) @@ -297,34 +296,32 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): 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.assertTrue(verify_router_attributes(router, self.router_creator)) + self.assertEqual(router, self.router_creator.get_router()) # Instantiate second identical creator to ensure a second router # has not been created @@ -333,15 +330,17 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): 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, @@ -349,26 +348,62 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): 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.assertEquals(router, self.router_creator.get_router()) - self.assertTrue(verify_router_attributes( - router, self.router_creator, ext_gateway=self.ext_net_name)) + 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): @@ -398,8 +433,8 @@ 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) @@ -409,48 +444,10 @@ class CreateRouterNegativeTests(OSIntegrationTestCase): """ 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