197 lines
8.6 KiB
Python
197 lines
8.6 KiB
Python
# Copyright (c) 2013 OpenStack Foundation.
|
|
# 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 unittest import mock
|
|
|
|
from neutron_lib.api.definitions import external_net as extnet_apidef
|
|
from neutron_lib import constants
|
|
from neutron_lib import context
|
|
from neutron_lib.plugins import constants as plugin_constants
|
|
from neutron_lib.plugins import directory
|
|
from oslo_utils import uuidutils
|
|
import testtools
|
|
from webob import exc
|
|
|
|
from neutron.db import external_net_db
|
|
from neutron.db import models_v2
|
|
from neutron.tests.unit.api.v2 import test_base
|
|
from neutron.tests.unit.db import test_db_base_plugin_v2
|
|
|
|
|
|
_uuid = uuidutils.generate_uuid
|
|
_get_path = test_base._get_path
|
|
|
|
|
|
class ExtNetTestExtensionManager(object):
|
|
|
|
def get_resources(self):
|
|
return []
|
|
|
|
def get_actions(self):
|
|
return []
|
|
|
|
def get_request_extensions(self):
|
|
return []
|
|
|
|
|
|
class ExtNetDBTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
|
|
|
|
def _create_network(self, fmt, name, admin_state_up, **kwargs):
|
|
"""Override the routine for allowing the router:external attribute."""
|
|
# attributes containing a colon should be passed with
|
|
# a double underscore
|
|
new_args = dict(zip(map(lambda x: x.replace('__', ':'), kwargs),
|
|
kwargs.values()))
|
|
arg_list = new_args.pop('arg_list', ()) + (extnet_apidef.EXTERNAL,)
|
|
return super(ExtNetDBTestCase, self)._create_network(
|
|
fmt, name, admin_state_up, arg_list=arg_list, **new_args)
|
|
|
|
def setUp(self):
|
|
plugin = 'neutron.tests.unit.extensions.test_l3.TestNoL3NatPlugin'
|
|
ext_mgr = ExtNetTestExtensionManager()
|
|
super(ExtNetDBTestCase, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
|
|
|
|
def _set_net_external(self, net_id):
|
|
self._update('networks', net_id,
|
|
{'network': {extnet_apidef.EXTERNAL: True}})
|
|
|
|
def test_list_nets_external(self):
|
|
with self.network() as n1:
|
|
self._set_net_external(n1['network']['id'])
|
|
with self.network():
|
|
body = self._list('networks')
|
|
self.assertEqual(2, len(body['networks']))
|
|
|
|
body = self._list('networks',
|
|
query_params="%s=True" %
|
|
extnet_apidef.EXTERNAL)
|
|
self.assertEqual(1, len(body['networks']))
|
|
|
|
body = self._list('networks',
|
|
query_params="%s=False" %
|
|
extnet_apidef.EXTERNAL)
|
|
self.assertEqual(1, len(body['networks']))
|
|
|
|
def test_list_nets_external_pagination(self):
|
|
if self._skip_native_pagination:
|
|
self.skipTest("Skip test for not implemented pagination feature")
|
|
with self.network(name='net1') as n1, self.network(name='net3') as n3:
|
|
self._set_net_external(n1['network']['id'])
|
|
self._set_net_external(n3['network']['id'])
|
|
with self.network(name='net2') as n2:
|
|
self._test_list_with_pagination(
|
|
'network', (n1, n3), ('name', 'asc'), 1, 3,
|
|
query_params='router:external=True')
|
|
self._test_list_with_pagination(
|
|
'network', (n2, ), ('name', 'asc'), 1, 2,
|
|
query_params='router:external=False')
|
|
|
|
def test_get_network_succeeds_without_filter(self):
|
|
plugin = directory.get_plugin()
|
|
ctx = context.Context(None, None, is_admin=True)
|
|
result = plugin.get_networks(ctx, filters=None)
|
|
self.assertEqual([], result)
|
|
|
|
def test_update_network_set_external_non_admin_fails(self):
|
|
# Assert that a non-admin user cannot update the
|
|
# router:external attribute
|
|
with self.network(tenant_id='noadmin') as network:
|
|
data = {'network': {'router:external': True}}
|
|
req = self.new_update_request('networks',
|
|
data,
|
|
network['network']['id'])
|
|
req.environ['neutron.context'] = context.Context('', 'noadmin')
|
|
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 = directory.get_plugin()
|
|
result = plugin.get_networks(ctx)
|
|
self.assertFalse(result[0]['shared'])
|
|
|
|
def test_network_filter_hook_admin_context(self):
|
|
ctx = context.Context(None, None, is_admin=True)
|
|
model = models_v2.Network
|
|
conditions = external_net_db._network_filter_hook(ctx, model, [])
|
|
self.assertEqual([], conditions)
|
|
|
|
def test_network_filter_hook_nonadmin_context(self):
|
|
ctx = context.Context('edinson', 'cavani')
|
|
model = models_v2.Network
|
|
txt = ("networkrbacs.action = :action_1 AND "
|
|
"networkrbacs.target_tenant = :target_tenant_1 OR "
|
|
"networkrbacs.target_tenant = :target_tenant_2")
|
|
conditions = external_net_db._network_filter_hook(ctx, model, [])
|
|
self.assertEqual(conditions.__str__(), txt)
|
|
# Try to concatenate conditions
|
|
txt2 = (txt.replace('tenant_1', 'tenant_3').
|
|
replace('tenant_2', 'tenant_4').
|
|
replace('action_1', 'action_2'))
|
|
conditions = external_net_db._network_filter_hook(ctx, model,
|
|
conditions)
|
|
self.assertEqual(conditions.__str__(), "%s OR %s" % (txt, txt2))
|
|
|
|
def test_create_port_external_network_non_admin_fails(self):
|
|
with self.network(router__external=True) as ext_net:
|
|
with self.subnet(network=ext_net) as ext_subnet:
|
|
with testtools.ExpectedException(
|
|
exc.HTTPClientError) as ctx_manager:
|
|
with self.port(subnet=ext_subnet,
|
|
set_context='True',
|
|
tenant_id='noadmin'):
|
|
pass
|
|
self.assertEqual(403, ctx_manager.exception.code)
|
|
|
|
def test_create_port_external_network_admin_succeeds(self):
|
|
with self.network(router__external=True) as ext_net:
|
|
with self.subnet(network=ext_net) as ext_subnet:
|
|
with self.port(subnet=ext_subnet) as port:
|
|
self.assertEqual(port['port']['network_id'],
|
|
ext_net['network']['id'])
|
|
|
|
def test_create_external_network_non_admin_fails(self):
|
|
with testtools.ExpectedException(exc.HTTPClientError) as ctx_manager:
|
|
with self.network(router__external=True,
|
|
set_context='True',
|
|
tenant_id='noadmin'):
|
|
pass
|
|
self.assertEqual(403, ctx_manager.exception.code)
|
|
|
|
def test_create_external_network_admin_succeeds(self):
|
|
with self.network(router__external=True) as ext_net:
|
|
self.assertTrue(ext_net['network'][extnet_apidef.EXTERNAL])
|
|
|
|
def test_delete_network_check_disassociated_floatingips(self):
|
|
l3_mock = mock.Mock()
|
|
directory.add_plugin(plugin_constants.L3, l3_mock)
|
|
with self.network() as net:
|
|
req = self.new_delete_request('networks', net['network']['id'])
|
|
res = req.get_response(self.api)
|
|
self.assertEqual(exc.HTTPNoContent.code, res.status_int)
|
|
(l3_mock.delete_disassociated_floatingips
|
|
.assert_called_once_with(mock.ANY, net['network']['id']))
|