x/kingbird
Code Issues Proposed changes

Support V1.1 for kingbird quota management.

Browse Source 
-Modified existing policy.json.
-Added enforcer.py to use policy.json and there by enforcing restrictions.
-Trimmed curl requests of version V1
-Added new working curl requests in Readme.rst
-Added quota-class management curl requests in "kingbird/api/controllers/v1/README.rst"

Partially Implements: blueprint kingbird-api-v2

Change-Id: I513a60b2cc643d6d7458cc955f135dce88637c47
This commit is contained in:
Goutham Pratapa 2017-04-10 18:23:35 +05:30
parent 7f89f08407
commit 8a453465bc
11 changed files with 1006 additions and 321 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
etc/kingbird/policy.json
View File

@ -1,8 +1,9 @@
{ {
"context_is_admin": "role:admin", "context_is_admin": "role:admin",
"admin_or_owner": "is_admin:True or project_id:%(project_id)s", "admin_or_owner": "is_admin:True or project_id:%(project_id)s",
"default": "rule:admin_or_owner", "default": "rule:admin_or_owner",
"kingbird:create_quota": "rule:admin", "kingbird:update_quota": "rule:admin",
"kingbird:update_quota": "rule:admin" "kingbird:get_all_quota": "rule:admin_only",
"kingbird:delete_quota": "rule:admin_only"
} }

88
kingbird/api/README.rst
View File

@ -27,89 +27,5 @@ app.py:
apicfg.py: apicfg.py:
API configuration loading and init API configuration loading and init
============================================== enforcer.py
Example API CURL requests for quota management Enforces policies on the version2 API's
==============================================
Note:
admin_tenant_id: Tenant ID of admin.
tenant_1: Tenant ID of the project for which we want to perform operation.
===
PUT
===
Creates/Updates quota for a project
Can be called only by Admin user
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
-X PUT \
-d '{"quota_set":{"instances":20,"cores": 20,"ram": 12300,"floating_ips": 50,"metadata_items": 200,"security_groups": 50,"security_group_rules": 50,"key_pairs": 200 }}' \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/tenant_1
======
DELETE
======
Can be called only by Admin user
1. To delete all resources for tenant_1 from DB
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
-X DELETE \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/tenant_1
2. To delete resources mentioned in quota_set for tenant_1 from DB
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
-X DELETE \
-d '{"quota_set": [ "instances", "floating_ips", "metadata_items", "security_groups", "security_group_rules", "key_pairs"]}' \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/tenant_1
===
GET
===
Can be called by both Admin/Non-Admin user
1. To get quota limits for all resources in tenant_1
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/tenant_1
2. To get the default quota limits from conf file
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/defaults
3. To get the total resource usages for a tenant
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/tanant_1/detail
==========
QUOTA SYNC - for a project
==========
Can be called only by Admin user
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
-X PUT \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/tenant_1/sync

6
kingbird/api/controllers/root.py
View File

@ -23,7 +23,11 @@ class RootController(object):
@pecan.expose('json') @pecan.expose('json')
def _lookup(self, version, *remainder): def _lookup(self, version, *remainder):
if version == 'v1.0': version = str(version)
minor_version = version[-1]
major_version = version[1]
remainder = remainder + (minor_version,)
if major_version == '1':
return v1_root.Controller(), remainder return v1_root.Controller(), remainder
@pecan.expose(generic=True, template='json') @pecan.expose(generic=True, template='json')

222
kingbird/api/controllers/v1/README.rst Executable file
View File

