openstack/keystone
Code Issues Proposed changes

Expose unified limit APIs

Browse Source 
This patch exposed all the unified limit APIs.

Change-Id: Iceb85e5c4cb92a4b393b373ccf6b41fc22d1c4c6
bp: unified-limits
This commit is contained in:
wangxiyuan 2017-12-15 17:27:30 +08:00
parent 9ba24b91a4
commit 4606183c8d
8 changed files with 1216 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
keystone/common/json_home.py
View File

@ -54,6 +54,8 @@ class Parameters(object):
SERVICE_ID = build_v3_parameter_relation('service_id')
USER_ID = build_v3_parameter_relation('user_id')
TAG_VALUE = build_v3_parameter_relation('tag_value')
REGISTERED_LIMIT_ID = build_v3_parameter_relation('registered_limit_id')
LIMIT_ID = build_v3_parameter_relation('limit_id')
class Status(object):

130
keystone/limit/controllers.py Normal file
View File

@ -0,0 +1,130 @@
# Copyright 2018 Huawei
# 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 keystone.common import controller
from keystone.common import validation
from keystone import exception
from keystone.i18n import _
from keystone.limit import schema
class RegisteredLimitV3(controller.V3Controller):
collection_name = 'registered_limits'
member_name = 'registered_limit'
def __init__(self):
super(RegisteredLimitV3, self).__init__()
self.get_member_from_driver = (
self.unified_limit_api.get_registered_limit
)
@controller.protected()
def create_registered_limits(self, request, registered_limits):
validation.lazy_validate(schema.registered_limit_create,
registered_limits)
registered_limits = [self._assign_unique_id(self._normalize_dict(
registered_limit)) for registered_limit in registered_limits]
refs = self.unified_limit_api.create_registered_limits(
registered_limits)
refs = RegisteredLimitV3.wrap_collection(request.context_dict, refs)
refs.pop("links")
return refs
@controller.protected()
def update_registered_limits(self, request, registered_limits):
validation.lazy_validate(schema.registered_limit_update,
registered_limits)
refs = self.unified_limit_api.update_registered_limits(
[self._normalize_dict(registered_limit) for registered_limit in
registered_limits])
refs = RegisteredLimitV3.wrap_collection(request.context_dict, refs)
refs.pop("links")
return refs
@controller.filterprotected('service_id', 'region_id', 'resource_name')
def list_registered_limits(self, request, filters):
hints = RegisteredLimitV3.build_driver_hints(request, filters)
refs = self.unified_limit_api.list_registered_limits(hints)
return RegisteredLimitV3.wrap_collection(request.context_dict, refs,
hints=hints)
@controller.protected()
def get_registered_limit(self, request, registered_limit_id):
ref = self.unified_limit_api.get_registered_limit(
registered_limit_id)
return RegisteredLimitV3.wrap_member(request.context_dict, ref)
@controller.protected()
def delete_registered_limit(self, request, registered_limit_id):
return self.unified_limit_api.delete_registered_limit(
registered_limit_id)
class LimitV3(controller.V3Controller):
collection_name = 'limits'
member_name = 'limit'
def __init__(self):
super(LimitV3, self).__init__()
self.get_member_from_driver = self.unified_limit_api.get_limit
@controller.protected()
def create_limits(self, request, limits):
validation.lazy_validate(schema.limit_create, limits)
limits = [self._assign_unique_id(self._normalize_dict(limit))
for limit in limits]
refs = self.unified_limit_api.create_limits(limits)
refs = LimitV3.wrap_collection(request.context_dict, refs)
refs.pop("links")
return refs
@controller.protected()
def update_limits(self, request, limits):
validation.lazy_validate(schema.limit_update, limits)
refs = self.unified_limit_api.update_limits(
[self._normalize_dict(limit) for limit in limits])
refs = LimitV3.wrap_collection(request.context_dict, refs)
refs.pop("links")
return refs
@controller.filterprotected('service_id', 'region_id', 'resource_name')
def list_limits(self, request, filters):
hints = LimitV3.build_driver_hints(request, filters)
# TODO(wxy): Add system-scope check. If the request is system-scoped,
# it can get all limits.
context = request.context
if not context.is_admin and not ('admin' in context.roles):
project_id = context.project_id
if project_id:
hints.add_filter('project_id', project_id)
refs = self.unified_limit_api.list_limits(hints)
return LimitV3.wrap_collection(request.context_dict, refs, hints=hints)
@controller.protected()
def get_limit(self, request, limit_id):
ref = self.unified_limit_api.get_limit(limit_id)
# TODO(wxy): Add system-scope check. If the request is system-scoped,
# it can get any limits.
context = request.context
if not context.is_admin and not ('admin' in context.roles):
project_id = context.project_id
if project_id and project_id != ref['project_id']:
action = _("The authenticated project should match the "
"project_id")
raise exception.Forbidden(action=action)
return LimitV3.wrap_member(request.context_dict, ref)
@controller.protected()
def delete_limit(self, request, limit_id):
return self.unified_limit_api.delete_limit(limit_id)

