diff --git a/neutron/db/db_base_plugin_v2.py b/neutron/db/db_base_plugin_v2.py index 54b00b5b8c5..02b941b8248 100644 --- a/neutron/db/db_base_plugin_v2.py +++ b/neutron/db/db_base_plugin_v2.py @@ -23,6 +23,7 @@ from oslo_utils import excutils from oslo_utils import uuidutils from sqlalchemy import and_ from sqlalchemy import event +from sqlalchemy import not_ from neutron._i18n import _, _LE, _LI from neutron.api.rpc.agentnotifiers import l3_rpc_agent_api @@ -208,13 +209,9 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon, if updated['shared'] == original.shared or updated['shared']: return ports = self._model_query( - context, models_v2.Port).filter( - 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)) + context, models_v2.Port).filter(models_v2.Port.network_id == id) + ports = ports.filter(not_(models_v2.Port.device_owner.startswith( + constants.DEVICE_OWNER_NETWORK_PREFIX))) subnets = self._model_query( context, models_v2.Subnet).filter( models_v2.Subnet.network_id == id) diff --git a/neutron/tests/unit/extensions/test_external_net.py b/neutron/tests/unit/extensions/test_external_net.py index 131fefd598c..3920e92a073 100644 --- a/neutron/tests/unit/extensions/test_external_net.py +++ b/neutron/tests/unit/extensions/test_external_net.py @@ -18,6 +18,7 @@ from oslo_utils import uuidutils import testtools from webob import exc +from neutron.common import constants from neutron import context from neutron.db import models_v2 from neutron.extensions import external_net as external_net @@ -112,6 +113,23 @@ class ExtNetDBTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase): res = req.get_response(self.api) self.assertEqual(exc.HTTPForbidden.code, res.status_int) + def test_update_network_external_net_with_ports_set_not_shared(self): + with self.network(router__external=True, shared=True) as ext_net,\ + self.subnet(network=ext_net) as ext_subnet, \ + self.port(subnet=ext_subnet, + tenant_id='', + device_owner=constants.DEVICE_OWNER_ROUTER_SNAT): + data = {'network': {'shared': False}} + req = self.new_update_request('networks', + data, + ext_net['network']['id']) + res = req.get_response(self.api) + self.assertEqual(exc.HTTPOk.code, res.status_int) + ctx = context.Context(None, None, is_admin=True) + plugin = manager.NeutronManager.get_plugin() + result = plugin.get_networks(ctx) + self.assertFalse(result[0]['shared']) + def test_network_filter_hook_admin_context(self): plugin = manager.NeutronManager.get_plugin() ctx = context.Context(None, None, is_admin=True)