@ -0,0 +1,222 @@
=================================================
Example API CURL requests for quota management V1
=================================================
Note:
admin_tenant_id: Tenant ID of admin.
tenant_1: Tenant ID of the project for which we want to perform operation.
===
PUT
===
Creates/Updates quota for a project
Can be called only by Admin user
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
-X PUT \
-d '{"quota_set":{"instances":20,"cores": 20,"ram": 12300,"floating_ips": 50,"metadata_items": 200,"security_groups": 50,"security_group_rules": 50,"key_pairs": 200 }}' \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/tenant_1
======
DELETE
======
Can be called only by Admin user
1. To delete all resources for tenant_1 from DB
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
-X DELETE \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/tenant_1
2. To delete resources mentioned in quota_set for tenant_1 from DB
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
-X DELETE \
-d '{"quota_set": [ "instances", "floating_ips", "metadata_items", "security_groups", "security_group_rules", "key_pairs"]}' \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/tenant_1
===
GET
===
Can be called by both Admin/Non-Admin user
1. To get quota limits for all resources in tenant_1
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/tenant_1
2. To get the default quota limits from conf file
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/defaults
3. To get the total resource usages for a tenant
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/tanant_1/detail
==========
QUOTA SYNC - for a project
==========
Can be called only by Admin user
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
-X PUT \
http://127.0.0.1:8118/v1.0/admin_tenant_id/os-quota-sets/tenant_1/sync
=======================================================
Example API CURL requests for quota class management V1
=======================================================
===
PUT
===
Can be called only by Admin user
curl \
-H “Content-Type: application/json” \
-H “X-Auth-Token: $TOKEN” \
-H “ROLE: admin” \
-X PUT -d \
‘{“quota_class_set”:{“cores”: 100, “network”:50,”security_group”: 50,”security_group_rule”: 50}}’ \
http://$kb_ip_addr:8118/v1.0/$admin_tenant_id/os-quota-class-sets/$class_name
===
GET
===
Get default quota class
curl \
-H “Content-Type: application/json” \
-H “X-Auth-Token: $TOKEN” \
-H “X_ROLE: admin” \
http://$kb_ip_addr:8118/v1.0/$admin_tenant_id/os-quota-class-sets/$class_name
======
DELETE
======
Delete default quota class
curl \
-H “Content-Type: application/json” \
-H “X-Auth-Token: $TOKEN” \
-H “ROLE: admin” \
-X DELETE \
http://$kb_ip_addr:8118/v1.0/$admin_tenant_id/os-quota-class-sets/$class_name
=================================================
Example API CURL requests for quota management V1.1
=================================================
Note:
tenant_id: Tenant ID of the project for which we want to perform operation.
===
PUT
===
Creates/Updates quota for a project
Can be called only by Admin user
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
-X PUT \
-d '{"quota_set":{"instances":20,"cores": 20,"ram": 12300,"floating_ips": 50,"metadata_items": 200,"security_groups": 50,"security_group_rules": 50,"key_pairs": 200 }}' \
http://127.0.0.1:8118/v1.1/tenant_id/os-quota-sets
======
DELETE
======
Can be called only by Admin user
1. To delete all resources for tenant_1 from DB
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
-X DELETE \
http://127.0.0.1:8118/v1.1/tenant_id/os-quota-sets/
2. To delete resources mentioned in quota_set for tenant_1 from DB
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
-X DELETE \
http://127.0.0.1:8118/v1.1/tenant_id/os-quota-sets/?'fixed_ips=15&backups=12'
===
GET
===
Can be called by both Admin/Non-Admin user
1. To get quota limits for all resources in tenant_1
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
http://127.0.0.1:8118/v1.1/tenant_id/os-quota-sets/
2. To get the default quota limits from conf file
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
http://127.0.0.1:8118/v1.1/tenant_id/os-quota-sets/defaults
3. To get the total resource usages for a tenant
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
http://127.0.0.1:8118/v1.1/tenant_id/os-quota-sets/detail
4. To get quota limits for another tenant
Can be called only by Admin user
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
http://127.0.0.1:8118/v1.1/tenant_id/os-quota-sets/detail
==========
QUOTA SYNC - for a project
==========
Can be called only by Admin user
curl \
-H "Content-Type: application/json" \
-H "X-Auth-Token: $TOKEN" \
-H "X_ROLE: admin" \
-X PUT \
http://127.0.0.1:8118/v1.1/tenant_id/os-quota-sets/sync

499
kingbird/api/controllers/v1/quota_manager.py
View File

