openstack/neutron
Code Issues Proposed changes

Merge "Support subnetpool association to an address scope"

Browse Source
This commit is contained in:
Jenkins 2015-08-07 03:22:49 +00:00 committed by Gerrit Code Review
commit b19772c592
13 changed files with 533 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
neutron/common/exceptions.py
View File

@ -449,6 +449,21 @@ class SubnetAllocationError(NeutronException):
message = _("Failed to allocate subnet: %(reason)s")
class AddressScopePrefixConflict(Conflict):
message = _("Failed to associate address scope: subnetpools "
"within an address scope must have unique prefixes")
class IllegalSubnetPoolAssociationToAddressScope(BadRequest):
message = _("Illegal subnetpool association: subnetpool %(subnetpool_id)s "
" cannot be associated with address scope"
" %(address_scope_id)s")
class IllegalSubnetPoolUpdate(BadRequest):
message = _("Illegal subnetpool update : %(reason)s")
class MinPrefixSubnetAllocationError(BadRequest):
message = _("Unable to allocate subnet with prefix length %(prefixlen)s, "
"minimum allowed prefix is %(min_prefixlen)s")

17
neutron/db/address_scope_db.py
View File

@ -51,6 +51,21 @@ class AddressScopeDbMixin(ext_address_scope.AddressScopePluginBase):
except exc.NoResultFound:
raise ext_address_scope.AddressScopeNotFound(address_scope_id=id)
def is_address_scope_owned_by_tenant(self, context, id):
"""Check if address scope id is owned by the tenant or not.
AddressScopeNotFound is raised if the
- address scope id doesn't exist or
- if the (unshared) address scope id is not owned by this tenant.
@return Returns true if the user is admin or tenant is owner
Returns false if the address scope id is shared and not
owned by the tenant.
"""
address_scope = self._get_address_scope(context, id)
return context.is_admin or (
address_scope.tenant_id == context.tenant_id)
def create_address_scope(self, context, address_scope):
"""Create a address scope."""
a_s = address_scope['address_scope']
@ -101,5 +116,7 @@ class AddressScopeDbMixin(ext_address_scope.AddressScopePluginBase):
def delete_address_scope(self, context, id):
with context.session.begin(subtransactions=True):
if self._get_subnetpools_by_address_scope_id(context, id):
raise ext_address_scope.AddressScopeInUse(address_scope_id=id)
address_scope = self._get_address_scope(context, id)
context.session.delete(address_scope)

9
neutron/db/db_base_plugin_common.py
View File