66
keystone/limit/routers.py Normal file
View File

@ -0,0 +1,66 @@
# Copyright 2018 Huawei
#
# 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 keystone.common import json_home
from keystone.common import wsgi
from keystone.limit import controllers
class Routers(wsgi.RoutersBase):
def append_v3_routers(self, mapper, routers):
self._add_resource(
mapper, controllers.RegisteredLimitV3(),
path='/registered_limits',
post_action='create_registered_limits',
put_action='update_registered_limits',
get_head_action='list_registered_limits',
status=json_home.Status.EXPERIMENTAL,
rel=json_home.build_v3_resource_relation('registered_limits')
)
self._add_resource(
mapper, controllers.RegisteredLimitV3(),
path='/registered_limits/{registered_limit_id}',
get_head_action='get_registered_limit',
delete_action='delete_registered_limit',
status=json_home.Status.EXPERIMENTAL,
rel=json_home.build_v3_resource_relation('registered_limits'),
path_vars={
'registered_limit_id':
json_home.Parameters.REGISTERED_LIMIT_ID}
)
self._add_resource(
mapper, controllers.LimitV3(),
path='/limits',
post_action='create_limits',
put_action='update_limits',
get_head_action='list_limits',
status=json_home.Status.EXPERIMENTAL,
rel=json_home.build_v3_resource_relation('limits')
)
self._add_resource(
mapper, controllers.LimitV3(),
path='/limits/{limit_id}',
get_head_action='get_limit',
delete_action='delete_limit',
status=json_home.Status.EXPERIMENTAL,
rel=json_home.build_v3_resource_relation('limits'),
path_vars={
'limit_id':
json_home.Parameters.LIMIT_ID}
)

116
keystone/limit/schema.py Normal file
View File

@ -0,0 +1,116 @@
# Copyright 2018 Huawei
#
# 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 keystone.common.validation import parameter_types
_registered_limit_create_properties = {
'service_id': parameter_types.id_string,
'region_id': {
'type': 'string'
},
'resource_name': {
'type': 'string'
},
'default_limit': {
'type': 'integer'
}
}
_registered_limit_create = {
'type': 'object',
'properties': _registered_limit_create_properties,
'additionalProperties': False,
'required': ['service_id', 'resource_name', 'default_limit']
}
registered_limit_create = {
'type': 'array',
'items': _registered_limit_create,
'minItems': 1
}
_registered_limit_update_properties = {
'id': parameter_types.id_string,
'service_id': parameter_types.id_string,
'region_id': {
'type': 'string'
},
'resource_name': {
'type': 'string'
},
'default_limit': {
'type': 'integer'
}
}
_registered_limit_update = {
'type': 'object',
'properties': _registered_limit_update_properties,
'additionalProperties': False,
'required': ['id', ]
}
registered_limit_update = {
'type': 'array',
'items': _registered_limit_update,
'minItems': 1
}
_limit_create_properties = {
'project_id': parameter_types.id_string,
'service_id': parameter_types.id_string,
'region_id': {
'type': 'string'
},
'resource_name': {
'type': 'string'
},
'resource_limit': {
'type': 'integer'
}
}
_limit_create = {
'type': 'object',
'properties': _limit_create_properties,
'additionalProperties': False,
'required': ['project_id', 'service_id', 'resource_name', 'resource_limit']
}
limit_create = {
'type': 'array',
'items': _limit_create,
'minItems': 1
}
_limit_update_properties = {
'id': parameter_types.id_string,
'resource_limit': {
'type': 'integer'
}
}
_limit_update = {
'type': 'object',
'properties': _limit_update_properties,
'additionalProperties': False,
'required': ['id', 'resource_limit']
}
limit_update = {
'type': 'array',
'items': _limit_update,
'minItems': 1
}

667
keystone/tests/unit/test_limits.py Normal file
View File