@ -23,6 +23,7 @@ from pecan import expose
from pecan import request from pecan import request
from kingbird.api.controllers import restcomm from kingbird.api.controllers import restcomm
from kingbird.api import enforcer as enf
from kingbird.common import exceptions from kingbird.common import exceptions
from kingbird.common.i18n import _ from kingbird.common.i18n import _
from kingbird.common import utils from kingbird.common import utils
@ -46,159 +47,47 @@ CONF.register_opt(rpc_api_cap_opt, 'upgrade_levels')
LOG = logging.getLogger(__name__) LOG = logging.getLogger(__name__)
class QuotaManagerController(object): class BaseController(object):
VERSION_ALIASES = { """Base controller of quota_management for API version 1.0 & 1.1.
'mitaka': '1.0',
} It references all other resources belonging to both the API's.
"""
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
super(QuotaManagerController, self).__init__(*args, **kwargs) super(BaseController, self).__init__(*args, **kwargs)
self.rpc_client = rpc_client.EngineClient() self.rpc_client = rpc_client.EngineClient()
# to do the version compatibility for future purpose def get_quota(self, context, project_id, action=None):
def _determine_version_cap(self, target): """Get quota for a specified tenant.
version_cap = 1.0
return version_cap
@expose(generic=True, template='json') :param context: context object.
def index(self):
# Route the request to specific methods with parameters
pass
@index.when(method='GET', template='json') :param project_id: It's UUID of the project.
def get(self, project_id, target_project_id=None, action=None): Note: In v1.0 it can be defaults sometimes.
context = restcomm.extract_context_from_environ() Only specified tenant quota is retrieved from database
valid_project_id = uuidutils.is_uuid_like(project_id) using this param.
if not valid_project_id:
pecan.abort(400, _('Invalid request URL')) :param action: Optional. If provided, it can be 'detail'
if project_id != context.project and not context.is_admin: action - Gets details quota for the specified tenant.
pecan.abort(400, _('Invalid request URL')) """
if not uuidutils.is_uuid_like(target_project_id)\
and target_project_id != 'defaults':
pecan.abort(400, _('Invalid request URL'))
quota = collections.defaultdict(dict)
result = collections.defaultdict(dict) result = collections.defaultdict(dict)
try: if project_id == 'defaults' or action == 'defaults':
if context.is_admin or (project_id == target_project_id)\ # Get default quota limits from conf file
or (target_project_id == 'defaults'): result = self.get_defaults(context,
if target_project_id == 'defaults': CONF.kingbird_global_limit)
# Get default quota limits from conf file else:
result = self._get_defaults(context, if action and action != 'detail':
CONF.kingbird_global_limit) pecan.abort(404, _('Invalid request URL'))
else: elif action == 'detail':
if action and action != 'detail': # Get the current quota usages for a project
pecan.abort(404, _('Invalid request URL')) result = self.rpc_client.get_total_usage_for_tenant(
elif action == 'detail': context, project_id)
# Get the current quota usages for a project
result = self.rpc_client.get_total_usage_for_tenant(
context, target_project_id)
else:
# Get quota limits for all the resources for a project
result = db_api.quota_get_all_by_project(
context, target_project_id)
quota['quota_set'] = result
return quota
else: else:
pecan.abort(403, _('Admin required ')) # Get quota limits for all the resources for a project
# Could be raised by get total usage call result = db_api.quota_get_all_by_project(context, project_id)
except exceptions.InternalError: return result
pecan.abort(400, _('Error while requesting usage'))
# Tries to update quota limits for a project, if it fails then def get_defaults(self, context, config_defaults):
# it creates a new entry in DB for that project
@index.when(method='PUT', template='json')
def put(self, project_id, target_project_id, action=None):
context = restcomm.extract_context_from_environ()
valid_project_id = uuidutils.is_uuid_like(project_id)
if not valid_project_id:
pecan.abort(400, _('Invalid request URL'))
if project_id != context.project and not context.is_admin:
pecan.abort(400, _('Invalid request URL'))
if not uuidutils.is_uuid_like(target_project_id):
pecan.abort(400, _('Invalid request URL'))
quota = collections.defaultdict(dict)
quota[target_project_id] = collections.defaultdict(dict)
if action and action != 'sync':
pecan.abort(404, 'Invalid action, only sync is allowed')
elif action == 'sync':
return self.sync(target_project_id, context)
if not context.is_admin:
pecan.abort(403, _('Admin required'))
if not request.body:
pecan.abort(400, _('Body required'))
payload = eval(request.body)
payload = payload.get('quota_set')
if not payload:
pecan.abort(400, _('quota_set in body is required'))
try:
utils.validate_quota_limits(payload)
for resource, limit in payload.iteritems():
try:
# Update quota limit in DB
result = db_api.quota_update(
context,
project_id=target_project_id,
resource=resource,
limit=limit)
except exceptions.ProjectQuotaNotFound:
# If update fails due to project/quota not found
# then create the quota limit
result = db_api.quota_create(
context,
project_id=target_project_id,
resource=resource,
limit=limit)
quota[target_project_id][result.resource] = result.hard_limit
return quota
except exceptions.InvalidInputError:
pecan.abort(400, _('Invalid input for quota limits'))
@index.when(method='delete', template='json')
def delete(self, project_id, target_project_id):
context = restcomm.extract_context_from_environ()
valid_project_id = uuidutils.is_uuid_like(project_id)
if not valid_project_id:
pecan.abort(400, _('Invalid request URL'))
if project_id != context.project and not context.is_admin:
pecan.abort(400, _('Invalid request URL'))
if not uuidutils.is_uuid_like(target_project_id):
pecan.abort(400, _('Invalid request URL'))
if not context.is_admin:
pecan.abort(403, _('Admin required'))
try:
if request.body:
# Delete the mentioned quota limit for the project
payload = eval(request.body)
payload = payload.get('quota_set')
if not payload:
pecan.abort(400, _('quota_set in body required'))
utils.validate_quota_limits(payload)
for resource in payload:
db_api.quota_destroy(context, target_project_id, resource)
return {'Deleted quota limits': payload}
else:
# Delete all quota limits for the project
db_api.quota_destroy_all(context, target_project_id)
return "Deleted all quota limits for the given project"
except exceptions.ProjectQuotaNotFound:
pecan.abort(404, _('Project quota not found'))
except exceptions.InvalidInputError:
pecan.abort(400, _('Invalid input for quota'))
# Private method called by put method for on demand quota sync
def sync(self, project_id, context):
if pecan.request.method != 'PUT':
pecan.abort(405, _('Bad method. Use PUT instead'))
if not context.is_admin:
pecan.abort(403, _('Admin required'))
self.rpc_client.quota_sync_for_project(
context, project_id)
return 'triggered quota sync for ' + project_id
@staticmethod
def _get_defaults(context, config_defaults):
"""Get default quota values. """Get default quota values.
If the default class is defined, use the values defined If the default class is defined, use the values defined
@ -225,5 +114,325 @@ class QuotaManagerController(object):
"default quota class for default " "default quota class for default "
"quota.") % {'res': resource_name}) "quota.") % {'res': resource_name})
quotas[resource_name] = default_quotas.get(resource_name, default) quotas[resource_name] = default_quotas.get(resource_name, default)
return quotas return quotas
def sync(self, context, project_id):
"""Sync quota of a tenant.
Private method called by put method for on demand quota sync
:param context: context object.
:param project_id: It's UUID of the project.
On demand quota sync is triggered only for specified tenant
using this param.
"""
if pecan.request.method != 'PUT':
pecan.abort(405, _('Bad method. Use PUT instead'))
self.rpc_client.quota_sync_for_project(
context, project_id)
return 'triggered quota sync for ' + project_id
def delete_quota_resources(self, context, project_id, payload):
"""Delete quota for a specified resource of a tenant.
:param context: context object.
:param project_id: It's UUID of the project.
Only specified tenant quota is retrieved from database
using this param.
:param payload: Deletes quota of specified resources for a tenant.
Note:- Support only through CURL request for V1.0.
"""
try:
# Delete the mentioned quota limit for the project
utils.validate_quota_limits(payload)
for resource in payload:
db_api.quota_destroy(context, project_id, resource)
return {'Deleted quota limits': payload}
except exceptions.ProjectQuotaNotFound:
pecan.abort(404, _('Project quota not found'))
except exceptions.InvalidInputError:
pecan.abort(400, _('Invalid input for quota'))
def delete_quota(self, context, project_id):
"""Delete entire quota for a specified tenant.
:param context: context object.
:param project_id: It's UUID of the project.
Only specified tenant quota is retrieved from database
using this param.
"""
try:
db_api.quota_destroy_all(context, project_id)
return "Deleted all quota limits for the given project"
except exceptions.ProjectQuotaNotFound:
pecan.abort(404, _('Project quota not found'))
def update_quota(self, context, request, project_id):
"""Update quota for specified tenant.
:param context: context object.
:param request: request object.
:param project_id: It's UUID of the project.
Only specified tenant quota is updated in database
using this param.
"""
quota = collections.defaultdict(dict)
quota[project_id] = collections.defaultdict(dict)
if not request.body:
pecan.abort(400, _('Body required'))
payload = eval(request.body).get('quota_set')
if not payload:
pecan.abort(400, _('quota_set in body is required'))
try:
utils.validate_quota_limits(payload)
for resource, limit in payload.iteritems():
try:
# Update quota limit in DB
result = db_api.quota_update(
context,
project_id=project_id,
resource=resource,
limit=limit)
except exceptions.ProjectQuotaNotFound:
# If update fails due to project/quota not found
# then create the quota limit
result = db_api.quota_create(
context,
project_id=project_id,
resource=resource,
limit=limit)
quota[project_id][result.resource] = result.hard_limit
return quota
except exceptions.InvalidInputError:
pecan.abort(400, _('Invalid input for quota limits'))
class QuotaManagerController(BaseController):
"""Quota Management API controller for API version 1.0.
It references all other resources belonging to the API v1.0.
"""
VERSION_ALIASES = {
'mitaka': '1.0',
}
# to do the version compatibility for future purpose
def _determine_version_cap(self, target):
version_cap = 1.0
return version_cap
@expose(generic=True, template='json')
def index(self):
# Route the request to specific methods with parameters
pass
@index.when(method='GET', template='json')
def get(self, project_id, target_project_id=None, action=None):
"""Get quota for a specified tenant.
:param project_id: It's UUID of the project.
:param target_project_id: It's UUID of the project.
Note: In v1.0 it can be defaults sometimes.
Only specified tenant quota is retrieved from database
using this param.
:param action: Optional. If provided, it can be 'detail'
detail - Gets detail quota usage for the specified tenant.
"""
context = restcomm.extract_context_from_environ()
valid_project_id = uuidutils.is_uuid_like(project_id)
if not valid_project_id:
pecan.abort(400, _('Invalid request URL'))
if project_id != context.project and not context.is_admin:
pecan.abort(400, _('Invalid request URL'))
if not uuidutils.is_uuid_like(target_project_id)\
and target_project_id != 'defaults':
pecan.abort(400, _('Invalid request URL'))
quota = collections.defaultdict(dict)
try:
if context.is_admin or (project_id == target_project_id)\
or (target_project_id == 'defaults'):
result = self.get_quota(context, target_project_id, action)
quota['quota_set'] = result
return quota
else:
pecan.abort(403, _('Admin required '))
except exceptions.InternalError:
pecan.abort(400, _('Error while requesting usage'))
@index.when(method='PUT', template='json')
def put(self, project_id, target_project_id, action=None):
"""Update quota limits for a project.
If it fails, Then creates a new entry in DB for that project.
:param project_id: It's UUID of the project.
:param target_project_id: It's UUID of the project.
Note: In v1.0 it can be defaults sometimes.
Only specified tenant quota is retrieved from database
using this param.
:param action: Optional. If provided, it can be 'detail'
detail - Gets detail quota usage for the specified tenant.
"""
context = restcomm.extract_context_from_environ()
valid_project_id = uuidutils.is_uuid_like(project_id)
if not context.is_admin:
pecan.abort(403, _('Admin required'))
if not valid_project_id:
pecan.abort(400, _('Invalid request URL'))
if project_id != context.project and not context.is_admin:
pecan.abort(400, _('Invalid request URL'))
if not uuidutils.is_uuid_like(target_project_id):
pecan.abort(400, _('Invalid request URL'))
if action and action != 'sync':
pecan.abort(404, 'Invalid action, only sync is allowed')
elif action == 'sync':
return self.sync(context, target_project_id)
quota = self.update_quota(context, request, target_project_id)
return quota
@index.when(method='delete', template='json')
def delete(self, project_id, target_project_id):
"""Delete quota for a specified tenant.
Resources for a specific tenant can also be deleted.
:param project_id: It's UUID of the project.
:param target_project_id: It's UUID of the project.
Note: In v1.0 it can be defaults sometimes.
Only specified tenant quota is retrieved from database
using this param.
#NOTE: Support to delete quota for a specific resource is through CURL
request in V1.0.
"""
context = restcomm.extract_context_from_environ()
valid_project_id = uuidutils.is_uuid_like(project_id)
if not valid_project_id:
pecan.abort(400, _('Invalid request URL'))
if project_id != context.project and not context.is_admin:
pecan.abort(400, _('Invalid request URL'))
if not uuidutils.is_uuid_like(target_project_id):
pecan.abort(400, _('Invalid request URL'))
if not context.is_admin:
pecan.abort(403, _('Admin required'))
if request.body:
payload = eval(request.body).get('quota_set')
if not payload:
pecan.abort(400, _('quota_set in body required'))
self.delete_quota_resources(context, target_project_id, payload)
return {'Deleted quota limits': payload}
else:
self.delete_quota(context, target_project_id)
return "Deleted all quota limits for the given project"
class QuotaManagerV1Controller(BaseController):
"""Quota Management API controller for API version 1.1.
It references all other resources belonging to the API v1.1.
"""
VERSION_ALIASES = {
'PIKE': '1.1',
}
# to do the version compatibility for future purpose
def _determine_version_cap(self, target):
version_cap = 1.1
return version_cap
@expose(generic=True, template='json')
def index(self):
# Route the request to specific methods with parameters
pass
@index.when(method='GET', template='json')
def get(self, project_id, action=None):
"""Get quota of a tenant.
:param project_id: It's UUID of the project.
Only specified quota details can be viewed using this param.
:param action: Optional. If provided, it can be 'defaults' or 'detail'
defaults - returns the quotas limits from the conf file.
detail - returns the current quota usages of the tenant
"""
context = restcomm.extract_context_from_environ()
quota = collections.defaultdict(dict)
result = collections.defaultdict(dict)
if not uuidutils.is_uuid_like(project_id):
pecan.abort(400)
enforce = enf.enforce('kingbird:get_all_quota', context)
try:
if enforce or (project_id == context.project)\
or (action == 'defaults'):
result = self.get_quota(context, project_id, action)
quota['quota_set'] = result
return quota
else:
pecan.abort(403, _('Admin required '))
except exceptions.InternalError:
pecan.abort(400, _('Error while requesting usage'))
@index.when(method='PUT', template='json')
def put(self, project_id, action=None):
"""Update quota of a tenant.
:param project_id: It's UUID of the project.
Only specified tenant quota is updated using this param.
:param action: Optional. If provided, it can be 'sync'
action - syncs quota for the specified tenant
based on the kingbird magic.
"""
context = restcomm.extract_context_from_environ()
quota = collections.defaultdict(dict)
quota[project_id] = collections.defaultdict(dict)
if not uuidutils.is_uuid_like(project_id):
pecan.abort(400)
enforce = enf.enforce('kingbird:update_quota', context)
if not enforce:
pecan.abort(403, _('Admin required'))
if action not in ('sync', None):
pecan.abort(404, 'Invalid action, only sync is allowed')
elif action == 'sync':
return self.sync(context, project_id)
quota = self.update_quota(context, request, project_id)
return quota
@index.when(method='delete', template='json')
def delete(self, project_id, **args):
"""Delete quota of a tenant.
:param project_id: It's UUID of the project.
Only specified tenant quota is deleted using this param.
"""
context = restcomm.extract_context_from_environ()
if not uuidutils.is_uuid_like(project_id):
pecan.abort(400)
enforce = enf.enforce('kingbird:delete_quota', context)
if not enforce:
pecan.abort(403, _('Admin required'))
if args:
payload = args.keys()
if not payload:
pecan.abort(400, _('quota_set in body required'))
self.delete_quota_resources(context, project_id, payload)
return {'Deleted quota limits': payload}
else:
# Delete all quota limits for the project
self.delete_quota(context, project_id)
return "Deleted all quota limits for the given project"

