Merge "Allow admin to create port on networks of different projects" into stable/pike

This commit is contained in:
Zuul 2018-02-05 05:26:19 +00:00 committed by Gerrit Code Review
commit b9657c5129
4 changed files with 61 additions and 10 deletions

View File

@ -863,12 +863,23 @@ def network_get(request, network_id, expand_subnet=True, **params):
network = neutronclient(request).show_network(network_id,
**params).get('network')
if expand_subnet:
if request.user.tenant_id == network['tenant_id'] or network['shared']:
# NOTE(amotoki): There are some cases where a user has no permission
# to get subnet details, but the condition is complicated. We first
# try to fetch subnet details. If successful, the subnet details are
# set to network['subnets'] as a list of "Subent" object.
# If NotFound exception is returned by neutron, network['subnets'] is
# left untouched and a list of subnet IDs are stored.
# Neutron returns NotFound exception if a request user has enough
# permission to access a requested resource, so we catch only
# NotFound exception here.
try:
# Since the number of subnets per network must be small,
# call subnet_get() for each subnet instead of calling
# subnet_list() once.
network['subnets'] = [subnet_get(request, sid)
for sid in network['subnets']]
except neutron_exc.NotFound:
pass
return Network(network)

View File

@ -18,6 +18,8 @@ from django.conf import settings
from django.core.urlresolvers import reverse
from django.utils.translation import ugettext_lazy as _
from neutronclient.common import exceptions as neutron_exc
from horizon import exceptions
from horizon import forms
from horizon import messages
@ -51,7 +53,6 @@ class CreatePort(forms.SelfHandlingForm):
specify_ip = forms.ThemableChoiceField(
label=_("Specify IP address or subnet"),
help_text=_("To specify a subnet or a fixed IP, select any options."),
initial=False,
required=False,
choices=[('', _("Unspecified")),
('subnet_id', _("Subnet")),
@ -119,8 +120,17 @@ class CreatePort(forms.SelfHandlingForm):
except Exception:
return []
# NOTE(amotoki): When a user cannot retrieve a subnet info,
# subnet ID is stored in network.subnets field.
# If so, we skip such subnet as subnet choices.
# This happens usually for external networks.
# TODO(amotoki): Ideally it is better to disable/hide
# Create Port button in the port table, but as of Pike
# the default neutron policy.json for "create_port" is empty
# and there seems no appropriate policy. This is a dirty hack.
return [(subnet.id, '%s %s' % (subnet.name_or_id, subnet.cidr))
for subnet in network.subnets]
for subnet in network.subnets
if isinstance(subnet, api.neutron.Subnet)]
def handle(self, request, data):
try:
@ -158,8 +168,13 @@ class CreatePort(forms.SelfHandlingForm):
except Exception as e:
LOG.info('Failed to create a port for network %(id)s: %(exc)s',
{'id': self.initial['network_id'], 'exc': e})
msg = (_('Failed to create a port for network %s')
% self.initial['network_id'])
if isinstance(e, neutron_exc.Forbidden):
msg = (_('You are not allowed to create a port '
'for network %s.')
% self.initial['network_id'])
else:
msg = (_('Failed to create a port for network %s')
% self.initial['network_id'])
redirect = reverse(self.failure_url,
args=(self.initial['network_id'],))
exceptions.handle(request, msg, redirect=redirect)

View File

@ -356,18 +356,25 @@ class NetworkPortTests(test.TestCase):
self.assertRedirectsNoFollow(res, url)
self.assertMessageCount(success=1)
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',)})
def test_port_create_get(self):
self._test_port_create_get()
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',)})
def test_port_create_get_with_mac_learning(self):
self._test_port_create_get(mac_learning=True)
def _test_port_create_get(self, mac_learning=False, binding=False):
def test_port_create_get_without_subnet_detail(self):
self._test_port_create_get(no_subnet_detail=True)
@test.create_stubs({api.neutron: ('network_get',
'is_extension_supported',)})
def _test_port_create_get(self, mac_learning=False, binding=False,
no_subnet_detail=False):
network = self.networks.first()
if no_subnet_detail:
# Set Subnet UUID list to network.subnets to emulate
# a situation where a user has no enough permission to
# retrieve subnet details.
network.subnets = [s.id for s in network.subnets]
api.neutron.network_get(IsA(http.HttpRequest),
network.id) \
.AndReturn(self.networks.first())

View File

@ -175,6 +175,24 @@ class NeutronApiTests(test.APITestCase):
ret_val = api.neutron.network_get(self.request, network_id)
self.assertIsInstance(ret_val, api.neutron.Network)
self.assertEqual(1, len(ret_val['subnets']))
self.assertIsInstance(ret_val['subnets'][0], api.neutron.Subnet)
def test_network_get_with_subnet_get_notfound(self):
network = {'network': self.api_networks.first()}
network_id = self.api_networks.first()['id']
subnet_id = self.api_networks.first()['subnets'][0]
neutronclient = self.stub_neutronclient()
neutronclient.show_network(network_id).AndReturn(network)
neutronclient.show_subnet(subnet_id).AndRaise(neutron_exc.NotFound)
self.mox.ReplayAll()
ret_val = api.neutron.network_get(self.request, network_id)
self.assertIsInstance(ret_val, api.neutron.Network)
self.assertEqual(1, len(ret_val['subnets']))
self.assertNotIsInstance(ret_val['subnets'][0], api.neutron.Subnet)
self.assertIsInstance(ret_val['subnets'][0], str)
def test_network_create(self):
network = {'network': self.api_networks.first()}