@ -0,0 +1,667 @@
# Copyright 2018 Huawei
#
# 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 six.moves import http_client
import uuid
from keystone.common import provider_api
from keystone.tests import unit
from keystone.tests.unit import test_v3
from keystone.tests.unit import utils as test_utils
PROVIDERS = provider_api.ProviderAPIs
class RegisteredLimitsTestCase(test_v3.RestfulTestCase):
"""Test registered_limits CRUD."""
def setUp(self):
super(RegisteredLimitsTestCase, self).setUp()
# There is already a sample service and region created from
# load_sample_data() but we're going to create another service and
# region for specific testing purposes.
response = self.post('/regions', body={'region': {}})
self.region2 = response.json_body['region']
self.region_id2 = self.region2['id']
service_ref = {'service': {
'name': uuid.uuid4().hex,
'enabled': True,
'type': 'type2'}}
response = self.post('/services', body=service_ref)
self.service2 = response.json_body['service']
self.service_id2 = self.service2['id']
def test_create_registered_limit(self):
ref = unit.new_registered_limit_ref(service_id=self.service_id,
region_id=self.region_id)
r = self.post(
'/registered_limits',
body={'registered_limits': [ref]},
expected_status=http_client.CREATED)
registered_limits = r.result['registered_limits']
for key in ['service_id', 'region_id', 'resource_name',
'default_limit']:
self.assertEqual(registered_limits[0][key], ref[key])
def test_create_registered_limit_without_region(self):
ref = unit.new_registered_limit_ref(service_id=self.service_id)
r = self.post(
'/registered_limits',
body={'registered_limits': [ref]},
expected_status=http_client.CREATED)
registered_limits = r.result['registered_limits']
for key in ['service_id', 'resource_name', 'default_limit']:
self.assertEqual(registered_limits[0][key], ref[key])
self.assertIsNone(registered_limits[0].get('region_id'))
def test_create_multi_registered_limit(self):
ref1 = unit.new_registered_limit_ref(service_id=self.service_id,
region_id=self.region_id,
resource_name='volume')
ref2 = unit.new_registered_limit_ref(service_id=self.service_id,
resource_name='snapshot')
r = self.post(
'/registered_limits',
body={'registered_limits': [ref1, ref2]},
expected_status=http_client.CREATED)
registered_limits = r.result['registered_limits']
for key in ['service_id', 'resource_name', 'default_limit']:
self.assertEqual(registered_limits[0][key], ref1[key])
self.assertEqual(registered_limits[1][key], ref2[key])
self.assertEqual(registered_limits[0]['region_id'], ref1['region_id'])
self.assertIsNone(registered_limits[1].get('region_id'))
def test_create_registered_limit_with_invalid_input(self):
ref1 = unit.new_registered_limit_ref()
ref2 = unit.new_registered_limit_ref(default_limit='not_int')
ref3 = unit.new_registered_limit_ref(resource_name=123)
ref4 = unit.new_registered_limit_ref(region_id='fake_region')
for input_limit in [ref1, ref2, ref3, ref4]:
self.post(
'/registered_limits',
body={'registered_limits': [input_limit]},
expected_status=http_client.BAD_REQUEST)
def test_create_registered_limit_duplicate(self):
ref = unit.new_registered_limit_ref(service_id=self.service_id,
region_id=self.region_id)
self.post(
'/registered_limits',
body={'registered_limits': [ref]},
expected_status=http_client.CREATED)
self.post(
'/registered_limits',
body={'registered_limits': [ref]},
expected_status=http_client.CONFLICT)
def test_update_registered_limit(self):
ref = unit.new_registered_limit_ref(service_id=self.service_id,
region_id=self.region_id,
resource_name='volume',
default_limit=10)
r = self.post(
'/registered_limits',
body={'registered_limits': [ref]},
expected_status=http_client.CREATED)
update_ref = {
'id': r.result['registered_limits'][0]['id'],
'service_id': self.service_id2,
'region_id': self.region_id2,
'resource_name': 'snapshot',
'default_limit': 5
}
r = self.put(
'/registered_limits',
body={'registered_limits': [update_ref]},
expected_status=http_client.OK)
new_registered_limits = r.result['registered_limits'][0]
self.assertEqual(new_registered_limits['service_id'], self.service_id2)
self.assertEqual(new_registered_limits['region_id'], self.region_id2)
self.assertEqual(new_registered_limits['resource_name'], 'snapshot')
self.assertEqual(new_registered_limits['default_limit'], 5)
def test_update_multi_registered_limit(self):
ref = unit.new_registered_limit_ref(service_id=self.service_id,
region_id=self.region_id,
resource_name='volume',
default_limit=10)
ref2 = unit.new_registered_limit_ref(service_id=self.service_id,
region_id=self.region_id,
resource_name='snapshot',
default_limit=10)
r = self.post(
'/registered_limits',
body={'registered_limits': [ref, ref2]},
expected_status=http_client.CREATED)
update_ref = {
'id': r.result['registered_limits'][0]['id'],
'service_id': self.service_id2,
'region_id': self.region_id2,
'resource_name': 'snapshot',
'default_limit': 5
}
update_ref2 = {
'id': r.result['registered_limits'][1]['id'],
'service_id': self.service_id2,
'region_id': self.region_id2,
'resource_name': 'volume',
'default_limit': 5
}
r = self.put(
'/registered_limits',
body={'registered_limits': [update_ref, update_ref2]},
expected_status=http_client.OK)
new_registered_limits = r.result['registered_limits']
for key in ['id', 'service_id', 'region_id', 'resource_name',
'default_limit']:
self.assertEqual(new_registered_limits[0][key], update_ref[key])
self.assertEqual(new_registered_limits[1][key], update_ref2[key])
def test_update_registered_limit_not_found(self):
update_ref = {
'id': uuid.uuid4().hex,
'service_id': self.service_id,
'region_id': self.region_id,
'resource_name': 'snapshot',
'default_limit': 5
}
self.put(
'/registered_limits',
body={'registered_limits': [update_ref]},
expected_status=http_client.NOT_FOUND)
def test_update_registered_limit_with_invalid_input(self):
update_ref1 = unit.new_registered_limit_ref(id=uuid.uuid4().hex,
service_id='fake_id')
update_ref2 = unit.new_registered_limit_ref(id=uuid.uuid4().hex,
default_limit='not_int')
update_ref3 = unit.new_registered_limit_ref(id=uuid.uuid4().hex,
resource_name=123)
update_ref4 = unit.new_registered_limit_ref(id=uuid.uuid4().hex,
region_id='fake_region')
for input_limit in [update_ref1, update_ref2, update_ref3,
update_ref4]:
self.put(
'/registered_limits',
body={'registered_limits': [input_limit]},
expected_status=http_client.BAD_REQUEST)
@test_utils.wip("Skipped until Bug 1744195 is resolved")
def test_update_registered_limit_with_referenced_limit(self):
ref = unit.new_registered_limit_ref(service_id=self.service_id,
region_id=self.region_id,
resource_name='volume',
default_limit=10)
r = self.post(
'/registered_limits',
body={'registered_limits': [ref]},
expected_status=http_client.CREATED)
ref = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id,
region_id=self.region_id,
resource_name='volume')
self.post(
'/limits',
body={'limits': [ref]},
expected_status=http_client.CREATED)
update_ref = {
'id': r.result['registered_limits'][0]['id'],
'service_id': self.service_id2,
'region_id': self.region_id2,
'resource_name': 'snapshot',
'default_limit': 5
}
self.put(
'/registered_limits',
body={'registered_limits': [update_ref]},
expected_status=http_client.FORBIDDEN)
def test_list_registered_limit(self):
r = self.get(
'/registered_limits',
expected_status=http_client.OK)
self.assertEqual([], r.result.get('registered_limits'))
ref1 = unit.new_registered_limit_ref(service_id=self.service_id,
resource_name='test_resource',
region_id=self.region_id)
ref2 = unit.new_registered_limit_ref(service_id=self.service_id2,
resource_name='test_resource',
region_id=self.region_id2)
r = self.post(
'/registered_limits',
body={'registered_limits': [ref1, ref2]},
expected_status=http_client.CREATED)
id1 = r.result['registered_limits'][0]['id']
r = self.get(
'/registered_limits',
expected_status=http_client.OK)
registered_limits = r.result['registered_limits']
self.assertEqual(len(registered_limits), 2)
for key in ['service_id', 'region_id', 'resource_name',
'default_limit']:
if registered_limits[0]['id'] == id1:
self.assertEqual(registered_limits[0][key], ref1[key])
self.assertEqual(registered_limits[1][key], ref2[key])
break
self.assertEqual(registered_limits[1][key], ref1[key])
self.assertEqual(registered_limits[0][key], ref2[key])
r = self.get(
'/registered_limits?service_id=%s' % self.service_id,
expected_status=http_client.OK)
registered_limits = r.result['registered_limits']
self.assertEqual(len(registered_limits), 1)
for key in ['service_id', 'region_id', 'resource_name',
'default_limit']:
self.assertEqual(registered_limits[0][key], ref1[key])
r = self.get(
'/registered_limits?region_id=%s' % self.region_id2,
expected_status=http_client.OK)
registered_limits = r.result['registered_limits']
self.assertEqual(len(registered_limits), 1)
for key in ['service_id', 'region_id', 'resource_name',
'default_limit']:
self.assertEqual(registered_limits[0][key], ref2[key])
r = self.get(
'/registered_limits?resource_name=test_resource',
expected_status=http_client.OK)
registered_limits = r.result['registered_limits']
self.assertEqual(len(registered_limits), 2)
def test_show_registered_limit(self):
ref1 = unit.new_registered_limit_ref(service_id=self.service_id,
region_id=self.region_id)
ref2 = unit.new_registered_limit_ref(service_id=self.service_id2,
region_id=self.region_id2)
r = self.post(
'/registered_limits',
body={'registered_limits': [ref1, ref2]},
expected_status=http_client.CREATED)
id1 = r.result['registered_limits'][0]['id']
self.get(
'/registered_limits/fake_id',
expected_status=http_client.NOT_FOUND)
r = self.get(
'/registered_limits/%s' % id1,
expected_status=http_client.OK)
registered_limit = r.result['registered_limit']
for key in ['service_id', 'region_id', 'resource_name',
'default_limit']:
self.assertEqual(registered_limit[key], ref1[key])
def test_delete_registered_limit(self):
ref1 = unit.new_registered_limit_ref(service_id=self.service_id,
region_id=self.region_id)
ref2 = unit.new_registered_limit_ref(service_id=self.service_id2,
region_id=self.region_id2)
r = self.post(
'/registered_limits',
body={'registered_limits': [ref1, ref2]},
expected_status=http_client.CREATED)
id1 = r.result['registered_limits'][0]['id']
self.delete('/registered_limits/%s' % id1,
expected_status=http_client.NO_CONTENT)
self.delete('/registered_limits/fake_id',
expected_status=http_client.NOT_FOUND)
r = self.get(
'/registered_limits',
expected_status=http_client.OK)
registered_limits = r.result['registered_limits']
self.assertEqual(len(registered_limits), 1)
@test_utils.wip("Skipped until Bug 1744195 is resolved")
def test_delete_registered_limit_with_referenced_limit(self):
ref = unit.new_registered_limit_ref(service_id=self.service_id,
region_id=self.region_id,
resource_name='volume',
default_limit=10)
r = self.post(
'/registered_limits',
body={'registered_limits': [ref]},
expected_status=http_client.CREATED)
ref = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id,
region_id=self.region_id,
resource_name='volume')
self.post(
'/limits',
body={'limits': [ref]},
expected_status=http_client.CREATED)
id = r.result['registered_limits'][0]['id']
self.delete('/registered_limits/%s' % id,
expected_status=http_client.FORBIDDEN)
class LimitsTestCase(test_v3.RestfulTestCase):
"""Test limits CRUD."""
def setUp(self):
super(LimitsTestCase, self).setUp()
# There is already a sample service and region created from
# load_sample_data() but we're going to create another service and
# region for specific testing purposes.
response = self.post('/regions', body={'region': {}})
self.region2 = response.json_body['region']
self.region_id2 = self.region2['id']
service_ref = {'service': {
'name': uuid.uuid4().hex,
'enabled': True,
'type': 'type2'}}
response = self.post('/services', body=service_ref)
self.service2 = response.json_body['service']
self.service_id2 = self.service2['id']
ref1 = unit.new_registered_limit_ref(service_id=self.service_id,
region_id=self.region_id,
resource_name='volume')
ref2 = unit.new_registered_limit_ref(service_id=self.service_id2,
resource_name='snapshot')
r = self.post(
'/registered_limits',
body={'registered_limits': [ref1, ref2]},
expected_status=http_client.CREATED)
self.registered_limit1 = r.result['registered_limits'][0]
self.registered_limit2 = r.result['registered_limits'][1]
def test_create_limit(self):
ref = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id,
region_id=self.region_id,
resource_name='volume')
r = self.post(
'/limits',
body={'limits': [ref]},
expected_status=http_client.CREATED)
limits = r.result['limits']
self. assertIsNotNone(limits[0]['id'])
self. assertIsNotNone(limits[0]['project_id'])
for key in ['service_id', 'region_id', 'resource_name',
'resource_limit']:
self.assertEqual(limits[0][key], ref[key])
def test_create_limit_without_region(self):
ref = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id2,
resource_name='snapshot')
r = self.post(
'/limits',
body={'limits': [ref]},
expected_status=http_client.CREATED)
limits = r.result['limits']
self. assertIsNotNone(limits[0]['id'])
self. assertIsNotNone(limits[0]['project_id'])
for key in ['service_id', 'resource_name', 'resource_limit']:
self.assertEqual(limits[0][key], ref[key])
self.assertIsNone(limits[0].get('region_id'))
def test_create_multi_limit(self):
ref1 = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id,
region_id=self.region_id,
resource_name='volume')
ref2 = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id2,
resource_name='snapshot')
r = self.post(
'/limits',
body={'limits': [ref1, ref2]},
expected_status=http_client.CREATED)
limits = r.result['limits']
for key in ['service_id', 'resource_name', 'resource_limit']:
self.assertEqual(limits[0][key], ref1[key])
self.assertEqual(limits[1][key], ref2[key])
self.assertEqual(limits[0]['region_id'], ref1['region_id'])
self.assertIsNone(limits[1].get('region_id'))
def test_create_limit_with_invalid_input(self):
ref1 = unit.new_limit_ref(project_id=self.project_id,
resource_limit='not_int')
ref2 = unit.new_limit_ref(project_id=self.project_id,
resource_name=123)
ref3 = unit.new_limit_ref(project_id=self.project_id,
region_id='fake_region')
for input_limit in [ref1, ref2, ref3]:
self.post(
'/limits',
body={'limits': [input_limit]},
expected_status=http_client.BAD_REQUEST)
def test_create_limit_duplicate(self):
ref = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id,
region_id=self.region_id,
resource_name='volume')
self.post(
'/limits',
body={'limits': [ref]},
expected_status=http_client.CREATED)
self.post(
'/limits',
body={'limits': [ref]},
expected_status=http_client.CONFLICT)
@test_utils.wip("Skipped until Bug 1744195 is resolved")
def test_create_limit_without_reference_registered_limit(self):
ref = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id,
region_id=self.region_id2,
resource_name='volume')
self.post(
'/limits',
body={'limits': [ref]},
expected_status=http_client.FORBIDDEN)
def test_update_limit(self):
ref = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id,
region_id=self.region_id,
resource_name='volume',
resource_limit=10)
r = self.post(
'/limits',
body={'limits': [ref]},
expected_status=http_client.CREATED)
update_ref = {
'id': r.result['limits'][0]['id'],
'resource_limit': 5
}
r = self.put(
'/limits',
body={'limits': [update_ref]},
expected_status=http_client.OK)
new_limits = r.result['limits'][0]
self.assertEqual(new_limits['resource_limit'], 5)
def test_update_multi_limit(self):
ref = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id,
region_id=self.region_id,
resource_name='volume',
resource_limit=10)
ref2 = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id2,
resource_name='snapshot',
resource_limit=10)
r = self.post(
'/limits',
body={'limits': [ref, ref2]},
expected_status=http_client.CREATED)
id1 = r.result['limits'][0]['id']
update_ref = {
'id': id1,
'resource_limit': 5
}
update_ref2 = {
'id': r.result['limits'][1]['id'],
'resource_limit': 6
}
r = self.put(
'/limits',
body={'limits': [update_ref, update_ref2]},
expected_status=http_client.OK)
new_limits = r.result['limits']
for limit in new_limits:
if limit['id'] == id1:
self.assertEqual(limit['resource_limit'],
update_ref['resource_limit'])
else:
self.assertEqual(limit['resource_limit'],
update_ref2['resource_limit'])
def test_update_limit_not_found(self):
update_ref = {
'id': uuid.uuid4().hex,
'resource_limit': 5
}
self.put(
'/limits',
body={'limits': [update_ref]},
expected_status=http_client.NOT_FOUND)
def test_update_limit_with_invalid_input(self):
update_ref1 = {
'id': 'fake_id',
'resource_limit': 5
}
update_ref2 = {
'id': uuid.uuid4().hex,
'resource_limit': 'not_int'
}
for input_limit in [update_ref1, update_ref2]:
self.put(
'/limits',
body={'limits': [input_limit]},
expected_status=http_client.BAD_REQUEST)
def test_list_limit(self):
r = self.get(
'/limits',
expected_status=http_client.OK)
self.assertEqual([], r.result.get('limits'))
ref1 = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id,
region_id=self.region_id,
resource_name='volume')
ref2 = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id2,
resource_name='snapshot')
r = self.post(
'/limits',
body={'limits': [ref1, ref2]},
expected_status=http_client.CREATED)
id1 = r.result['limits'][0]['id']
r = self.get(
'/limits',
expected_status=http_client.OK)
limits = r.result['limits']
self.assertEqual(len(limits), 2)
if limits[0]['id'] == id1:
self.assertEqual(limits[0]['region_id'], ref1['region_id'])
self.assertIsNone(limits[1].get('region_id'))
for key in ['service_id', 'resource_name', 'resource_limit']:
self.assertEqual(limits[0][key], ref1[key])
self.assertEqual(limits[1][key], ref2[key])
else:
self.assertEqual(limits[1]['region_id'], ref1['region_id'])
self.assertIsNone(limits[0].get('region_id'))
for key in ['service_id', 'resource_name', 'resource_limit']:
self.assertEqual(limits[1][key], ref1[key])
self.assertEqual(limits[0][key], ref2[key])
r = self.get(
'/limits?service_id=%s' % self.service_id2,
expected_status=http_client.OK)
limits = r.result['limits']
self.assertEqual(len(limits), 1)
for key in ['service_id', 'resource_name', 'resource_limit']:
self.assertEqual(limits[0][key], ref2[key])
r = self.get(
'/limits?region_id=%s' % self.region_id,
expected_status=http_client.OK)
limits = r.result['limits']
self.assertEqual(len(limits), 1)
for key in ['service_id', 'region_id', 'resource_name',
'resource_limit']:
self.assertEqual(limits[0][key], ref1[key])
r = self.get(
'/limits?resource_name=volume',
expected_status=http_client.OK)
limits = r.result['limits']
self.assertEqual(len(limits), 1)
for key in ['service_id', 'region_id', 'resource_name',
'resource_limit']:
self.assertEqual(limits[0][key], ref1[key])
def test_show_limit(self):
ref1 = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id,
region_id=self.region_id,
resource_name='volume')
ref2 = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id2,
resource_name='snapshot')
r = self.post(
'/limits',
body={'limits': [ref1, ref2]},
expected_status=http_client.CREATED)
id1 = r.result['limits'][0]['id']
self.get('/limits/fake_id',
expected_status=http_client.NOT_FOUND)
r = self.get('/limits/%s' % id1,
expected_status=http_client.OK)
limit = r.result['limit']
for key in ['service_id', 'region_id', 'resource_name',
'resource_limit']:
self.assertEqual(limit[key], ref1[key])
def test_delete_limit(self):
ref1 = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id,
region_id=self.region_id,
resource_name='volume')
ref2 = unit.new_limit_ref(project_id=self.project_id,
service_id=self.service_id2,
resource_name='snapshot')
r = self.post(
'/limits',
body={'limits': [ref1, ref2]},
expected_status=http_client.CREATED)
id1 = r.result['limits'][0]['id']
self.delete('/limits/%s' % id1,
expected_status=http_client.NO_CONTENT)
self.delete('/limits/fake_id',
expected_status=http_client.NOT_FOUND)
r = self.get(
'/limits',
expected_status=http_client.OK)
limits = r.result['limits']
self.assertEqual(len(limits), 1)