@ -116,7 +116,8 @@ class DbBasePluginCommon(common_db_mixin.CommonDbMixin):
'prefixes': [prefix['cidr']
for prefix in subnetpool['prefixes']],
'ip_version': subnetpool['ip_version'],
'default_quota': subnetpool['default_quota']}
'default_quota': subnetpool['default_quota'],
'address_scope_id': subnetpool['address_scope_id']}
return self._fields(res, fields)
def _make_port_dict(self, port, fields=None,
@ -163,6 +164,12 @@ class DbBasePluginCommon(common_db_mixin.CommonDbMixin):
# NOTE(tidwellr): see note in _get_all_subnets()
return context.session.query(models_v2.SubnetPool).all()
def _get_subnetpools_by_address_scope_id(self, context, address_scope_id):
# NOTE(vikram.choudhary): see note in _get_all_subnets()
subnetpool_qry = context.session.query(models_v2.SubnetPool)
return subnetpool_qry.filter_by(
address_scope_id=address_scope_id).all()
def _get_port(self, context, id):
try:
port = self._get_by_id(context, models_v2.Port, id)

63
neutron/db/db_base_plugin_v2.py
View File

@ -773,11 +773,63 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
subnetpool_prefix = models_v2.SubnetPoolPrefix(**prefix_args)
context.session.add(subnetpool_prefix)
def _validate_address_scope_id(self, context, address_scope_id,
subnetpool_id, sp_prefixes):
"""Validate the address scope before associating.
Subnetpool can associate with an address scope if
- the tenant user is the owner of both the subnetpool and
address scope
- the admin is associating the subnetpool with the shared
address scope
- there is no prefix conflict with the existing subnetpools
associated with the address scope.
"""
if not attributes.is_attr_set(address_scope_id):
return
if not self.is_address_scope_owned_by_tenant(context,
address_scope_id):
raise n_exc.IllegalSubnetPoolAssociationToAddressScope(
subnetpool_id=subnetpool_id, address_scope_id=address_scope_id)
subnetpools = self._get_subnetpools_by_address_scope_id(
context, address_scope_id)
new_set = netaddr.IPSet(sp_prefixes)
for sp in subnetpools:
if sp.id == subnetpool_id:
continue
sp_set = netaddr.IPSet([prefix['cidr'] for prefix in sp.prefixes])
if sp_set.intersection(new_set):
raise n_exc.AddressScopePrefixConflict()
def _check_subnetpool_update_allowed(self, context, subnetpool_id,
address_scope_id):
"""Check if the subnetpool can be updated or not.
If the subnetpool is associated to a shared address scope not owned
by the tenant, then the subnetpool cannot be updated.
"""
if not self.is_address_scope_owned_by_tenant(context,
address_scope_id):
msg = _("subnetpool %(subnetpool_id)s cannot be updated when"
" associated with shared address scope "
"%(address_scope_id)s") % {
'subnetpool_id': subnetpool_id,
'address_scope_id': address_scope_id}
raise n_exc.IllegalSubnetPoolUpdate(reason=msg)
def create_subnetpool(self, context, subnetpool):
"""Create a subnetpool"""
sp = subnetpool['subnetpool']
sp_reader = subnet_alloc.SubnetPoolReader(sp)
if sp_reader.address_scope_id is attributes.ATTR_NOT_SPECIFIED:
sp_reader.address_scope_id = None
self._validate_address_scope_id(context, sp_reader.address_scope_id,
id, sp_reader.prefixes)
tenant_id = self._get_tenant_id_for_create(context, sp)
with context.session.begin(subtransactions=True):
pool_args = {'tenant_id': tenant_id,
@ -789,7 +841,8 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
'min_prefixlen': sp_reader.min_prefixlen,
'max_prefixlen': sp_reader.max_prefixlen,
'shared': sp_reader.shared,
'default_quota': sp_reader.default_quota}
'default_quota': sp_reader.default_quota,
'address_scope_id': sp_reader.address_scope_id}
subnetpool = models_v2.SubnetPool(**pool_args)
context.session.add(subnetpool)
for prefix in sp_reader.prefixes:
@ -826,7 +879,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
for key in ['id', 'name', 'ip_version', 'min_prefixlen',
'max_prefixlen', 'default_prefixlen', 'shared',
'default_quota']:
'default_quota', 'address_scope_id']:
self._write_key(key, updated, model, new_pool)
return updated
@ -847,6 +900,12 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
updated = self._updated_subnetpool_dict(orig_sp, new_sp)
updated['tenant_id'] = orig_sp.tenant_id
reader = subnet_alloc.SubnetPoolReader(updated)
if orig_sp.address_scope_id:
self._check_subnetpool_update_allowed(context, id,
orig_sp.address_scope_id)
self._validate_address_scope_id(context, reader.address_scope_id,
id, reader.prefixes)
orig_sp.update(self._filter_non_model_columns(
reader.subnetpool,
models_v2.SubnetPool))

2
neutron/db/migration/alembic_migrations/versions/HEADS
View File

@ -1,3 +1,3 @@
1c844d1677f7
1b4c6e320f79
2a16083502f3
kilo

36
neutron/db/migration/alembic_migrations/versions/liberty/expand/1b4c6e320f79_address_scope_support_in_subnetpool.py Normal file
View File

@ -0,0 +1,36 @@
# Copyright 2015 Huawei Technologies India Pvt. Ltd.
#
# 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.
#
"""address scope support in subnetpool
Revision ID: 1b4c6e320f79
Revises: 1c844d1677f7
Create Date: 2015-07-03 09:48:39.491058
"""
# revision identifiers, used by Alembic.
revision = '1b4c6e320f79'
down_revision = '1c844d1677f7'
from alembic import op
import sqlalchemy as sa
def upgrade():
op.add_column('subnetpools',
sa.Column('address_scope_id',
sa.String(length=36),
nullable=True))

1
neutron/db/models_v2.py
View File

