neutron/neutron/tests/unit/extensions/test_address_scope.py

523 lines
26 KiB
Python

# Copyright (c) 2015 Red Hat, Inc.
#
# 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.
import contextlib
import mock
import netaddr
import webob.exc
from neutron.api.v2 import attributes as attr
from neutron.callbacks import events
from neutron.callbacks import registry
from neutron.callbacks import resources
from neutron.common import constants
from neutron import context
from neutron.db import address_scope_db
from neutron.db import db_base_plugin_v2
from neutron.extensions import address_scope as ext_address_scope
from neutron.tests.unit.db import test_db_base_plugin_v2
DB_PLUGIN_KLASS = ('neutron.tests.unit.extensions.test_address_scope.'
'AddressScopeTestPlugin')
class AddressScopeTestExtensionManager(object):
def get_resources(self):
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attr.RESOURCE_ATTRIBUTE_MAP.update(
ext_address_scope.RESOURCE_ATTRIBUTE_MAP)
return ext_address_scope.Address_scope.get_resources()
def get_actions(self):
return []
def get_request_extensions(self):
return []
class AddressScopeTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
def _create_address_scope(self, fmt, ip_version=constants.IP_VERSION_4,
expected_res_status=None, admin=False, **kwargs):
address_scope = {'address_scope': {}}
address_scope['address_scope']['ip_version'] = ip_version
for k, v in kwargs.items():
address_scope['address_scope'][k] = str(v)
address_scope_req = self.new_create_request('address-scopes',
address_scope, fmt)
if not admin:
neutron_context = context.Context('', kwargs.get('tenant_id',
self._tenant_id))
address_scope_req.environ['neutron.context'] = neutron_context
address_scope_res = address_scope_req.get_response(self.ext_api)
if expected_res_status:
self.assertEqual(expected_res_status, address_scope_res.status_int)
return address_scope_res
def _make_address_scope(self, fmt, ip_version, admin=False, **kwargs):
res = self._create_address_scope(fmt, ip_version,
admin=admin, **kwargs)
if res.status_int >= webob.exc.HTTPClientError.code:
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res)
@contextlib.contextmanager
def address_scope(self, ip_version=constants.IP_VERSION_4,
admin=False, **kwargs):
addr_scope = self._make_address_scope(self.fmt, ip_version,
admin, **kwargs)
yield addr_scope
def _test_create_address_scope(self, ip_version=constants.IP_VERSION_4,
admin=False, expected=None, **kwargs):
keys = kwargs.copy()
keys.setdefault('tenant_id', self._tenant_id)
with self.address_scope(ip_version,
admin=admin, **keys) as addr_scope:
keys['ip_version'] = ip_version
self._validate_resource(addr_scope, keys, 'address_scope')
if expected:
self._compare_resource(addr_scope, expected, 'address_scope')
return addr_scope
def _test_update_address_scope(self, addr_scope_id, data, admin=False,
expected=None, tenant_id=None):
update_req = self.new_update_request(
'address-scopes', data, addr_scope_id)
if not admin:
neutron_context = context.Context('', tenant_id or self._tenant_id)
update_req.environ['neutron.context'] = neutron_context
update_res = update_req.get_response(self.ext_api)
if expected:
addr_scope = self.deserialize(self.fmt, update_res)
self._compare_resource(addr_scope, expected, 'address_scope')
return addr_scope
return update_res
class AddressScopeTestPlugin(db_base_plugin_v2.NeutronDbPluginV2,
address_scope_db.AddressScopeDbMixin):
__native_pagination_support = True
__native_sorting_support = True
supported_extension_aliases = ["address-scope"]
class TestAddressScope(AddressScopeTestCase):
def setUp(self):
plugin = DB_PLUGIN_KLASS
ext_mgr = AddressScopeTestExtensionManager()
super(TestAddressScope, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
def test_create_address_scope_ipv4(self):
expected_addr_scope = {'name': 'foo-address-scope',
'tenant_id': self._tenant_id,
'shared': False,
'ip_version': constants.IP_VERSION_4}
self._test_create_address_scope(name='foo-address-scope',
expected=expected_addr_scope)
def test_create_address_scope_ipv6(self):
expected_addr_scope = {'name': 'foo-address-scope',
'tenant_id': self._tenant_id,
'shared': False,
'ip_version': constants.IP_VERSION_6}
self._test_create_address_scope(constants.IP_VERSION_6,
name='foo-address-scope',
expected=expected_addr_scope)
def test_create_address_scope_empty_name(self):
expected_addr_scope = {'name': '',
'tenant_id': self._tenant_id,
'shared': False}
self._test_create_address_scope(name='', expected=expected_addr_scope)
# no name specified
self._test_create_address_scope(expected=expected_addr_scope)
def test_create_address_scope_shared_admin(self):
expected_addr_scope = {'name': 'foo-address-scope', 'shared': True}
self._test_create_address_scope(name='foo-address-scope', admin=True,
shared=True,
expected=expected_addr_scope)
def test_created_address_scope_shared_non_admin(self):
res = self._create_address_scope(self.fmt, name='foo-address-scope',
tenant_id=self._tenant_id,
admin=False, shared=True)
self.assertEqual(webob.exc.HTTPForbidden.code, res.status_int)
def test_created_address_scope_specify_id(self):
res = self._create_address_scope(self.fmt, name='foo-address-scope',
id='foo-id')
self.assertEqual(webob.exc.HTTPClientError.code, res.status_int)
def test_delete_address_scope(self):
with self.address_scope(name='foo-address-scope') as addr_scope:
self._delete('address-scopes', addr_scope['address_scope']['id'])
self._show('address-scopes', addr_scope['address_scope']['id'],
expected_code=webob.exc.HTTPNotFound.code)
def test_update_address_scope(self):
addr_scope = self._test_create_address_scope(name='foo-address-scope')
data = {'address_scope': {'name': 'bar-address-scope'}}
self._test_update_address_scope(addr_scope['address_scope']['id'],
data, expected=data['address_scope'])
def test_update_address_scope_shared_true_admin(self):
addr_scope = self._test_create_address_scope(name='foo-address-scope')
data = {'address_scope': {'shared': True}}
self._test_update_address_scope(addr_scope['address_scope']['id'],
data, admin=True,
expected=data['address_scope'])
def test_update_address_scope_shared_true_non_admin(self):
addr_scope = self._test_create_address_scope(name='foo-address-scope')
data = {'address_scope': {'shared': True}}
res = self._test_update_address_scope(
addr_scope['address_scope']['id'], data, admin=False)
self.assertEqual(webob.exc.HTTPForbidden.code, res.status_int)
def test_update_address_scope_shared_false_admin(self):
addr_scope = self._test_create_address_scope(name='foo-address-scope',
admin=True, shared=True)
data = {'address_scope': {'shared': False}}
res = self._test_update_address_scope(
addr_scope['address_scope']['id'], data, admin=True)
self.assertEqual(webob.exc.HTTPClientError.code, res.status_int)
def test_get_address_scope(self):
addr_scope = self._test_create_address_scope(name='foo-address-scope')
req = self.new_show_request('address-scopes',
addr_scope['address_scope']['id'])
res = self.deserialize(self.fmt, req.get_response(self.ext_api))
self.assertEqual(addr_scope['address_scope']['id'],
res['address_scope']['id'])
def test_get_address_scope_different_tenants_not_shared(self):
addr_scope = self._test_create_address_scope(name='foo-address-scope')
req = self.new_show_request('address-scopes',
addr_scope['address_scope']['id'])
neutron_context = context.Context('', 'not-the-owner')
req.environ['neutron.context'] = neutron_context
res = req.get_response(self.ext_api)
self.assertEqual(webob.exc.HTTPNotFound.code, res.status_int)
def test_get_address_scope_different_tenants_shared(self):
addr_scope = self._test_create_address_scope(name='foo-address-scope',
shared=True, admin=True)
req = self.new_show_request('address-scopes',
addr_scope['address_scope']['id'])
neutron_context = context.Context('', 'test-tenant-2')
req.environ['neutron.context'] = neutron_context
res = self.deserialize(self.fmt, req.get_response(self.ext_api))
self.assertEqual(addr_scope['address_scope']['id'],
res['address_scope']['id'])
def test_list_address_scopes(self):
self._test_create_address_scope(name='foo-address-scope')
self._test_create_address_scope(constants.IP_VERSION_6,
name='bar-address-scope')
res = self._list('address-scopes')
self.assertEqual(2, len(res['address_scopes']))
def test_list_address_scopes_different_tenants_shared(self):
self._test_create_address_scope(name='foo-address-scope', shared=True,
admin=True)
admin_res = self._list('address-scopes')
mortal_res = self._list(
'address-scopes',
neutron_context=context.Context('', 'not-the-owner'))
self.assertEqual(1, len(admin_res['address_scopes']))
self.assertEqual(1, len(mortal_res['address_scopes']))
def test_list_address_scopes_different_tenants_not_shared(self):
self._test_create_address_scope(constants.IP_VERSION_6,
name='foo-address-scope')
admin_res = self._list('address-scopes')
mortal_res = self._list(
'address-scopes',
neutron_context=context.Context('', 'not-the-owner'))
self.assertEqual(1, len(admin_res['address_scopes']))
self.assertEqual(0, len(mortal_res['address_scopes']))
class TestSubnetPoolsWithAddressScopes(AddressScopeTestCase):
def setUp(self):
plugin = DB_PLUGIN_KLASS
ext_mgr = AddressScopeTestExtensionManager()
super(TestSubnetPoolsWithAddressScopes, self).setUp(plugin=plugin,
ext_mgr=ext_mgr)
def _test_create_subnetpool(self, prefixes, expected=None,
admin=False, **kwargs):
keys = kwargs.copy()
keys.setdefault('tenant_id', self._tenant_id)
with self.subnetpool(prefixes, admin, **keys) as subnetpool:
self._validate_resource(subnetpool, keys, 'subnetpool')
if expected:
self._compare_resource(subnetpool, expected, 'subnetpool')
return subnetpool
def test_create_subnetpool_associate_address_scope(self):
with self.address_scope(name='foo-address-scope') as addr_scope:
address_scope_id = addr_scope['address_scope']['id']
subnet = netaddr.IPNetwork('10.10.10.0/24')
expected = {'address_scope_id': address_scope_id}
self._test_create_subnetpool([subnet.cidr], expected=expected,
name='foo-subnetpool',
min_prefixlen='21',
address_scope_id=address_scope_id)
def test_create_subnetpool_associate_invalid_address_scope(self):
self.assertRaises(
webob.exc.HTTPClientError, self._test_create_subnetpool, [],
min_prefixlen='21', address_scope_id='foo-addr-scope-id')
def test_create_subnetpool_assoc_address_scope_with_prefix_intersect(self):
with self.address_scope(name='foo-address-scope') as addr_scope:
address_scope_id = addr_scope['address_scope']['id']
subnet = netaddr.IPNetwork('10.10.10.0/24')
expected = {'address_scope_id': address_scope_id}
self._test_create_subnetpool([subnet.cidr], expected=expected,
name='foo-subnetpool',
min_prefixlen='21',
address_scope_id=address_scope_id)
overlap_subnet = netaddr.IPNetwork('10.10.10.10/24')
self.assertRaises(
webob.exc.HTTPClientError, self._test_create_subnetpool,
[overlap_subnet.cidr], min_prefixlen='21',
address_scope_id=address_scope_id)
def test_update_subnetpool_associate_address_scope(self):
subnet = netaddr.IPNetwork('10.10.10.0/24')
initial_subnetpool = self._test_create_subnetpool([subnet.cidr],
name='foo-sp',
min_prefixlen='21')
with self.address_scope(name='foo-address-scope') as addr_scope:
address_scope_id = addr_scope['address_scope']['id']
data = {'subnetpool': {'address_scope_id': address_scope_id}}
req = self.new_update_request(
'subnetpools', data, initial_subnetpool['subnetpool']['id'])
api = self._api_for_resource('subnetpools')
res = self.deserialize(self.fmt, req.get_response(api))
self._compare_resource(res, data['subnetpool'], 'subnetpool')
def test_update_subnetpool_associate_invalid_address_scope(self):
subnet = netaddr.IPNetwork('10.10.10.0/24')
initial_subnetpool = self._test_create_subnetpool([subnet.cidr],
name='foo-sp',
min_prefixlen='21')
data = {'subnetpool': {'address_scope_id': 'foo-addr-scope-id'}}
req = self.new_update_request(
'subnetpools', data, initial_subnetpool['subnetpool']['id'])
api = self._api_for_resource('subnetpools')
res = req.get_response(api)
self.assertEqual(webob.exc.HTTPClientError.code, res.status_int)
def test_update_subnetpool_disassociate_address_scope(self):
with self.address_scope(name='foo-address-scope') as addr_scope:
address_scope_id = addr_scope['address_scope']['id']
subnet = netaddr.IPNetwork('10.10.10.0/24')
expected = {'address_scope_id': address_scope_id}
initial_subnetpool = self._test_create_subnetpool(
[subnet.cidr], expected=expected, name='foo-sp',
min_prefixlen='21', address_scope_id=address_scope_id)
data = {'subnetpool': {'address_scope_id': None}}
req = self.new_update_request(
'subnetpools', data, initial_subnetpool['subnetpool']['id'])
api = self._api_for_resource('subnetpools')
res = self.deserialize(self.fmt, req.get_response(api))
self._compare_resource(res, data['subnetpool'], 'subnetpool')
def test_update_subnetpool_associate_another_address_scope(self):
with self.address_scope(name='foo-address-scope') as addr_scope:
address_scope_id = addr_scope['address_scope']['id']
subnet = netaddr.IPNetwork('10.10.10.0/24')
expected = {'address_scope_id': address_scope_id}
initial_subnetpool = self._test_create_subnetpool(
[subnet.cidr], expected=expected, name='foo-sp',
min_prefixlen='21', address_scope_id=address_scope_id)
with self.address_scope(name='foo-address-scope') as other_a_s:
other_a_s_id = other_a_s['address_scope']['id']
update_data = {'subnetpool': {'address_scope_id':
other_a_s_id}}
req = self.new_update_request(
'subnetpools', update_data,
initial_subnetpool['subnetpool']['id'])
api = self._api_for_resource('subnetpools')
res = self.deserialize(self.fmt, req.get_response(api))
self._compare_resource(res, update_data['subnetpool'],
'subnetpool')
def _test_update_subnetpool_address_scope_notify(self, as_change=True):
with self.address_scope(name='foo-address-scope') as addr_scope:
foo_as_id = addr_scope['address_scope']['id']
subnet = netaddr.IPNetwork('10.10.10.0/24')
initial_subnetpool = self._test_create_subnetpool(
[subnet.cidr], name='foo-sp',
min_prefixlen='21', address_scope_id=foo_as_id)
subnetpool_id = initial_subnetpool['subnetpool']['id']
with self.address_scope(name='bar-address-scope') as other_as, \
self.network() as network:
data = {'subnet': {
'network_id': network['network']['id'],
'subnetpool_id': subnetpool_id,
'prefixlen': 24,
'ip_version': 4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
subnet = self.deserialize(self.fmt,
req.get_response(self.api))
with mock.patch.object(registry, 'notify') as notify:
plugin = db_base_plugin_v2.NeutronDbPluginV2()
plugin.is_address_scope_owned_by_tenant = mock.Mock(
return_value=True)
plugin._validate_address_scope_id = mock.Mock()
ctx = context.get_admin_context()
bar_as_id = other_as['address_scope']['id']
data = {'subnetpool': {
'name': 'bar-sp'}}
if as_change:
data['subnetpool']['address_scope_id'] = bar_as_id
updated_sp = plugin.update_subnetpool(
ctx, subnetpool_id, data)
self.assertEqual('bar-sp', updated_sp['name'])
if as_change:
self.assertEqual(bar_as_id,
updated_sp['address_scope_id'])
notify.assert_called_once_with(
resources.SUBNETPOOL_ADDRESS_SCOPE,
events.AFTER_UPDATE,
plugin.update_subnetpool, context=ctx,
subnetpool_id=subnetpool_id)
else:
self.assertEqual(foo_as_id,
updated_sp['address_scope_id'])
self.assertFalse(notify.called)
def test_update_subnetpool_address_scope_notify(self):
self._test_update_subnetpool_address_scope_notify()
def test_not_update_subnetpool_address_scope_not_notify(self):
self._test_update_subnetpool_address_scope_notify(False)
def test_network_create_contain_address_scope_attr(self):
with self.network() as network:
result = self._show('networks', network['network']['id'])
keys = [ext_address_scope.IPV4_ADDRESS_SCOPE,
ext_address_scope.IPV6_ADDRESS_SCOPE]
for k in keys:
# Correlated address scopes should initially be None
self.assertIsNone(result['network'][k])
def test_correlate_network_with_address_scope(self):
with self.address_scope(name='v4-as') as v4_addr_scope, \
self.address_scope(
name='v6-as',
ip_version=constants.IP_VERSION_6) as v6_addr_scope, \
self.network() as network:
v4_as_id = v4_addr_scope['address_scope']['id']
subnet = netaddr.IPNetwork('10.10.10.0/24')
v4_subnetpool = self._test_create_subnetpool(
[subnet.cidr], name='v4-sp',
min_prefixlen='24', address_scope_id=v4_as_id)
v4_subnetpool_id = v4_subnetpool['subnetpool']['id']
v6_as_id = v6_addr_scope['address_scope']['id']
subnet = netaddr.IPNetwork('fd5c:6ee1:c7ae::/64')
v6_subnetpool = self._test_create_subnetpool(
[subnet.cidr], name='v6-sp',
min_prefixlen='64', address_scope_id=v6_as_id)
v6_subnetpool_id = v6_subnetpool['subnetpool']['id']
data = {'subnet': {
'network_id': network['network']['id'],
'subnetpool_id': v4_subnetpool_id,
'ip_version': 4,
'tenant_id': network['network']['tenant_id']}}
req = self.new_create_request('subnets', data)
self.deserialize(self.fmt, req.get_response(self.api))
data['subnet']['subnetpool_id'] = v6_subnetpool_id
data['subnet']['ip_version'] = 6
req = self.new_create_request('subnets', data)
self.deserialize(self.fmt, req.get_response(self.api))
result = self._show('networks', network['network']['id'])
self.assertEqual(
v4_as_id,
result['network'][ext_address_scope.IPV4_ADDRESS_SCOPE])
self.assertEqual(
v6_as_id,
result['network'][ext_address_scope.IPV6_ADDRESS_SCOPE])
def test_delete_address_scope_in_use(self):
with self.address_scope(name='foo-address-scope') as addr_scope:
address_scope_id = addr_scope['address_scope']['id']
subnet = netaddr.IPNetwork('10.10.10.0/24')
expected = {'address_scope_id': address_scope_id}
self._test_create_subnetpool([subnet.cidr], expected=expected,
name='foo-subnetpool',
min_prefixlen='21',
address_scope_id=address_scope_id)
self._delete('address-scopes', address_scope_id,
expected_code=webob.exc.HTTPConflict.code)
def test_add_subnetpool_address_scope_wrong_address_family(self):
with self.address_scope(constants.IP_VERSION_6,
name='foo-address-scope') as addr_scope:
address_scope_id = addr_scope['address_scope']['id']
subnet = netaddr.IPNetwork('10.10.10.0/24')
self.assertRaises(webob.exc.HTTPClientError,
self._test_create_subnetpool,
[subnet.cidr], name='foo-subnetpool',
min_prefixlen='21',
address_scope_id=address_scope_id)
def test_update_subnetpool_associate_address_scope_wrong_family(self):
with self.address_scope(constants.IP_VERSION_6,
name='foo-address-scope') as addr_scope:
address_scope_id = addr_scope['address_scope']['id']
subnet = netaddr.IPNetwork('2001:db8::/64')
expected = {'address_scope_id': address_scope_id}
initial_subnetpool = self._test_create_subnetpool(
[subnet.cidr], expected=expected, name='foo-sp',
min_prefixlen='64', address_scope_id=address_scope_id)
with self.address_scope(name='foo-address-scope') as other_a_s:
other_a_s_id = other_a_s['address_scope']['id']
update_data = {'subnetpool': {'address_scope_id':
other_a_s_id}}
req = self.new_update_request(
'subnetpools', update_data,
initial_subnetpool['subnetpool']['id'])
api = self._api_for_resource('subnetpools')
res = req.get_response(api)
self.assertEqual(webob.exc.HTTPBadRequest.code,
res.status_int)