217
keystone/tests/unit/test_validation.py
View File

@ -23,6 +23,7 @@ from keystone import exception
from keystone.federation import schema as federation_schema
from keystone.identity.backends import resource_options as ro
from keystone.identity import schema as identity_schema
from keystone.limit import schema as limit_schema
from keystone.oauth1 import schema as oauth1_schema
from keystone.policy import schema as policy_schema
from keystone.resource import schema as resource_schema
@ -2455,3 +2456,219 @@ class PasswordValidationTestCase(unit.TestCase):
self.config_fixture.config(group='security_compliance',
password_regex='[\S]+')
validators.validate_password(password)
class LimitValidationTestCase(unit.BaseTestCase):
"""Test for V3 Limits API validation."""
def setUp(self):
super(LimitValidationTestCase, self).setUp()
create_registered_limits = limit_schema.registered_limit_create
update_registered_limits = limit_schema.registered_limit_update
create_limits = limit_schema.limit_create
update_limits = limit_schema.limit_update
self.create_registered_limits_validator = validators.SchemaValidator(
create_registered_limits)
self.update_registered_limits_validator = validators.SchemaValidator(
update_registered_limits)
self.create_limits_validator = validators.SchemaValidator(
create_limits)
self.update_limits_validator = validators.SchemaValidator(
update_limits)
def test_validate_registered_limit_create_request_succeeds(self):
request_to_validate = [{'service_id': uuid.uuid4().hex,
'region_id': 'RegionOne',
'resource_name': 'volume',
'default_limit': 10}]
self.create_registered_limits_validator.validate(request_to_validate)
def test_validate_registered_limit_create_request_without_region(self):
request_to_validate = [{'service_id': uuid.uuid4().hex,
'resource_name': 'volume',
'default_limit': 10}]
self.create_registered_limits_validator.validate(request_to_validate)
def test_validate_registered_limit_update_request_without_region(self):
request_to_validate = [{'id': uuid.uuid4().hex,
'service_id': uuid.uuid4().hex,
'resource_name': 'volume',
'default_limit': 10}]
self.update_registered_limits_validator.validate(request_to_validate)
def test_validate_registered_limit_request_with_no_parameters(self):
request_to_validate = []
# At least one property should be given.
self.assertRaises(exception.SchemaValidationError,
self.create_registered_limits_validator.validate,
request_to_validate)
self.assertRaises(exception.SchemaValidationError,
self.update_registered_limits_validator.validate,
request_to_validate)
def test_validate_registered_limit_create_request_with_invalid_input(self):
_INVALID_FORMATS = [{'service_id': 'fake_id'},
{'region_id': 123},
{'resource_name': 123},
{'default_limit': 'not_int'}]
for invalid_desc in _INVALID_FORMATS:
request_to_validate = [{'service_id': uuid.uuid4().hex,
'region_id': 'RegionOne',
'resource_name': 'volume',
'default_limit': 10}]
request_to_validate[0].update(invalid_desc)
self.assertRaises(exception.SchemaValidationError,
self.create_registered_limits_validator.validate,
request_to_validate)
def test_validate_registered_limit_update_request_with_invalid_input(self):
_INVALID_FORMATS = [{'service_id': 'fake_id'},
{'region_id': 123},
{'resource_name': 123},
{'default_limit': 'not_int'}]
for invalid_desc in _INVALID_FORMATS:
request_to_validate = [{'id': uuid.uuid4().hex,
'service_id': uuid.uuid4().hex,
'region_id': 'RegionOne',
'resource_name': 'volume',
'default_limit': 10}]
request_to_validate[0].update(invalid_desc)
self.assertRaises(exception.SchemaValidationError,
self.update_registered_limits_validator.validate,
request_to_validate)
def test_validate_registered_limit_create_request_with_addition(self):
request_to_validate = [{'service_id': uuid.uuid4().hex,
'region_id': 'RegionOne',
'resource_name': 'volume',
'default_limit': 10,
'more_key': 'more_value'}]
self.assertRaises(exception.SchemaValidationError,
self.create_registered_limits_validator.validate,
request_to_validate)
def test_validate_registered_limit_update_request_with_addition(self):
request_to_validate = [{'id': uuid.uuid4().hex,
'service_id': uuid.uuid4().hex,
'region_id': 'RegionOne',
'resource_name': 'volume',
'default_limit': 10,
'more_key': 'more_value'}]
self.assertRaises(exception.SchemaValidationError,
self.update_registered_limits_validator.validate,
request_to_validate)
def test_validate_registered_limit_create_request_without_required(self):
for key in ['service_id', 'resource_name', 'default_limit']:
request_to_validate = [{'service_id': uuid.uuid4().hex,
'region_id': 'RegionOne',
'resource_name': 'volume',
'default_limit': 10}]
request_to_validate[0].pop(key)
self.assertRaises(exception.SchemaValidationError,
self.create_registered_limits_validator.validate,
request_to_validate)
def test_validate_registered_limit_update_request_without_id(self):
request_to_validate = [{'service_id': uuid.uuid4().hex,
'region_id': 'RegionOne',
'resource_name': 'volume',
'default_limit': 10}]
self.assertRaises(exception.SchemaValidationError,
self.update_registered_limits_validator.validate,
request_to_validate)
def test_validate_limit_create_request_succeeds(self):
request_to_validate = [{'project_id': uuid.uuid4().hex,
'service_id': uuid.uuid4().hex,
'region_id': 'RegionOne',
'resource_name': 'volume',
'resource_limit': 10}]
self.create_limits_validator.validate(request_to_validate)
def test_validate_limit_create_request_without_region(self):
request_to_validate = [{'project_id': uuid.uuid4().hex,
'service_id': uuid.uuid4().hex,
'resource_name': 'volume',
'resource_limit': 10}]
self.create_limits_validator.validate(request_to_validate)
def test_validate_limit_update_request_succeeds(self):
request_to_validate = [{'id': uuid.uuid4().hex,
'resource_limit': 10}]
self.update_limits_validator.validate(request_to_validate)
def test_validate_limit_request_with_no_parameters(self):
request_to_validate = []
# At least one property should be given.
self.assertRaises(exception.SchemaValidationError,
self.create_limits_validator.validate,
request_to_validate)
self.assertRaises(exception.SchemaValidationError,
self.update_limits_validator.validate,
request_to_validate)
def test_validate_limit_create_request_with_invalid_input(self):
_INVALID_FORMATS = [{'project_id': 'fake_id'},
{'service_id': 'fake_id'},
{'region_id': 123},
{'resource_name': 123},
{'resource_limit': 'not_int'}]
for invalid_desc in _INVALID_FORMATS:
request_to_validate = [{'project_id': uuid.uuid4().hex,
'service_id': uuid.uuid4().hex,
'region_id': 'RegionOne',
'resource_name': 'volume',
'resource_limit': 10}]
request_to_validate[0].update(invalid_desc)
self.assertRaises(exception.SchemaValidationError,
self.create_limits_validator.validate,
request_to_validate)
def test_validate_limit_update_request_with_invalid_input(self):
request_to_validate = [{'id': uuid.uuid4().hex,
'resource_limit': 'not_int'}]
self.assertRaises(exception.SchemaValidationError,
self.update_limits_validator.validate,
request_to_validate)
def test_validate_limit_create_request_with_addition_input_fails(self):
request_to_validate = [{'service_id': uuid.uuid4().hex,
'region_id': 'RegionOne',
'resource_name': 'volume',
'resource_limit': 10,
'more_key': 'more_value'}]
self.assertRaises(exception.SchemaValidationError,
self.create_limits_validator.validate,
request_to_validate)
def test_validate_limit_update_request_with_addition_input_fails(self):
request_to_validate = [{'id': uuid.uuid4().hex,
'resource_limit': 10,
'more_key': 'more_value'}]
self.assertRaises(exception.SchemaValidationError,
self.update_limits_validator.validate,
request_to_validate)
def test_validate_limit_create_request_without_required_fails(self):
for key in ['service_id', 'resource_name', 'resource_limit']:
request_to_validate = [{'service_id': uuid.uuid4().hex,
'region_id': 'RegionOne',
'resource_name': 'volume',
'resource_limit': 10}]
request_to_validate[0].pop(key)
self.assertRaises(exception.SchemaValidationError,
self.create_limits_validator.validate,
request_to_validate)
def test_validate_limit_update_request_without_id_fails(self):
request_to_validate = [{'resource_limit': 10}]
self.assertRaises(exception.SchemaValidationError,
self.update_limits_validator.validate,
request_to_validate)

