diff options
| -rw-r--r-- | neutron/db/l3_dvr_db.py | 26 | ||||
| -rw-r--r-- | neutron/tests/functional/services/l3_router/test_l3_dvr_router_plugin.py | 57 | ||||
| -rw-r--r-- | neutron/tests/unit/db/test_l3_dvr_db.py | 30 |
3 files changed, 101 insertions, 12 deletions
diff --git a/neutron/db/l3_dvr_db.py b/neutron/db/l3_dvr_db.py index b847dc9b14..deb6c417d3 100644 --- a/neutron/db/l3_dvr_db.py +++ b/neutron/db/l3_dvr_db.py @@ -727,18 +727,20 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin, msg = _("Unable to create the SNAT Interface Port") raise n_exc.BadRequest(resource='router', msg=msg) - with context.session.begin(subtransactions=True): - router_port = l3_models.RouterPort( - port_id=snat_port['id'], - router_id=router.id, - port_type=const.DEVICE_OWNER_ROUTER_SNAT - ) - context.session.add(router_port) - - if do_pop: - return self._populate_mtu_and_subnets_for_ports(context, - [snat_port]) - return snat_port + with p_utils.delete_port_on_error( + self._core_plugin, context.elevated(), snat_port['id']): + with context.session.begin(subtransactions=True): + router_port = l3_models.RouterPort( + port_id=snat_port['id'], + router_id=router.id, + port_type=const.DEVICE_OWNER_ROUTER_SNAT + ) + context.session.add(router_port) + + if do_pop: + return self._populate_mtu_and_subnets_for_ports(context, + [snat_port]) + return snat_port def _create_snat_intf_ports_if_not_exists(self, context, router): """Function to return the snat interface port list. diff --git a/neutron/tests/functional/services/l3_router/test_l3_dvr_router_plugin.py b/neutron/tests/functional/services/l3_router/test_l3_dvr_router_plugin.py index df4d705509..f5ca1e72c8 100644 --- a/neutron/tests/functional/services/l3_router/test_l3_dvr_router_plugin.py +++ b/neutron/tests/functional/services/l3_router/test_l3_dvr_router_plugin.py @@ -82,6 +82,63 @@ class L3DvrTestCase(L3DvrTestCaseBase): self.l3_plugin._get_agent_gw_ports_exist_for_network( self.context, 'network_id', 'host', 'agent_id')) + def test_csnat_ports_are_created_and_deleted_based_on_router_subnet(self): + kwargs = {'arg_list': (external_net.EXTERNAL,), + external_net.EXTERNAL: True} + net1 = self._make_network(self.fmt, 'net1', True) + subnet1 = self._make_subnet( + self.fmt, net1, '10.1.0.1', '10.1.0.0/24', enable_dhcp=True) + subnet2 = self._make_subnet( + self.fmt, net1, '10.2.0.1', '10.2.0.0/24', enable_dhcp=True) + ext_net = self._make_network(self.fmt, 'ext_net', True, **kwargs) + self._make_subnet( + self.fmt, ext_net, '20.0.0.1', '20.0.0.0/24', enable_dhcp=True) + # Create first router and add an interface + router1 = self._create_router() + ext_net_id = ext_net['network']['id'] + net1_id = net1['network']['id'] + # Set gateway to router + self.l3_plugin._update_router_gw_info( + self.context, router1['id'], + {'network_id': ext_net_id}) + # Now add router interface (subnet1) from net1 to router + self.l3_plugin.add_router_interface( + self.context, router1['id'], + {'subnet_id': subnet1['subnet']['id']}) + # Now add router interface (subnet2) from net1 to router + self.l3_plugin.add_router_interface( + self.context, router1['id'], + {'subnet_id': subnet2['subnet']['id']}) + # Now check the valid snat interfaces passed to the agent + snat_router_intfs = self.l3_plugin._get_snat_sync_interfaces( + self.context, [router1['id']]) + self.assertEqual(2, len(snat_router_intfs[router1['id']])) + # Also make sure that there are no csnat ports created and + # left over. + csnat_ports = self.core_plugin.get_ports( + self.context, filters={ + 'network_id': [net1_id], + 'device_owner': [constants.DEVICE_OWNER_ROUTER_SNAT]}) + self.assertEqual(2, len(csnat_ports)) + # Now remove router interface (subnet1) from net1 to router + self.l3_plugin.remove_router_interface( + self.context, router1['id'], + {'subnet_id': subnet1['subnet']['id']}) + # Now remove router interface (subnet2) from net1 to router + self.l3_plugin.remove_router_interface( + self.context, router1['id'], + {'subnet_id': subnet2['subnet']['id']}) + snat_router_intfs = self.l3_plugin._get_snat_sync_interfaces( + self.context, [router1['id']]) + self.assertEqual(0, len(snat_router_intfs[router1['id']])) + # Also make sure that there are no csnat ports created and + # left over. + csnat_ports = self.core_plugin.get_ports( + self.context, filters={ + 'network_id': [net1_id], + 'device_owner': [constants.DEVICE_OWNER_ROUTER_SNAT]}) + self.assertEqual(0, len(csnat_ports)) + def _test_remove_router_interface_leaves_snat_intact(self, by_subnet): with self.subnet() as subnet1, \ self.subnet(cidr='20.0.0.0/24') as subnet2: diff --git a/neutron/tests/unit/db/test_l3_dvr_db.py b/neutron/tests/unit/db/test_l3_dvr_db.py index 4ff3ab1576..4461f15e8f 100644 --- a/neutron/tests/unit/db/test_l3_dvr_db.py +++ b/neutron/tests/unit/db/test_l3_dvr_db.py @@ -28,6 +28,7 @@ from neutron.db import agents_db from neutron.db import common_db_mixin from neutron.db import l3_agentschedulers_db from neutron.db import l3_dvr_db +from neutron.db.models import l3 as l3_models from neutron.extensions import l3 from neutron.extensions import portbindings from neutron.tests.unit.db import test_db_base_plugin_v2 @@ -793,6 +794,35 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase): self.assertEqual(const.DEVICE_OWNER_ROUTER_GW, router_ports[0]['device_owner']) + def test_csnat_port_not_created_on_RouterPort_update_exception(self): + router_dict = {'name': 'test_router', 'admin_state_up': True, + 'distributed': True} + router = self._create_router(router_dict) + with self.network() as net_ext,\ + self.subnet() as subnet: + ext_net_id = net_ext['network']['id'] + self.core_plugin.update_network( + self.ctx, ext_net_id, + {'network': {'router:external': True}}) + self.mixin.update_router( + self.ctx, router['id'], + {'router': {'external_gateway_info': + {'network_id': ext_net_id}}}) + net_id = subnet['subnet']['network_id'] + with mock.patch.object( + l3_models, 'RouterPort', side_effect=Exception): + self.assertRaises( + l3.RouterInterfaceAttachmentConflict, + self.mixin.add_router_interface, + self.ctx, router['id'], + {'subnet_id': subnet['subnet']['id']}) + filters = { + 'network_id': [net_id], + 'device_owner': [const.DEVICE_OWNER_ROUTER_SNAT] + } + router_ports = self.core_plugin.get_ports(self.ctx, filters) + self.assertEqual(0, len(router_ports)) + def test_add_router_interface_by_port_failure(self): router_dict = {'name': 'test_router', 'admin_state_up': True, |