@ -245,6 +245,7 @@ class SubnetPool(model_base.BASEV2, HasId, HasTenant):
shared = sa.Column(sa.Boolean, nullable=False)
default_quota = sa.Column(sa.Integer, nullable=True)
hash = sa.Column(sa.String(36), nullable=False, server_default='')
address_scope_id = sa.Column(sa.String(36), nullable=True)
prefixes = orm.relationship(SubnetPoolPrefix,
backref='subnetpools',
cascade='all, delete, delete-orphan',

16
neutron/extensions/address_scope.py
View File

@ -23,7 +23,7 @@ import six
ADDRESS_SCOPE = 'address_scope'
ADDRESS_SCOPES = '%ss' % ADDRESS_SCOPE
ADDRESS_SCOPE_ID = 'address_scope_id'
# Attribute Map
RESOURCE_ATTRIBUTE_MAP = {
@ -50,6 +50,13 @@ RESOURCE_ATTRIBUTE_MAP = {
'is_visible': True,
'required_by_policy': True,
'enforce_policy': True},
},
attr.SUBNETPOOLS: {
ADDRESS_SCOPE_ID: {'allow_post': True,
'allow_put': True,
'default': attr.ATTR_NOT_SPECIFIED,
'validate': {'type:uuid_or_none': None},
'is_visible': True}
}
}
@ -58,9 +65,10 @@ class AddressScopeNotFound(nexception.NotFound):
message = _("Address scope %(address_scope_id)s could not be found")
class AddressScopeDeleteError(nexception.BadRequest):
message = _("Unable to delete address scope %(address_scope_id)s : "
"%(reason)s")
class AddressScopeInUse(nexception.InUse):
message = _("Unable to complete operation on "
"address scope %(address_scope_id)s. There are one or more"
" subnet pools in use on the address scope")
class AddressScopeUpdateError(nexception.BadRequest):

6
neutron/ipam/subnet_alloc.py
View File