53
kingbird/api/controllers/v1/root.py
View File

@ -23,52 +23,37 @@ from kingbird.api.controllers.v1 import sync_manager
class Controller(object): class Controller(object):
def __init__(self):
self.sub_controllers = {
"os-quota-sets": quota_manager.QuotaManagerController,
"os-quota-class-sets": quota_class.QuotaClassSetController,
"os-sync": sync_manager.ResourceSyncController
}
for name, ctrl in self.sub_controllers.items():
setattr(self, name, ctrl)
def _get_resource_controller(self, tenant_id, remainder): def _get_resource_controller(self, tenant_id, remainder):
if not remainder: if not remainder:
pecan.abort(404) pecan.abort(404)
return return
minor_version = remainder[-1]
remainder = remainder[:-1]
sub_controllers = dict()
if minor_version == '0':
sub_controllers["os-quota-sets"] = quota_manager.\
QuotaManagerController
sub_controllers["os-quota-class-sets"] = quota_class.\
QuotaClassSetController
sub_controllers["os-sync"] = sync_manager.\
ResourceSyncController
elif minor_version == '1':
sub_controllers["os-quota-sets"] = quota_manager.\
QuotaManagerV1Controller
for name, ctrl in sub_controllers.items():
setattr(self, name, ctrl)
resource = remainder[0] resource = remainder[0]
if resource not in self.sub_controllers: if resource not in sub_controllers:
pecan.abort(404) pecan.abort(404)
return return
# Pass the tenant_id for verification # Pass the tenant_id for verification
remainder = (tenant_id,) + remainder[1:] remainder = (tenant_id,) + remainder[1:]
return self.sub_controllers[resource](), remainder return sub_controllers[resource](), remainder
@pecan.expose() @pecan.expose()
def _lookup(self, tenant_id, *remainder): def _lookup(self, tenant_id, *remainder):
return self._get_resource_controller(tenant_id, remainder) return self._get_resource_controller(tenant_id, remainder)
@pecan.expose(generic=True, template='json')
def index(self):
return {
"version": "1.0",
"links": [
{"rel": "self",
"href": pecan.request.application_url + "/v1.0"}
] + [
{"rel": name,
"href": pecan.request.application_url +
"/v1.0/{tenant_id}/" + name}
for name in sorted(self.sub_controllers)
]
}
@index.when(method='POST')
@index.when(method='PUT')
@index.when(method='DELETE')
@index.when(method='HEAD')
@index.when(method='PATCH')
def not_supported(self):
pecan.abort(405)

