openstack
/
neutron
Code Issues Proposed changes

fix _validate_shared_update for dvr router ports 

When trying to update external network from shared to not,
_validate_shared_update will be called to validate ports
on external network to confirm that, except router ports,
there is no port from other tenants. If some other tenants
ports still exist on external network, shared attribute
updating will fail.

However, current _validate_shared_update will only ignore router
ports with device_owner such as DEVICE_OWNER_ROUTER_GW and
DEVICE_OWNER_FLOATINGIP when query ports, but indeed ports with
device_owner such as DEVICE_OWNER_AGENT_GW and DEVICE_OWNER_ROUTER_SNAT
should be skipped, otherwise an empty tenant_id like '' will cause
_validate_shared_update fails to work as expected.

Change-Id: Ibf5daf14192ce51d4e9ce10dc0c806ea230adec3
Closes-Bug: #1522482
This commit is contained in:
lzklibj 2015-12-04 02:34:59 +08:00 committed by Brian Haley
parent f4f4c57782
commit 08d720e1ca
2 changed files with 22 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
neutron/db/db_base_plugin_v2.py
View File

@ -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)

18
neutron/tests/unit/extensions/test_external_net.py
View File

@ -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)