diff options
Diffstat (limited to 'snaps/openstack/tests/create_router_tests.py')
| -rw-r--r-- | snaps/openstack/tests/create_router_tests.py | 481 |
1 files changed, 449 insertions, 32 deletions
diff --git a/snaps/openstack/tests/create_router_tests.py b/snaps/openstack/tests/create_router_tests.py index 09471a3..a305cf8 100644 --- a/snaps/openstack/tests/create_router_tests.py +++ b/snaps/openstack/tests/create_router_tests.py @@ -15,19 +15,23 @@ 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, RouterCreationError) +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 +from snaps.openstack.utils import neutron_utils, settings_utils, keystone_utils __author__ = 'mmakati' cidr1 = '10.200.201.0/24' cidr2 = '10.200.202.0/24' +cidr3 = '10.200.203.0/24' static_gateway_ip1 = '10.200.201.1' static_gateway_ip2 = '10.200.202.1' @@ -128,7 +132,8 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): self.router_creator = None self.network_creator1 = None self.network_creator2 = None - self.neutron = neutron_utils.neutron_client(self.os_creds) + self.neutron = neutron_utils.neutron_client( + self.os_creds, self.os_session) def tearDown(self): """ @@ -152,12 +157,13 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): 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) @@ -177,32 +183,34 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): self.admin_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.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 = 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(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.assertEqual(self.router_creator.get_router().id, router.id) self.check_router_recreation(router, router_settings) @@ -219,18 +227,67 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): 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 self.router_creator.clean() + def test_create_with_internal_sub(self): + """ + Test internal_subnets works. + """ + network_settings1 = 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_creator1 = OpenStackNetwork(self.os_creds, + network_settings1) + + self.network_creator1.create() + self.router_settings = RouterConfig( + name=self.guid + '-pub-router', external_gateway=self.ext_net_name, + internal_subnets=[network_settings1.subnet_settings[0].name]) + + self.router_creator = create_router.OpenStackRouter( + self.os_creds, self.router_settings) + created_router = self.router_creator.create() + self.assertIsNotNone(created_router) + + def test_create_with_invalid_internal_sub(self): + """ + Test adding an internal subnet owned by admin which should fail. + """ + network_settings1 = 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_creator1 = OpenStackNetwork(self.admin_os_creds, + network_settings1) + + self.network_creator1.create() + self.router_settings = RouterConfig( + name=self.guid + '-pub-router', external_gateway=self.ext_net_name, + internal_subnets=[network_settings1.subnet_settings[0].name]) + + self.router_creator = create_router.OpenStackRouter( + self.os_creds, self.router_settings) + + with self.assertRaises(RouterCreationError): + self.router_creator.create() + def test_create_router_admin_state_false(self): """ Test creation of a basic router with admin state down. @@ -242,8 +299,9 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): 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) @@ -262,7 +320,8 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): 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) @@ -303,25 +362,24 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): network_settings1.subnet_settings[0].name, 'ip': static_gateway_ip1 }], - network_name=network_settings1.name, - project_name=self.os_creds.project_name), + 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 = 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.OpenStackRouter( + self.os_creds, router_settings) 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()) @@ -355,8 +413,7 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): 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 = RouterConfig( name=self.guid + '-pub-router', external_gateway=self.ext_net_name, @@ -366,12 +423,58 @@ class CreateRouterSuccessTests(OSIntegrationTestCase): 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()) self.check_router_recreation(router, router_settings) + def test_create_router_with_ext_port(self): + """ + Test creation of a router with a port to an external network as an + 'admin' user. + """ + port_settings = [ + create_network.PortConfig( + name=self.guid + '-port1', + network_name=self.ext_net_name)] + + router_settings = RouterConfig( + name=self.guid + '-pub-router', port_settings=port_settings) + self.router_creator = create_router.OpenStackRouter( + self.admin_os_creds, router_settings) + self.router_creator.create() + + admin_neutron = neutron_utils.neutron_client( + self.admin_os_creds, self.admin_os_session) + admin_keystone = keystone_utils.keystone_client( + self.admin_os_creds, self.admin_os_session) + router = neutron_utils.get_router( + admin_neutron, admin_keystone, router_settings=router_settings, + project_name=self.admin_os_creds.project_name) + + self.assertIsNotNone(router) + self.assertEquals(router, self.router_creator.get_router()) + + ext_net = neutron_utils.get_network( + admin_neutron, admin_keystone, network_name=self.ext_net_name) + + self.assertIsNotNone(ext_net) + self.assertIsNotNone(router.port_subnets) + + id_found = False + for port, subnets in router.port_subnets: + self.assertIsNotNone(subnets) + self.assertIsNotNone(port) + + if ext_net.id == port.network_id: + id_found = True + for subnet in subnets: + self.assertIsNotNone(subnet) + self.assertEqual(ext_net.id, subnet.network_id) + self.assertTrue(id_found) + def check_router_recreation(self, router, orig_settings): """ Validates the derived RouterConfig with the original @@ -421,6 +524,7 @@ class CreateRouterNegativeTests(OSIntegrationTestCase): super(self.__class__, self).__start__() self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4()) + self.network_creator = None self.router_creator = None def tearDown(self): @@ -430,6 +534,9 @@ class CreateRouterNegativeTests(OSIntegrationTestCase): 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): @@ -454,3 +561,313 @@ class CreateRouterNegativeTests(OSIntegrationTestCase): 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, self.os_session) + + 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, self.os_session) + + 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() + + +class CreateRouterSharedNetworksTests(OSIntegrationTestCase): + """ + Class for testing routers external and/or shared networks + """ + + 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 + + ext_network_settings = NetworkConfig( + name=self.guid + '-ext-net', + external=True, + subnet_settings=[ + create_network.SubnetConfig( + cidr=cidr1, name=self.guid + '-ext-subnet1')]) + self.ext_network_creator = OpenStackNetwork( + self.admin_os_creds, ext_network_settings) + self.ext_network_creator.create() + + shared_network_settings = NetworkConfig( + name=self.guid + '-shared-net', + shared=True, + subnet_settings=[ + create_network.SubnetConfig( + cidr=cidr2, name=self.guid + '-shared-subnet1')]) + self.shared_network_creator = OpenStackNetwork( + self.admin_os_creds, shared_network_settings) + self.shared_network_creator.create() + + overlay_network_settings = NetworkConfig( + name=self.guid + '-overlay-net', + subnet_settings=[ + create_network.SubnetConfig( + cidr=cidr3, name=self.guid + '-overlay-subnet1')]) + self.overlay_network_creator = OpenStackNetwork( + self.os_creds, overlay_network_settings) + self.overlay_network_creator.create() + + self.neutron = neutron_utils.neutron_client( + self.os_creds, self.os_session) + + def tearDown(self): + """ + Cleans the remote OpenStack objects used for router testing + """ + if self.router_creator: + self.router_creator.clean() + + if self.overlay_network_creator: + self.overlay_network_creator.clean() + + if self.shared_network_creator: + self.shared_network_creator.clean() + + if self.ext_network_creator: + self.ext_network_creator.clean() + + super(self.__class__, self).__clean__() + + def test_create_router_external(self): + """ + Test creation of a router with a custom external network created by + admin. + """ + router_settings = RouterConfig( + name=self.guid + '-pub-router', + external_gateway=self.ext_network_creator.get_network().name) + self.router_creator = create_router.OpenStackRouter( + self.os_creds, router_settings) + self.router_creator.create() + + def test_create_router_port_external(self): + """ + Test creation of a router with a port to an custom external network + created by admin. + """ + router_settings = RouterConfig( + name=self.guid + '-pub-router', + network_name=self.ext_network_creator.get_network().name) + self.router_creator = create_router.OpenStackRouter( + self.os_creds, router_settings) + self.router_creator.create() + + def test_create_router_port_shared(self): + """ + Test creation of a router with a port to an custom shared network + created by admin. + """ + port_settings = [ + create_network.PortConfig( + name=self.guid + '-port1', + network_name=self.shared_network_creator.get_network().name)] + + 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() |