71
kingbird/api/enforcer.py Normal file
View File

@ -0,0 +1,71 @@
# Copyright 2017 Ericsson AB.
#
# 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.
"""Policy enforcer for Kingbird."""
from oslo_config import cfg
from oslo_policy import policy
from kingbird.common import exceptions as exc
_ENFORCER = None
def enforce(action, context, target=None, do_raise=True,
exc=exc.NotAuthorized):
"""Verify that the action is valid on the target in this context.
:param action: String, representing the action to be checked.
This should be colon separated for clarity.
i.e. ``sync:list``
:param context: Kingbird context.
:param target: Dictionary, representing the object of the action.
For object creation, this should be a dictionary
representing the location of the object.
e.g. ``{'project_id': context.project}``
:param do_raise: if True (the default), raises specified exception.
:param exc: Exception to be raised if not authorized. Default is
kingbird.common.exceptions.NotAuthorized.
:return: returns True if authorized and False if not authorized and
do_raise is False.
"""
if cfg.CONF.auth_strategy != 'keystone':
# Policy enforcement is supported now only with Keystone
# authentication.
return
target_obj = {
'project_id': context.project,
'user_id': context.user,
}
target_obj.update(target or {})
_ensure_enforcer_initialization()
try:
_ENFORCER.enforce(action, target_obj, context.to_dict(),
do_raise=do_raise, exc=exc)
return True
except Exception:
return False
def _ensure_enforcer_initialization():
global _ENFORCER
if not _ENFORCER:
_ENFORCER = policy.Enforcer(cfg.CONF)
_ENFORCER.load_rules()