@ -226,6 +226,7 @@ class SubnetPoolReader(object):
self._read_prefix_bounds(subnetpool)
self._read_attrs(subnetpool,
['tenant_id', 'name', 'shared'])
self._read_address_scope(subnetpool)
self.subnetpool = {'id': self.id,
'name': self.name,
'tenant_id': self.tenant_id,
@ -237,6 +238,7 @@ class SubnetPoolReader(object):
'default_prefix': self.default_prefix,
'default_prefixlen': self.default_prefixlen,
'default_quota': self.default_quota,
'address_scope_id': self.address_scope_id,
'shared': self.shared}
def _read_attrs(self, subnetpool, keys):
@ -313,6 +315,10 @@ class SubnetPoolReader(object):
self.ip_version = ip_version
self.prefixes = self._compact_subnetpool_prefix_list(prefix_list)
def _read_address_scope(self, subnetpool):
self.address_scope_id = subnetpool.get('address_scope_id',
attributes.ATTR_NOT_SPECIFIED)
def _compact_subnetpool_prefix_list(self, prefix_list):
"""Compact any overlapping prefixes in prefix_list and return the
result

15
neutron/tests/api/test_address_scopes_negative.py
View File

@ -75,3 +75,18 @@ class AddressScopeTestNegative(test_address_scopes.AddressScopeTestBase):
self.assertRaises(lib_exc.BadRequest,
self.admin_client.update_address_scope,
address_scope['id'], name='new-name', shared=False)
@test.attr(type=['negative', 'smoke'])
@test.idempotent_id('1e471e5c-6f9c-437a-9257-fd9bc4b6f0fb')
def test_delete_address_scope_associated_with_subnetpool(self):
address_scope = self._create_address_scope()
prefixes = [u'10.11.12.0/24']
subnetpool_data = {'subnetpool': {
'name': 'foo-subnetpool',
'min_prefixlen': '29', 'prefixes': prefixes,
'address_scope_id': address_scope['id']}}
body = self.client.create_subnetpool(subnetpool_data)
subnetpool = body['subnetpool']
self.addCleanup(self.client.delete_subnetpool, subnetpool['id'])
self.assertRaises(lib_exc.Conflict, self.client.delete_address_scope,
address_scope['id'])

66
neutron/tests/api/test_subnetpools.py
View File

@ -239,6 +239,72 @@ class SubnetPoolsTest(base.BaseNetworkTest):
self.assertEqual(pool_id, subnet['subnetpool_id'])
self.assertTrue(cidr.endswith(str(self.max_prefixlen)))
@test.attr(type='smoke')
@test.idempotent_id('49b44c64-1619-4b29-b527-ffc3c3115dc4')
def test_create_subnetpool_associate_address_scope(self):
address_scope = self.create_address_scope(
name=data_utils.rand_name('smoke-address-scope'))
name, pool_id = self._create_subnetpool(
self.client, pool_values={'address_scope_id': address_scope['id']})
self.addCleanup(self.client.delete_subnetpool, pool_id)
body = self.client.get_subnetpool(pool_id)
self.assertEqual(address_scope['id'],
body['subnetpool']['address_scope_id'])
@test.attr(type='smoke')
@test.idempotent_id('910b6393-db24-4f6f-87dc-b36892ad6c8c')
def test_update_subnetpool_associate_address_scope(self):
address_scope = self.create_address_scope(
name=data_utils.rand_name('smoke-address-scope'))
name, pool_id = self._create_subnetpool(self.client)
self.addCleanup(self.client.delete_subnetpool, pool_id)
body = self.client.get_subnetpool(pool_id)
self.assertIsNone(body['subnetpool']['address_scope_id'])
subnetpool_data = {'subnetpool': {'address_scope_id':
address_scope['id']}}
self.client.update_subnetpool(pool_id, subnetpool_data)
body = self.client.get_subnetpool(pool_id)
self.assertEqual(address_scope['id'],
body['subnetpool']['address_scope_id'])
@test.attr(type='smoke')
@test.idempotent_id('18302e80-46a3-4563-82ac-ccd1dd57f652')
def test_update_subnetpool_associate_another_address_scope(self):
address_scope = self.create_address_scope(
name=data_utils.rand_name('smoke-address-scope'))
another_address_scope = self.create_address_scope(
name=data_utils.rand_name('smoke-address-scope'))
name, pool_id = self._create_subnetpool(
self.client, pool_values={'address_scope_id':
address_scope['id']})
self.addCleanup(self.client.delete_subnetpool, pool_id)
body = self.client.get_subnetpool(pool_id)
self.assertEqual(address_scope['id'],
body['subnetpool']['address_scope_id'])
subnetpool_data = {'subnetpool': {'address_scope_id':
another_address_scope['id']}}
self.client.update_subnetpool(pool_id, subnetpool_data)
body = self.client.get_subnetpool(pool_id)
self.assertEqual(another_address_scope['id'],
body['subnetpool']['address_scope_id'])
@test.attr(type='smoke')
@test.idempotent_id('f8970048-e41b-42d6-934b-a1297b07706a')
def test_update_subnetpool_disassociate_address_scope(self):
address_scope = self.create_address_scope(
name=data_utils.rand_name('smoke-address-scope'))
name, pool_id = self._create_subnetpool(
self.client, pool_values={'address_scope_id': address_scope['id']})
self.addCleanup(self.client.delete_subnetpool, pool_id)
body = self.client.get_subnetpool(pool_id)
self.assertEqual(address_scope['id'],
body['subnetpool']['address_scope_id'])
subnetpool_data = {'subnetpool': {'address_scope_id': None}}
self.client.update_subnetpool(pool_id, subnetpool_data)
body = self.client.get_subnetpool(pool_id)
self.assertIsNone(body['subnetpool']['address_scope_id'])
class SubnetPoolsTestV6(SubnetPoolsTest):

171
neutron/tests/api/test_subnetpools_negative.py
View File

@ -14,6 +14,7 @@
# under the License.
import copy
import uuid
from tempest_lib.common.utils import data_utils
from tempest_lib import exceptions as lib_exc
@ -139,3 +140,173 @@ class SubnetPoolsNegativeTestJSON(base.BaseNetworkTest):
network_id=network['id'],
ip_version=4,
subnetpool_id=pool_id)
@test.attr(type=['negative', 'smoke'])
@test.idempotent_id('9589e332-638e-476e-81bd-013d964aa3cb')
def test_create_subnetpool_associate_invalid_address_scope(self):
subnetpool_data = copy.deepcopy(self._subnetpool_data)
subnetpool_data['subnetpool']['address_scope_id'] = 'foo-addr-scope'
self.assertRaises(lib_exc.BadRequest, self.client.create_subnetpool,
subnetpool_data)
@test.attr(type=['negative', 'smoke'])
@test.idempotent_id('3b6c5942-485d-4964-a560-55608af020b5')
def test_create_subnetpool_associate_non_exist_address_scope(self):
subnetpool_data = copy.deepcopy(self._subnetpool_data)
non_exist_address_scope_id = str(uuid.uuid4())
subnetpool_data['subnetpool']['address_scope_id'] = (
non_exist_address_scope_id)
self.assertRaises(lib_exc.NotFound, self.client.create_subnetpool,
subnetpool_data)
@test.attr(type=['negative', 'smoke'])
@test.idempotent_id('2dfb4269-8657-485a-a053-b022e911456e')
def test_create_subnetpool_associate_address_scope_prefix_intersect(self):
address_scope = self.create_address_scope(
name=data_utils.rand_name('smoke-address-scope'))
addr_scope_id = address_scope['id']
pool_id = self._create_subnetpool(
self.client, pool_values={'address_scope_id': addr_scope_id})
self.addCleanup(self.client.delete_subnetpool, pool_id)
subnetpool_data = {'subnetpool': {'name': 'foo-subnetpool',
'prefixes': [u'10.11.12.13/24'],
'min_prefixlen': '29',
'address_scope_id': addr_scope_id}}
self.assertRaises(lib_exc.Conflict, self.client.create_subnetpool,
subnetpool_data)
@test.attr(type=['negative', 'smoke'])
@test.idempotent_id('83a19a13-5384-42e2-b579-43fc69c80914')
def test_create_sp_associate_address_scope_multiple_prefix_intersect(self):
address_scope = self.create_address_scope(
name=data_utils.rand_name('smoke-address-scope'))
addr_scope_id = address_scope['id']
pool_values = {'address_scope_id': addr_scope_id,
'prefixes': [u'20.0.0.0/18', u'30.0.0.0/18']}
pool_id = self._create_subnetpool(
self.client, pool_values=pool_values)
self.addCleanup(self.client.delete_subnetpool, pool_id)
prefixes = [u'40.0.0.0/18', u'50.0.0.0/18', u'30.0.0.0/12']
subnetpool_data = {'subnetpool': {'name': 'foo-subnetpool',
'prefixes': prefixes,
'min_prefixlen': '29',
'address_scope_id': addr_scope_id}}
self.assertRaises(lib_exc.Conflict, self.client.create_subnetpool,
subnetpool_data)
@test.attr(type=['negative', 'smoke'])
@test.idempotent_id('f06d8e7b-908b-4e94-b570-8156be6a4bf1')
def test_create_subnetpool_associate_address_scope_of_other_owner(self):
address_scope = self.create_address_scope(
name=data_utils.rand_name('smoke-address-scope'), is_admin=True)
address_scope_id = address_scope['id']
subnetpool_data = copy.deepcopy(self._subnetpool_data)
subnetpool_data['subnetpool']['address_scope_id'] = address_scope_id
self.assertRaises(lib_exc.NotFound, self.client.create_subnetpool,
subnetpool_data)
@test.attr(type=['negative', 'smoke'])
@test.idempotent_id('3396ec6c-cb80-4ebe-b897-84e904580bdf')
def test_tenant_create_subnetpool_associate_shared_address_scope(self):
address_scope = self.create_address_scope(
name=data_utils.rand_name('smoke-address-scope'), is_admin=True,
shared=True)
subnetpool_data = copy.deepcopy(self._subnetpool_data)
subnetpool_data['subnetpool']['address_scope_id'] = (
address_scope['id'])
self.assertRaises(lib_exc.BadRequest, self.client.create_subnetpool,
subnetpool_data)
@test.attr(type='smoke')
@test.idempotent_id('6d3d9ad5-32d4-4d63-aa00-8c62f73e2881')
def test_update_subnetpool_associate_address_scope_of_other_owner(self):
address_scope = self.create_address_scope(
name=data_utils.rand_name('smoke-address-scope'), is_admin=True)
address_scope_id = address_scope['id']
pool_id = self._create_subnetpool(self.client)
self.addCleanup(self.client.delete_subnetpool, pool_id)
subnetpool_data = {'subnetpool': {'address_scope_id':
address_scope_id}}
self.assertRaises(lib_exc.NotFound, self.client.update_subnetpool,
pool_id, subnetpool_data)
def _test_update_subnetpool_prefix_intersect_helper(
self, pool_1_prefixes, pool_2_prefixes, pool_1_updated_prefixes):
# create two subnet pools associating to an address scope.
# Updating the first subnet pool with the prefix intersecting
# with the second one should be a failure
address_scope = self.create_address_scope(
name=data_utils.rand_name('smoke-address-scope'))
addr_scope_id = address_scope['id']
pool_values = {'address_scope_id': addr_scope_id,
'prefixes': pool_1_prefixes}
pool_id_1 = self._create_subnetpool(self.client,
pool_values=pool_values)
self.addCleanup(self.client.delete_subnetpool, pool_id_1)
pool_values = {'address_scope_id': addr_scope_id,
'prefixes': pool_2_prefixes}
pool_id_2 = self._create_subnetpool(self.client,
pool_values=pool_values)
self.addCleanup(self.client.delete_subnetpool, pool_id_2)
# now update the pool_id_1 with the prefix interesecting with
# pool_id_2
subnetpool_data = {'subnetpool': {'prefixes':
pool_1_updated_prefixes}}
self.assertRaises(lib_exc.Conflict, self.client.update_subnetpool,
pool_id_1, subnetpool_data)
@test.attr(type=['negative', 'smoke'])
@test.idempotent_id('96006292-7214-40e0-a471-153fb76e6b31')
def test_update_subnetpool_prefix_intersect(self):
pool_1_prefix = [u'20.0.0.0/18']
pool_2_prefix = [u'20.10.0.0/24']
pool_1_updated_prefix = [u'20.0.0.0/12']
self._test_update_subnetpool_prefix_intersect_helper(
pool_1_prefix, pool_2_prefix, pool_1_updated_prefix)
@test.attr(type=['negative', 'smoke'])
@test.idempotent_id('4d3f8a79-c530-4e59-9acf-6c05968adbfe')
def test_update_subnetpool_multiple_prefix_intersect(self):
pool_1_prefixes = [u'20.0.0.0/18', u'30.0.0.0/18']
pool_2_prefixes = [u'20.10.0.0/24', u'40.0.0.0/18', '50.0.0.0/18']
pool_1_updated_prefixes = [u'20.0.0.0/18', u'30.0.0.0/18',
u'50.0.0.0/12']
self._test_update_subnetpool_prefix_intersect_helper(
pool_1_prefixes, pool_2_prefixes, pool_1_updated_prefixes)
@test.attr(type=['negative', 'smoke'])
@test.idempotent_id('7438e49e-1351-45d8-937b-892059fb97f5')
def test_tenant_update_sp_prefix_associated_with_shared_addr_scope(self):
address_scope = self.create_address_scope(
name=data_utils.rand_name('smoke-address-scope'), is_admin=True,
shared=True)
addr_scope_id = address_scope['id']
pool_values = {'prefixes': [u'20.0.0.0/18', u'30.0.0.0/18']}
pool_id = self._create_subnetpool(
self.client, pool_values=pool_values)
self.addCleanup(self.client.delete_subnetpool, pool_id)
# associate the subnetpool to the address scope as an admin
subnetpool_data = {'subnetpool': {'address_scope_id':
addr_scope_id}}
self.admin_client.update_subnetpool(pool_id, subnetpool_data)
body = self.admin_client.get_subnetpool(pool_id)
self.assertEqual(addr_scope_id,
body['subnetpool']['address_scope_id'])
# updating the subnetpool prefix by the tenant user should fail
# since the tenant is not the owner of address scope
update_prefixes = [u'20.0.0.0/18', u'30.0.0.0/18', u'40.0.0.0/18']
subnetpool_data = {'subnetpool': {'prefixes': update_prefixes}}
self.assertRaises(lib_exc.BadRequest, self.client.update_subnetpool,
pool_id, subnetpool_data)
# admin can update the prefixes
self.admin_client.update_subnetpool(pool_id, subnetpool_data)
body = self.admin_client.get_subnetpool(pool_id)
self.assertEqual(update_prefixes,
body['subnetpool']['prefixes'])

124
neutron/tests/unit/extensions/test_address_scope.py
View File

@ -13,6 +13,7 @@
# under the License.
import contextlib
import netaddr
import webob.exc
@ -237,3 +238,126 @@ class TestAddressScope(AddressScopeTestCase):
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_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)