Merge "fix _validate_shared_update for dvr router ports"

This commit is contained in:
Jenkins 2016-01-25 15:20:33 +00:00 committed by Gerrit Code Review
commit 74d48843f8
2 changed files with 22 additions and 7 deletions

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)

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)