4
kingbird/common/context.py
View File

@ -19,7 +19,7 @@ from oslo_utils import encodeutils
from kingbird.common import policy from kingbird.common import policy
from kingbird.db import api as db_api from kingbird.db import api as db_api
ALLOWED_WITHOUT_AUTH = ['/', '/v1.0'] ALLOWED_WITHOUT_AUTH = '/'
class RequestContext(base_context.RequestContext): class RequestContext(base_context.RequestContext):
@ -131,7 +131,7 @@ def get_service_context(**args):
class AuthHook(hooks.PecanHook): class AuthHook(hooks.PecanHook):
def before(self, state): def before(self, state):
if state.request.path in ALLOWED_WITHOUT_AUTH: if state.request.path == ALLOWED_WITHOUT_AUTH:
return return
req = state.request req = state.request
identity_status = req.headers.get('X-Identity-Status') identity_status = req.headers.get('X-Identity-Status')

48
kingbird/tests/unit/api/test_root_controller.py
View File

@ -103,46 +103,6 @@ class TestRootController(KBApiTest):
self._test_method_returns_405('head') self._test_method_returns_405('head')
class TestV1Controller(KBApiTest):
def test_get(self):
response = self.app.get('/v1.0')
self.assertEqual(response.status_int, 200)
json_body = jsonutils.loads(response.body)
version = json_body.get('version')
self.assertEqual('1.0', version)
links = json_body.get('links')
v1_link = links[0]
quota_class_link = links[1]
quota_manager_link = links[2]
sync_manager_link = links[3]
self.assertEqual('self', v1_link['rel'])
self.assertEqual('os-quota-sets', quota_manager_link['rel'])
self.assertEqual('os-quota-class-sets', quota_class_link['rel'])
self.assertEqual('os-sync', sync_manager_link['rel'])
def _test_method_returns_405(self, method):
api_method = getattr(self.app, method)
response = api_method('/v1.0', expect_errors=True)
self.assertEqual(response.status_int, 405)
def test_post(self):
self._test_method_returns_405('post')
def test_put(self):
self._test_method_returns_405('put')
def test_patch(self):
self._test_method_returns_405('patch')
def test_delete(self):
self._test_method_returns_405('delete')
def test_head(self):
self._test_method_returns_405('head')
class TestErrors(KBApiTest): class TestErrors(KBApiTest):
def setUp(self): def setUp(self):
@ -191,11 +151,3 @@ class TestKeystoneAuth(KBApiTest):
def test_auth_not_enforced_for_root(self): def test_auth_not_enforced_for_root(self):
response = self.app.get('/') response = self.app.get('/')
self.assertEqual(response.status_int, 200) self.assertEqual(response.status_int, 200)
def test_auth_not_enforced_for_v1(self):
response = self.app.get('/v1.0')
self.assertEqual(response.status_int, 200)
def test_auth_enforced(self):
response = self.app.get('/v1.0/', expect_errors=True)
self.assertEqual(response.status_int, 401)

2
kingbird/tests/unit/api/v1/controllers/test_quota_manager.py
View File

@ -323,7 +323,7 @@ class TestQuotaManager(testroot.KBApiTest):
@mock.patch.object(rpc_client, 'EngineClient') @mock.patch.object(rpc_client, 'EngineClient')
def test_put_invalid_curl_req_nonadmin(self, mock_rpc_client): def test_put_invalid_curl_req_nonadmin(self, mock_rpc_client):
FAKE_URL = '/v1.0/dummy/os-quota-sets/dummy2/sync' FAKE_URL = '/v1.0/dummy/os-quota-sets/dummy2/sync'
self.assertRaisesRegexp(webtest.app.AppError, "400 *", self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.put, FAKE_URL, self.app.put, FAKE_URL,
headers=NON_ADMIN_HEADERS) headers=NON_ADMIN_HEADERS)

325
kingbird/tests/unit/api/v1/controllers/test_quota_manager_v1.py Normal file
View File

