From 161d465305777226a15ced334acfd06142597f85 Mon Sep 17 00:00:00 2001 From: Mike Kolesnik Date: Sun, 23 Mar 2014 15:48:42 +0200 Subject: [PATCH] Allow unsharing a network used as gateway/floatingip If network has ports that connect it as gateway to another network, such as an external network, or "floating ip" ports it should still be possible to "unshare" it since these ports represent usage of the network as an external network, not as a VM data network. Change-Id: Ic33ae7ac193e4fc9fe06a4cfd579a4aacf0e6354 Closes-bug: #1293184 (cherry picked from commit 7a2053c7a27bdec02e97b46736c5fcce5f2d053a) --- neutron/db/db_base_plugin_v2.py | 8 ++- neutron/tests/unit/test_db_plugin_level.py | 83 ++++++++++++++++++++++ 2 files changed, 90 insertions(+), 1 deletion(-) create mode 100644 neutron/tests/unit/test_db_plugin_level.py diff --git a/neutron/db/db_base_plugin_v2.py b/neutron/db/db_base_plugin_v2.py index 3397cd79db8..31ba7c8f2de 100644 --- a/neutron/db/db_base_plugin_v2.py +++ b/neutron/db/db_base_plugin_v2.py @@ -20,6 +20,7 @@ import weakref import netaddr from oslo.config import cfg +from sqlalchemy import and_ from sqlalchemy import event from sqlalchemy import orm from sqlalchemy.orm import exc @@ -822,7 +823,12 @@ class NeutronDbPluginV2(neutron_plugin_base_v2.NeutronPluginBaseV2, return ports = self._model_query( context, models_v2.Port).filter( - models_v2.Port.network_id == id) + and_( + models_v2.Port.network_id == id, + models_v2.Port.device_owner != + constants.DEVICE_OWNER_ROUTER_GW, + models_v2.Port.device_owner != + constants.DEVICE_OWNER_FLOATINGIP)) subnets = self._model_query( context, models_v2.Subnet).filter( models_v2.Subnet.network_id == id) diff --git a/neutron/tests/unit/test_db_plugin_level.py b/neutron/tests/unit/test_db_plugin_level.py new file mode 100644 index 00000000000..3fe2be1fa22 --- /dev/null +++ b/neutron/tests/unit/test_db_plugin_level.py @@ -0,0 +1,83 @@ +# Copyright (c) 2014 Red Hat, Inc. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +from neutron.api.v2 import attributes +from neutron.common import constants +from neutron.common import exceptions as n_exc +from neutron import context +from neutron import manager +from neutron.tests import base +from neutron.tests.unit import test_db_plugin +from neutron.tests.unit import testlib_api + + +class TestNetworks(base.BaseTestCase): + def setUp(self): + super(TestNetworks, self).setUp() + self._tenant_id = 'test-tenant' + + # Update the plugin + self.setup_coreplugin(test_db_plugin.DB_PLUGIN_KLASS) + + def _create_network(self, plugin, ctx, shared=True): + network = {'network': {'name': 'net', + 'shared': shared, + 'admin_state_up': True, + 'tenant_id': self._tenant_id}} + created_network = plugin.create_network(ctx, network) + return (network, created_network['id']) + + def _create_port(self, plugin, ctx, net_id, device_owner, tenant_id): + port = {'port': {'name': 'port', + 'network_id': net_id, + 'mac_address': attributes.ATTR_NOT_SPECIFIED, + 'fixed_ips': attributes.ATTR_NOT_SPECIFIED, + 'admin_state_up': True, + 'device_id': 'device_id', + 'device_owner': device_owner, + 'tenant_id': tenant_id}} + plugin.create_port(ctx, port) + + def _test_update_shared_net_used(self, + device_owner, + expected_exception=None): + plugin = manager.NeutronManager.get_plugin() + ctx = context.get_admin_context() + network, net_id = self._create_network(plugin, ctx) + + self._create_port(plugin, + ctx, + net_id, + device_owner, + self._tenant_id + '1') + + network['network']['shared'] = False + + if (expected_exception): + with testlib_api.ExpectedException(expected_exception): + plugin.update_network(ctx, net_id, network) + else: + plugin.update_network(ctx, net_id, network) + + def test_update_shared_net_used_fails(self): + self._test_update_shared_net_used('', n_exc.InvalidSharedSetting) + + def test_update_shared_net_used_as_router_gateway(self): + self._test_update_shared_net_used( + constants.DEVICE_OWNER_ROUTER_GW) + + def test_update_shared_net_used_by_floating_ip(self): + self._test_update_shared_net_used( + constants.DEVICE_OWNER_FLOATINGIP)