17
keystone/tests/unit/test_versions.py
View File

@ -635,7 +635,22 @@ V3_JSON_HOME_RESOURCES = {
'href-template': '/domains/config/{group}/{option}/default',
'href-vars': {
'group': json_home.build_v3_parameter_relation('config_group'),
'option': json_home.build_v3_parameter_relation('config_option')}}
'option': json_home.build_v3_parameter_relation('config_option')}},
json_home.build_v3_resource_relation('registered_limits'): {
'href-template': '/registered_limits/{registered_limit_id}',
'href-vars': {
'registered_limit_id': json_home.build_v3_parameter_relation(
'registered_limit_id')
},
'hints': {'status': 'experimental'}
},
json_home.build_v3_resource_relation('limits'): {
'href-template': '/limits/{limit_id}',
'href-vars': {
'limit_id': json_home.build_v3_parameter_relation('limit_id')
},
'hints': {'status': 'experimental'}
},
}

2
keystone/version/service.py
View File

@ -28,6 +28,7 @@ from keystone.credential import routers as credential_routers
from keystone.endpoint_policy import routers as endpoint_policy_routers
from keystone.federation import routers as federation_routers
from keystone.identity import routers as identity_routers
from keystone.limit import routers as limit_routers
from keystone.oauth1 import routers as oauth1_routers
from keystone.policy import routers as policy_routers
from keystone.resource import routers as resource_routers
@ -126,6 +127,7 @@ def v3_app_factory(global_conf, **local_conf):
catalog_routers,
credential_routers,
identity_routers,
limit_routers,
policy_routers,
resource_routers,
revoke_routers,