7594bb0627
Now that we are python3 only, we should move to using the built in version of mock that supports all of our testing needs and remove the dependency on the "mock" package. This patch moves all references to "import mock" to "from unittest import mock". It also cleans up some new line inconsistency. Fixed an inconsistency in the OVSBridge.deferred() definition as it needs to also have an *args argument. Fixed an issue where an l3-agent test was mocking functools.partial, causing a python3.8 failure. Unit tests only, removing from tests/base.py affects functional tests which need additional work. Change-Id: I40e8a8410840c3774c72ae1a8054574445d66ece
691 lines
34 KiB
Python
691 lines
34 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
|
|
from unittest import mock
|
|
|
|
import netaddr
|
|
from neutron_lib.api.definitions import address_scope as apidef
|
|
from neutron_lib.callbacks import events
|
|
from neutron_lib.callbacks import registry
|
|
from neutron_lib.callbacks import resources
|
|
from neutron_lib import constants
|
|
from neutron_lib import context
|
|
from neutron_lib.plugins import directory
|
|
import webob.exc
|
|
|
|
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):
|
|
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)
|
|
update_req.environ['neutron.context'] = context.Context(
|
|
'', tenant_id or self._tenant_id, is_admin=admin)
|
|
|
|
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 = [apidef.ALIAS]
|
|
|
|
|
|
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': constants.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, 'publish') as publish:
|
|
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'])
|
|
publish.assert_called_once_with(
|
|
resources.SUBNETPOOL_ADDRESS_SCOPE,
|
|
events.AFTER_UPDATE,
|
|
plugin.update_subnetpool, payload=mock.ANY)
|
|
payload = publish.mock_calls[0][2]['payload']
|
|
self.assertEqual(ctx, payload.context)
|
|
self.assertEqual(subnetpool_id, payload.resource_id)
|
|
else:
|
|
self.assertEqual(foo_as_id,
|
|
updated_sp['address_scope_id'])
|
|
self.assertFalse(publish.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 = [apidef.IPV4_ADDRESS_SCOPE,
|
|
apidef.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': constants.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'] = constants.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'][apidef.IPV4_ADDRESS_SCOPE])
|
|
self.assertEqual(
|
|
v6_as_id,
|
|
result['network'][apidef.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)
|
|
|
|
def test_create_two_subnets_different_subnetpools_same_network(self):
|
|
with self.address_scope(constants.IP_VERSION_4,
|
|
name='foo-address-scope') as addr_scope:
|
|
addr_scope = addr_scope['address_scope']
|
|
with self.subnetpool(
|
|
['10.10.0.0/16'],
|
|
name='subnetpool_a',
|
|
tenant_id=addr_scope['tenant_id'],
|
|
default_prefixlen=24,
|
|
address_scope_id=addr_scope['id']) as subnetpool_a,\
|
|
self.subnetpool(
|
|
['10.20.0.0/16'],
|
|
name='subnetpool_b',
|
|
tenant_id=addr_scope['tenant_id'],
|
|
default_prefixlen=24,
|
|
address_scope_id=addr_scope['id']) as subnetpool_b:
|
|
subnetpool_a = subnetpool_a['subnetpool']
|
|
subnetpool_b = subnetpool_b['subnetpool']
|
|
|
|
with self.network(
|
|
tenant_id=addr_scope['tenant_id']) as network:
|
|
subnet_a = self._make_subnet(
|
|
self.fmt,
|
|
network,
|
|
constants.ATTR_NOT_SPECIFIED,
|
|
None,
|
|
subnetpool_id=subnetpool_a['id'],
|
|
ip_version=constants.IP_VERSION_4,
|
|
tenant_id=addr_scope['tenant_id'])
|
|
subnet_b = self._make_subnet(
|
|
self.fmt,
|
|
network,
|
|
constants.ATTR_NOT_SPECIFIED,
|
|
None,
|
|
subnetpool_id=subnetpool_b['id'],
|
|
ip_version=constants.IP_VERSION_4,
|
|
tenant_id=addr_scope['tenant_id'])
|
|
|
|
# Look up subnet counts and perform assertions
|
|
ctx = context.Context('', addr_scope['tenant_id'])
|
|
pl = directory.get_plugin()
|
|
total_count = pl.get_subnets_count(
|
|
ctx,
|
|
filters={'network_id':
|
|
[network['network']['id']]})
|
|
subnets_pool_a_count = pl.get_subnets_count(
|
|
ctx,
|
|
filters={'id': [subnet_a['subnet']['id']],
|
|
'subnetpool_id': [subnetpool_a['id']],
|
|
'network_id': [network['network']['id']]})
|
|
subnets_pool_b_count = pl.get_subnets_count(
|
|
ctx,
|
|
filters={'id': [subnet_b['subnet']['id']],
|
|
'subnetpool_id': [subnetpool_b['id']],
|
|
'network_id': [network['network']['id']]})
|
|
self.assertEqual(2, total_count)
|
|
self.assertEqual(1, subnets_pool_a_count)
|
|
self.assertEqual(1, subnets_pool_b_count)
|
|
|
|
def test_block_update_subnetpool_network_affinity(self):
|
|
with self.address_scope(constants.IP_VERSION_4,
|
|
name='scope-a') as scope_a,\
|
|
self.address_scope(constants.IP_VERSION_4,
|
|
name='scope-b') as scope_b:
|
|
scope_a = scope_a['address_scope']
|
|
scope_b = scope_b['address_scope']
|
|
|
|
with self.subnetpool(
|
|
['10.10.0.0/16'],
|
|
name='subnetpool_a',
|
|
tenant_id=scope_a['tenant_id'],
|
|
default_prefixlen=24,
|
|
address_scope_id=scope_a['id']) as subnetpool_a,\
|
|
self.subnetpool(
|
|
['10.20.0.0/16'],
|
|
name='subnetpool_b',
|
|
tenant_id=scope_a['tenant_id'],
|
|
default_prefixlen=24,
|
|
address_scope_id=scope_a['id']) as subnetpool_b:
|
|
subnetpool_a = subnetpool_a['subnetpool']
|
|
subnetpool_b = subnetpool_b['subnetpool']
|
|
|
|
with self.network(
|
|
tenant_id=scope_a['tenant_id']) as network:
|
|
self._make_subnet(
|
|
self.fmt,
|
|
network,
|
|
constants.ATTR_NOT_SPECIFIED,
|
|
None,
|
|
subnetpool_id=subnetpool_a['id'],
|
|
ip_version=constants.IP_VERSION_4,
|
|
tenant_id=scope_a['tenant_id'])
|
|
self._make_subnet(
|
|
self.fmt,
|
|
network,
|
|
constants.ATTR_NOT_SPECIFIED,
|
|
None,
|
|
subnetpool_id=subnetpool_b['id'],
|
|
ip_version=constants.IP_VERSION_4,
|
|
tenant_id=scope_a['tenant_id'])
|
|
|
|
# Attempt to update subnetpool_b's address scope and
|
|
# assert failure.
|
|
data = {'subnetpool': {'address_scope_id':
|
|
scope_b['id']}}
|
|
req = self.new_update_request('subnetpools', data,
|
|
subnetpool_b['id'])
|
|
api = self._api_for_resource('subnetpools')
|
|
res = req.get_response(api)
|
|
self.assertEqual(webob.exc.HTTPBadRequest.code,
|
|
res.status_int)
|
|
|
|
def test_ipv6_pd_add_non_pd_subnet_to_same_network(self):
|
|
with self.address_scope(constants.IP_VERSION_6,
|
|
name='foo-address-scope') as addr_scope:
|
|
addr_scope = addr_scope['address_scope']
|
|
with self.subnetpool(
|
|
['2001:db8:1234::/48'],
|
|
name='non_pd_pool',
|
|
tenant_id=addr_scope['tenant_id'],
|
|
default_prefixlen=64,
|
|
address_scope_id=addr_scope['id']) as non_pd_pool:
|
|
non_pd_pool = non_pd_pool['subnetpool']
|
|
|
|
with self.network(
|
|
tenant_id=addr_scope['tenant_id']) as network:
|
|
with self.subnet(cidr=None,
|
|
network=network,
|
|
ip_version=constants.IP_VERSION_6,
|
|
subnetpool_id=constants.IPV6_PD_POOL_ID,
|
|
ipv6_ra_mode=constants.IPV6_SLAAC,
|
|
ipv6_address_mode=constants.IPV6_SLAAC):
|
|
res = self._create_subnet(
|
|
self.fmt,
|
|
cidr=None,
|
|
net_id=network['network']['id'],
|
|
subnetpool_id=non_pd_pool['id'],
|
|
tenant_id=addr_scope['tenant_id'],
|
|
ip_version=constants.IP_VERSION_6)
|
|
self.assertEqual(webob.exc.HTTPBadRequest.code,
|
|
res.status_int)
|
|
|
|
def test_ipv6_non_pd_add_pd_subnet_to_same_network(self):
|
|
with self.address_scope(constants.IP_VERSION_6,
|
|
name='foo-address-scope') as addr_scope:
|
|
addr_scope = addr_scope['address_scope']
|
|
with self.subnetpool(
|
|
['2001:db8:1234::/48'],
|
|
name='non_pd_pool',
|
|
tenant_id=addr_scope['tenant_id'],
|
|
default_prefixlen=64,
|
|
address_scope_id=addr_scope['id']) as non_pd_pool:
|
|
non_pd_pool = non_pd_pool['subnetpool']
|
|
|
|
with self.network(
|
|
tenant_id=addr_scope['tenant_id']) as network:
|
|
with self.subnet(cidr=None,
|
|
network=network,
|
|
ip_version=constants.IP_VERSION_6,
|
|
subnetpool_id=non_pd_pool['id']):
|
|
res = self._create_subnet(
|
|
self.fmt,
|
|
cidr=None,
|
|
net_id=network['network']['id'],
|
|
tenant_id=addr_scope['tenant_id'],
|
|
subnetpool_id=constants.IPV6_PD_POOL_ID,
|
|
ip_version=constants.IP_VERSION_6,
|
|
ipv6_ra_mode=constants.IPV6_SLAAC,
|
|
ipv6_address_mode=constants.IPV6_SLAAC)
|
|
self.assertEqual(webob.exc.HTTPBadRequest.code,
|
|
res.status_int)
|