@ -0,0 +1,325 @@
# Copyright (c) 2017 Ericsson AB.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
import webtest
from oslo_config import cfg
from kingbird.api.controllers.v1 import quota_manager
from kingbird.common import config
from kingbird.rpc import client as rpc_client
from kingbird.tests.unit.api import test_root_controller as testroot
from kingbird.tests import utils
config.register_options()
OPT_GROUP_NAME = 'keystone_authtoken'
cfg.CONF.import_group(OPT_GROUP_NAME, "keystonemiddleware.auth_token")
TARGET_FAKE_TENANT = utils.UUID1
FAKE_TENANT = utils.UUID2
HEADERS = {'X-Tenant-Id': TARGET_FAKE_TENANT,
'X-Identity-Status': 'Confirmed'}
class Result(object):
def __init__(self, project_id, resource, hard_limit):
self.project_id = project_id
self.resource = resource
self.hard_limit = hard_limit
class TestQuotaManager(testroot.KBApiTest):
def setUp(self):
super(TestQuotaManager, self).setUp()
cfg.CONF.set_override('admin_tenant', 'fake_tenant_id',
group='cache')
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
@mock.patch.object(quota_manager, 'enf')
def test_get_quota_details_admin(self, mock_enf, mock_db_api):
Res = Result(TARGET_FAKE_TENANT, 'ram', 100)
mock_enf.enforce.return_value = True
mock_db_api.quota_get_all_by_project.return_value = \
{"project_id": Res.project_id,
Res.resource: Res.hard_limit}
fake_url = '/v1.1/%s/os-quota-sets/'\
% (FAKE_TENANT)
response = self.app.get(
fake_url,
headers=HEADERS)
self.assertEqual(response.status_int, 200)
self.assertEqual({'quota_set':
{'project_id': TARGET_FAKE_TENANT, 'ram': 100}},
eval(response.text))
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
def test_get_default_limits(self, mock_db_api):
mock_db_api.quota_class_get_default.return_value = \
{'class_name': 'default'}
fake_url = '/v1.1/%s/os-quota-sets/defaults'\
% (TARGET_FAKE_TENANT)
response = self.app.get(
fake_url,
headers=HEADERS)
self.assertEqual(response.status_int, 200)
result = eval(response.text)
for resource in result['quota_set']:
self.assertEqual(
cfg.CONF.kingbird_global_limit['quota_' + resource],
result['quota_set'][resource])
@mock.patch.object(rpc_client, 'EngineClient')
@mock.patch.object(quota_manager, 'enf')
def test_get_another_tenant_quota_usages_admin(self, mock_enf,
mock_rpc_client):
expected_usage = {"ram": 10}
mock_enf.enforce.return_value = True
mock_rpc_client().get_total_usage_for_tenant.return_value = \
expected_usage
fake_url = '/v1.1/%s/os-quota-sets/detail'\
% (FAKE_TENANT)
response = self.app.get(
fake_url,
headers=HEADERS)
self.assertEqual(response.status_int, 200)
self.assertEqual(eval(response.body), {"quota_set": expected_usage})
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
@mock.patch.object(quota_manager, 'enf')
def test_update_quota_admin(self, mock_enf, mock_db_api):
Res = Result(FAKE_TENANT, 'cores', 10)
mock_enf.enforce.return_value = True
mock_db_api.quota_update.return_value = Res
data = {"quota_set": {Res.resource: Res.hard_limit}}
fake_url = '/v1.1/%s/os-quota-sets/'\
% (FAKE_TENANT)
response = self.app.put_json(
fake_url,
headers=HEADERS,
params=data)
self.assertEqual(response.status_int, 200)
self.assertEqual({Res.project_id: {Res.resource: Res.hard_limit}},
eval(response.text))
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
@mock.patch.object(quota_manager, 'enf')
def test_delete_quota_resources_admin(self, mock_enf, mock_db_api):
Res = Result(FAKE_TENANT, 'cores', 10)
mock_enf.enforce.return_value = True
mock_db_api.quota_destroy.return_value = Res
fake_url = '/v1.1/%s/os-quota-sets/?cores'\
% (FAKE_TENANT)
response = self.app.delete_json(
fake_url,
headers=HEADERS)
self.assertEqual(response.status_int, 200)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
@mock.patch.object(quota_manager, 'enf')
def test_delete_complete_quota_admin(self, mock_enf, mock_db_api):
Res = Result(FAKE_TENANT, 'cores', 10)
mock_enf.enforce.return_value = True
mock_db_api.quota_destroy_all.return_value = Res
fake_url = '/v1.1/%s/os-quota-sets'\
% (FAKE_TENANT)
response = self.app.delete_json(
fake_url,
headers=HEADERS)
self.assertEqual(response.status_int, 200)
self.assertEqual('Deleted all quota limits for the given project',
eval(response.text))
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'enf')
def test_quota_sync_admin(self, mock_enf):
fake_url = '/v1.1/%s/os-quota-sets/sync'\
% (FAKE_TENANT)
mock_enf.enforce.return_value = True
response = self.app.put_json(
fake_url,
headers=HEADERS)
self.assertEqual(response.status_int, 200)
self.assertEqual("triggered quota sync for " + FAKE_TENANT,
eval(response.text))
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'enf')
def test_update_nonadmin(self, mock_enf):
Res = Result(TARGET_FAKE_TENANT, 'cores', 10)
data = {"quota_set": {Res.resource: Res.hard_limit}}
fake_url = '/v1.1/%s/os-quota-sets/'\
% (FAKE_TENANT)
mock_enf.enforce.return_value = False
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.put_json, fake_url,
headers=HEADERS,
params=data)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'enf')
def test_delete_complete_quota_nonadmin(self, mock_enf):
fake_url = '/v1.1/%s/os-quota-sets/'\
% (FAKE_TENANT)
mock_enf.enforce.return_value = False
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.delete_json, fake_url,
headers=HEADERS)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'enf')
def test_delete_nonadmin(self, mock_enf):
Res = Result(FAKE_TENANT, 'cores', 10)
data = {"quota_set": {Res.resource: Res.hard_limit}}
fake_url = '/v1.1/%s/os-quota-sets/'\
% (FAKE_TENANT)
mock_enf.enforce.return_value = False
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.delete_json, fake_url,
headers=HEADERS,
params=data)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'enf')
def test_quota_sync_nonadmin(self, mock_enf):
fake_url = '/v1.1/%s/os-quota-sets/sync'\
% (FAKE_TENANT)
mock_enf.enforce.return_value = False
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.put_json, fake_url,
headers=HEADERS)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
def test_quota_sync_bad_request(self):
fake_url = '/v1.1/%s/os-quota-ssdfets/sync'\
% (FAKE_TENANT)
self.assertRaisesRegexp(webtest.app.AppError, "404 *",
self.app.post_json, fake_url,
headers=HEADERS)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
@mock.patch.object(quota_manager, 'enf')
def test_update_invalid_payload(self, mock_enf, mock_db_api):
Res = Result(FAKE_TENANT, 'cores', 10)
fake_url = '/v1.1/%s/os-quota-sets/'\
% (TARGET_FAKE_TENANT)
mock_db_api.quota_update.return_value = Res
mock_enf.enforce.return_value = True
data = {'quota': {Res.resource: Res.hard_limit}}
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.put_json, fake_url,
headers=HEADERS, params=data)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
@mock.patch.object(quota_manager, 'enf')
def test_update_invalid_input(self, mock_enf, mock_db_api):
Res = Result(FAKE_TENANT, 'cores', -10)
fake_url = '/v1.1/%s/os-quota-sets/'\
% (TARGET_FAKE_TENANT)
mock_db_api.quota_update.return_value = Res
mock_enf.enforce.return_value = True
data = {"quota_set": {Res.resource: Res.hard_limit}}
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.put_json, fake_url,
headers=HEADERS, params=data)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
def test_quota_sync_bad_action(self):
fake_url = '/v1.1/%s/os-quota-sets/syncing'\
% (TARGET_FAKE_TENANT)
self.assertRaisesRegexp(webtest.app.AppError, "404 *",
self.app.delete_json, fake_url,
headers=HEADERS)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'enf')
def test_get_another_tenant_quota_nonadmin(self, mock_enf):
fake_url = '/v1.1/%s/os-quota-sets'\
% (FAKE_TENANT)
mock_enf.enforce.return_value = False
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.get, fake_url,
headers=HEADERS)
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'db_api')
@mock.patch.object(quota_manager, 'enf')
def test_get_complete_quota_another_tenant_with_admin(self, mock_enf,
mock_db_api):
fake_url = '/v1.1/%s/os-quota-sets'\
% (FAKE_TENANT)
Res = Result(FAKE_TENANT, 'ram', 100)
mock_db_api.quota_get_all_by_project.return_value = \
{"project_id": Res.project_id,
Res.resource: Res.hard_limit}
mock_enf.enforce.return_value = True
response = self.app.get(
fake_url,
headers=HEADERS)
self.assertEqual(response.status_int, 200)
self.assertEqual({'quota_set': {'project_id': FAKE_TENANT,
'ram': 100}}, eval(response.text))
@mock.patch.object(rpc_client, 'EngineClient', new=mock.Mock())
@mock.patch.object(quota_manager, 'enf')
def test_get_usages_another_tenant_non_admin(self, mock_enf):
fake_url = '/v1.1/%s/os-quota-sets/detail'\
% (FAKE_TENANT)
mock_enf.enforce.return_value = False
self.assertRaisesRegexp(webtest.app.AppError, "403 *",
self.app.get, fake_url,
headers=HEADERS)
@mock.patch.object(rpc_client, 'EngineClient')
@mock.patch.object(quota_manager, 'enf')
def test_get_usages_another_tenant_admin(self, mock_enf, mock_rpc_client):
expected_usage = {"ram": 10}
fake_url = '/v1.1/%s/os-quota-sets/detail'\
% (FAKE_TENANT)
mock_enf.enforce.return_value = True
mock_rpc_client().get_total_usage_for_tenant.return_value = \
expected_usage
response = self.app.get(
fake_url,
headers=HEADERS)
self.assertEqual(response.status_int, 200)
self.assertEqual(eval(response.body), {"quota_set": expected_usage})
@mock.patch.object(rpc_client, 'EngineClient')
def test_get_with_invalid_curl_req(self, mock_rpc_client):
fake_url = '/v1.1/dummy/os-quota-sets/defaults'
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.get, fake_url,
headers=HEADERS)
@mock.patch.object(rpc_client, 'EngineClient')
def test_put_with_invalid_curl_req(self, mock_rpc_client):
fake_url = '/v1.1/dummy/os-quota-sets'
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.put, fake_url,
headers=HEADERS)
@mock.patch.object(rpc_client, 'EngineClient')
def test_delete_with_invalid_curl_req(self, mock_rpc_client):
fake_url = '/v1.1/dummy/os-quota-sets'
self.assertRaisesRegexp(webtest.app.AppError, "400 *",
self.app.delete, fake_url,
headers=HEADERS)