QUOTAS = quota.QUOTAS
NON_QUOTA_KEYS = ['tenant_id', 'id']
-
authorize_update = extensions.extension_authorizer('volume', 'quotas:update')
authorize_show = extensions.extension_authorizer('volume', 'quotas:show')
authorize_delete = extensions.extension_authorizer('volume', 'quotas:delete')
"resources.") % key
raise webob.exc.HTTPBadRequest(explanation=msg)
+ def _validate_quota_limit(self, quota, key, project_quotas=None,
+ parent_project_quotas=None):
+ limit = self.validate_integer(quota[key], key, min_value=-1,
+ max_value=db.MAX_INT)
+
+ if parent_project_quotas:
+ free_quota = (parent_project_quotas[key]['limit'] -
+ parent_project_quotas[key]['in_use'] -
+ parent_project_quotas[key]['reserved'] -
+ parent_project_quotas[key]['allocated'])
+
+ current = 0
+ if project_quotas.get(key):
+ current = project_quotas[key]['limit']
+
+ if limit - current > free_quota:
+ msg = _("Free quota available is %s.") % free_quota
+ raise webob.exc.HTTPBadRequest(explanation=msg)
+ return limit
+
def _get_quotas(self, context, id, usages=False, parent_project_id=None):
values = QUOTAS.get_project_quotas(context, id, usages=usages,
parent_project_id=parent_project_id)
else:
return {k: v['limit'] for k, v in values.items()}
+ def _authorize_update_or_delete(self, context_project,
+ target_project_id,
+ parent_id):
+ """Checks if update or delete are allowed in the current hierarchy.
+
+ With hierarchical projects, only the admin of the parent or the root
+ project has privilege to perform quota update and delete operations.
+
+ :param context_project: The project in which the user is scoped to.
+ :param target_project_id: The id of the project in which the
+ user want to perform an update or
+ delete operation.
+ :param parent_id: The parent id of the project in which the user
+ want to perform an update or delete operation.
+ """
+ if context_project.parent_id and parent_id != context_project.id:
+ msg = _("Update and delete quota operations can only be made "
+ "by an admin of immediate parent or by the CLOUD admin.")
+ raise webob.exc.HTTPForbidden(explanation=msg)
+
+ if context_project.id != target_project_id:
+ if not self._is_descendant(target_project_id,
+ context_project.subtree):
+ msg = _("Update and delete quota operations can only be made "
+ "to projects in the same hierarchy of the project in "
+ "which users are scoped to.")
+ raise webob.exc.HTTPForbidden(explanation=msg)
+ else:
+ msg = _("Update and delete quota operations can only be made "
+ "by an admin of immediate parent or by the CLOUD admin.")
+ raise webob.exc.HTTPForbidden(explanation=msg)
+
+ def _authorize_show(self, context_project, target_project):
+ """Checks if show is allowed in the current hierarchy.
+
+ With hierarchical projects, are allowed to perform quota show operation
+ users with admin role in, at least, one of the following projects: the
+ current project; the immediate parent project; or the root project.
+
+ :param context_project: The project in which the user
+ is scoped to.
+ :param target_project: The project in which the user wants
+ to perform a show operation.
+ """
+ if target_project.parent_id:
+ if target_project.id != context_project.id:
+ if not self._is_descendant(target_project.id,
+ context_project.subtree):
+ msg = _("Show operations can only be made to projects in "
+ "the same hierarchy of the project in which users "
+ "are scoped to.")
+ raise webob.exc.HTTPForbidden(explanation=msg)
+ if context_project.id != target_project.parent_id:
+ if context_project.parent_id:
+ msg = _("Only users with token scoped to immediate "
+ "parents or root projects are allowed to see "
+ "its children quotas.")
+ raise webob.exc.HTTPForbidden(explanation=msg)
+ elif context_project.parent_id:
+ msg = _("An user with a token scoped to a subproject is not "
+ "allowed to see the quota of its parents.")
+ raise webob.exc.HTTPForbidden(explanation=msg)
+
+ def _is_descendant(self, target_project_id, subtree):
+ if subtree is not None:
+ for key, value in subtree.items():
+ if key == target_project_id:
+ return True
+ if self._is_descendant(target_project_id, value):
+ return True
+ return False
+
def _get_project(self, context, id, subtree_as_ids=False):
"""A Helper method to get the project hierarchy.
@wsgi.serializers(xml=QuotaTemplate)
def show(self, req, id):
+ """Show quota for a particular tenant
+
+ This works for hierarchical and non-hierarchical projects. For
+ hierarchical projects admin of current project, immediate
+ parent of the project or the CLOUD admin are able to perform
+ a show.
+
+ :param req: request
+ :param id: target project id that needs to be updated
+ """
context = req.environ['cinder.context']
authorize_show(context)
-
params = req.params
+ target_project_id = id
+
if not hasattr(params, '__call__') and 'usage' in params:
usage = strutils.bool_from_string(params['usage'])
else:
usage = False
+ # With hierarchical projects, only the admin of the current project or
+ # the root project has privilege to perform quota show operations.
+ target_project = self._get_project(context, target_project_id)
+ context_project = self._get_project(context, context.project_id,
+ subtree_as_ids=True)
+
+ self._authorize_show(context_project, target_project)
try:
- sqlalchemy_api.authorize_project_context(context, id)
+ sqlalchemy_api.authorize_project_context(context,
+ target_project_id)
except exception.NotAuthorized:
raise webob.exc.HTTPForbidden()
- return self._format_quota_set(id, self._get_quotas(context, id, usage))
+ quotas = self._get_quotas(context, target_project_id, usage,
+ parent_project_id=target_project.parent_id)
+ return self._format_quota_set(target_project_id, quotas)
@wsgi.serializers(xml=QuotaTemplate)
def update(self, req, id, body):
+ """Update Quota for a particular tenant
+
+ This works for hierarchical and non-hierarchical projects. For
+ hierarchical projects only immediate parent admin or the
+ CLOUD admin are able to perform an update.
+
+ :param req: request
+ :param id: target project id that needs to be updated
+ :param body: key, value pair that that will be
+ applied to the resources if the update
+ succeeds
+ """
context = req.environ['cinder.context']
authorize_update(context)
self.validate_string_length(id, 'quota_set_name',
min_length=1, max_length=255)
- project_id = id
self.assert_valid_body(body, 'quota_set')
# Get the optional argument 'skip_validation' from body,
raise exception.InvalidParameterValue(err=msg)
skip_flag = strutils.bool_from_string(skip_flag)
+ target_project_id = id
bad_keys = []
# NOTE(ankit): Pass #1 - In this loop for body['quota_set'].items(),
msg = _("Bad key(s) in quota set: %s") % ",".join(bad_keys)
raise webob.exc.HTTPBadRequest(explanation=msg)
+ # Get the parent_id of the target project to verify whether we are
+ # dealing with hierarchical namespace or non-hierarchical namespace.
+ target_project = self._get_project(context, target_project_id)
+ parent_id = target_project.parent_id
+
+ if parent_id:
+ # Get the children of the project which the token is scoped to in
+ # order to know if the target_project is in its hierarchy.
+ context_project = self._get_project(context,
+ context.project_id,
+ subtree_as_ids=True)
+ self._authorize_update_or_delete(context_project,
+ target_project.id,
+ parent_id)
+ parent_project_quotas = QUOTAS.get_project_quotas(
+ context, parent_id, parent_project_id=parent_id)
+
# NOTE(ankit): Pass #2 - In this loop for body['quota_set'].keys(),
# we validate the quota limits to ensure that we can bail out if
# any of the items in the set is bad. Meanwhile we validate value
# to ensure that the value can't be lower than number of existing
# resources.
- quota_values = QUOTAS.get_project_quotas(context, project_id)
+ quota_values = QUOTAS.get_project_quotas(context, target_project_id,
+ defaults=False)
valid_quotas = {}
+ allocated_quotas = {}
for key in body['quota_set'].keys():
if key in NON_QUOTA_KEYS:
continue
- valid_quotas[key] = self.validate_integer(
- body['quota_set'][key], key, min_value=-1,
- max_value=db.MAX_INT)
-
if not skip_flag:
self._validate_existing_resource(key, value, quota_values)
+ if parent_id:
+ value = self._validate_quota_limit(body['quota_set'], key,
+ quota_values,
+ parent_project_quotas)
+ allocated_quotas[key] = (
+ parent_project_quotas[key]['allocated'] + value)
+ else:
+ value = self._validate_quota_limit(body['quota_set'], key)
+ valid_quotas[key] = value
+
# NOTE(ankit): Pass #3 - At this point we know that all the keys and
# values are valid and we can iterate and update them all in one shot
# without having to worry about rolling back etc as we have done
# the validation up front in the 2 loops above.
for key, value in valid_quotas.items():
try:
- db.quota_update(context, project_id, key, value)
+ db.quota_update(context, target_project_id, key, value)
except exception.ProjectQuotaNotFound:
- db.quota_create(context, project_id, key, value)
+ db.quota_create(context, target_project_id, key, value)
except exception.AdminRequired:
raise webob.exc.HTTPForbidden()
- return {'quota_set': self._get_quotas(context, id)}
+ # If hierarchical projects, update child's quota first
+ # and then parents quota. In future this needs to be an
+ # atomic operation.
+ if parent_id:
+ if key in allocated_quotas.keys():
+ db.quota_allocated_update(context, parent_id, key,
+ allocated_quotas[key])
+
+ return {'quota_set': self._get_quotas(context, target_project_id,
+ parent_project_id=parent_id)}
@wsgi.serializers(xml=QuotaTemplate)
def defaults(self, req, id):
@wsgi.serializers(xml=QuotaTemplate)
def delete(self, req, id):
+ """Delete Quota for a particular tenant.
+
+ This works for hierarchical and non-hierarchical projects. For
+ hierarchical projects only immediate parent admin or the
+ CLOUD admin are able to perform a delete.
+ :param req: request
+ :param id: target project id that needs to be updated
+ """
context = req.environ['cinder.context']
authorize_delete(context)
+ # Get the parent_id of the target project to verify whether we are
+ # dealing with hierarchical namespace or non-hierarchical namespace.
+ target_project = self._get_project(context, id)
+ parent_id = target_project.parent_id
+
try:
- db.quota_destroy_by_project(context, id)
- except exception.AdminRequired:
+ project_quotas = QUOTAS.get_project_quotas(
+ context, target_project.id, usages=True,
+ parent_project_id=parent_id)
+ except exception.NotAuthorized:
raise webob.exc.HTTPForbidden()
+ # If the project which is being deleted has allocated part of its quota
+ # to its subprojects, then subprojects' quotas should be deleted first.
+ for key, value in project_quotas.items():
+ if 'allocated' in project_quotas[key].keys():
+ if project_quotas[key]['allocated'] != 0:
+ msg = _("About to delete child projects having "
+ "non-zero quota. This should not be performed")
+ raise webob.exc.HTTPBadRequest(explanation=msg)
+
+ if parent_id:
+ # Get the children of the project which the token is scoped to in
+ # order to know if the target_project is in its hierarchy.
+ context_project = self._get_project(context,
+ context.project_id,
+ subtree_as_ids=True)
+ self._authorize_update_or_delete(context_project,
+ target_project.id,
+ parent_id)
+ parent_project_quotas = QUOTAS.get_project_quotas(
+ context, parent_id, parent_project_id=parent_id)
+
+ # Delete child quota first and later update parent's quota.
+ try:
+ db.quota_destroy_by_project(context, target_project.id)
+ except exception.AdminRequired:
+ raise webob.exc.HTTPForbidden()
+
+ # Update the allocated of the parent
+ for key, value in project_quotas.items():
+ project_hard_limit = project_quotas[key]['limit']
+ parent_allocated = parent_project_quotas[key]['allocated']
+ parent_allocated -= project_hard_limit
+ db.quota_allocated_update(context, parent_id, key,
+ parent_allocated)
+ else:
+ try:
+ db.quota_destroy_by_project(context, target_project.id)
+ except exception.AdminRequired:
+ raise webob.exc.HTTPForbidden()
+
class Quotas(extensions.ExtensionDescriptor):
"""Quota management support."""
return IMPL.quota_get_all_by_project(context, project_id)
+def quota_allocated_get_all_by_project(context, project_id):
+ """Retrieve all allocated quotas associated with a given project."""
+ return IMPL.quota_allocated_get_all_by_project(context, project_id)
+
+
+def quota_allocated_update(context, project_id,
+ resource, allocated):
+ """Update allocated quota to subprojects or raise if it does not exist.
+
+ :raises: cinder.exception.ProjectQuotaNotFound
+ """
+ return IMPL.quota_allocated_update(context, project_id,
+ resource, allocated)
+
+
def quota_update(context, project_id, resource, limit):
"""Update a quota or raise if it does not exist."""
return IMPL.quota_update(context, project_id, resource, limit)
return result
+@require_context
+def quota_allocated_get_all_by_project(context, project_id):
+ rows = model_query(context, models.Quota, read_deleted='no').filter_by(
+ project_id=project_id).all()
+ result = {'project_id': project_id}
+ for row in rows:
+ result[row.resource] = row.allocated
+ return result
+
+
@require_admin_context
def quota_create(context, project_id, resource, limit):
quota_ref = models.Quota()
return quota_ref
+@require_admin_context
+def quota_allocated_update(context, project_id, resource, allocated):
+ session = get_session()
+ with session.begin():
+ quota_ref = _quota_get(context, project_id, resource, session=session)
+ quota_ref.allocated = allocated
+ return quota_ref
+
+
@require_admin_context
def quota_destroy(context, project_id, resource):
session = get_session()
default value, if there is no value from the
quota class) will be reported if there is no
specific value for the resource.
- :param usages: If True, the current in_use and reserved counts
- will also be returned.
+ :param usages: If True, the current in_use, reserved and allocated
+ counts will also be returned.
:param parent_project_id: The id of the current project's parent,
if any.
"""
if usages:
project_usages = db.quota_usage_get_all_by_project(context,
project_id)
+ allocated_quotas = db.quota_allocated_get_all_by_project(
+ context, project_id)
+ allocated_quotas.pop('project_id')
# Get the quotas for the appropriate class. If the project ID
# matches the one in the context, we use the quota_class from
in_use=usage.get('in_use', 0),
reserved=usage.get('reserved', 0), )
+ if parent_project_id or allocated_quotas:
+ quotas[resource.name].update(
+ allocated=allocated_quotas.get(resource.name, 0), )
+
return quotas
def _get_quotas(self, context, resources, keys, has_sync, project_id=None,
default value, if there is no value from the
quota class) will be reported if there is no
specific value for the resource.
- :param usages: If True, the current in_use and reserved counts
- will also be returned.
+ :param usages: If True, the current in_use, reserved and
+ allocated counts will also be returned.
:param parent_project_id: The id of the current project's parent,
if any.
"""
self.req.environ = {'cinder.context': context.get_admin_context()}
self.req.environ['cinder.context'].is_admin = True
self.req.environ['cinder.context'].auth_token = uuid.uuid4().hex
+ self.req.environ['cinder.context'].project_id = 'foo'
self._create_project_hierarchy()
self.auth_url = CONF.keymgr.encryption_auth_url
self.controller._get_project = mock.Mock()
self.controller._get_project.side_effect = self._get_project
result = self.controller.defaults(self.req, 'foo')
- self.assertDictMatch(result, make_body())
+ self.assertDictMatch(make_body(), result)
def test_subproject_defaults(self):
self.controller._get_project = mock.Mock()
context.project_id = self.B.id
result = self.controller.defaults(self.req, self.B.id)
expected = make_subproject_body(tenant_id=self.B.id)
- self.assertDictMatch(result, expected)
+ self.assertDictMatch(expected, result)
def test_show(self):
self.controller._get_project = mock.Mock()
self.controller._get_project.side_effect = self._get_project
result = self.controller.show(self.req, 'foo')
- self.assertDictMatch(result, make_body())
+ self.assertDictMatch(make_body(), result)
+
+ def test_subproject_show(self):
+ self.controller._get_project = mock.Mock()
+ self.controller._get_project.side_effect = self._get_project
+ self.req.environ['cinder.context'].project_id = self.A.id
+ result = self.controller.show(self.req, self.B.id)
+ expected = make_subproject_body(tenant_id=self.B.id)
+ self.assertDictMatch(expected, result)
+
+ def test_subproject_show_in_hierarchy(self):
+ self.controller._get_project = mock.Mock()
+ self.controller._get_project.side_effect = self._get_project
+ # An user scoped to a root project in an hierarchy can see its children
+ # quotas.
+ self.req.environ['cinder.context'].project_id = self.A.id
+ result = self.controller.show(self.req, self.D.id)
+ expected = make_subproject_body(tenant_id=self.D.id)
+ self.assertDictMatch(result, expected)
+ # An user scoped to a parent project can see its immediate children
+ # quotas.
+ self.req.environ['cinder.context'].project_id = self.B.id
+ result = self.controller.show(self.req, self.D.id)
+ expected = make_subproject_body(tenant_id=self.D.id)
+ self.assertDictMatch(result, expected)
+
+ def test_subproject_show_target_project_equals_to_context_project(self):
+ self.controller._get_project = mock.Mock()
+ self.controller._get_project.side_effect = self._get_project
+ self.req.environ['cinder.context'].project_id = self.B.id
+ result = self.controller.show(self.req, self.B.id)
+ expected = make_subproject_body(tenant_id=self.B.id)
+ self.assertDictMatch(result, expected)
def test_show_not_authorized(self):
self.controller._get_project = mock.Mock()
self.assertRaises(webob.exc.HTTPForbidden, self.controller.show,
self.req, 'foo')
+ def test_subproject_show_not_authorized(self):
+ self.controller._get_project = mock.Mock()
+ self.controller._get_project.side_effect = self._get_project
+ self.req.environ['cinder.context'].project_id = self.B.id
+ self.assertRaises(webob.exc.HTTPForbidden, self.controller.show,
+ self.req, self.C.id)
+ self.req.environ['cinder.context'].project_id = self.B.id
+ self.assertRaises(webob.exc.HTTPForbidden, self.controller.show,
+ self.req, self.A.id)
+
def test_update(self):
self.controller._get_project = mock.Mock()
self.controller._get_project.side_effect = self._get_project
body = make_body(gigabytes=2000, snapshots=15,
volumes=5, backups=5, tenant_id=None)
result = self.controller.update(self.req, 'foo', body)
- self.assertDictMatch(result, body)
+ self.assertDictMatch(body, result)
body = make_body(gigabytes=db.MAX_INT, tenant_id=None)
result = self.controller.update(self.req, 'foo', body)
+ self.assertDictMatch(body, result)
+
+ def test_update_subproject(self):
+ self.controller._get_project = mock.Mock()
+ self.controller._get_project.side_effect = self._get_project
+ # Update the project A quota.
+ self.req.environ['cinder.context'].project_id = self.A.id
+ body = make_body(gigabytes=2000, snapshots=15,
+ volumes=5, backups=5, tenant_id=None)
+ result = self.controller.update(self.req, self.A.id, body)
+ self.assertDictMatch(result, body)
+ # Update the quota of B to be equal to its parent quota
+ self.req.environ['cinder.context'].project_id = self.A.id
+ body = make_body(gigabytes=2000, snapshots=15,
+ volumes=5, backups=5, tenant_id=None)
+ result = self.controller.update(self.req, self.B.id, body)
+ self.assertDictMatch(result, body)
+ # Try to update the quota of C, it will not be allowed, since the
+ # project A doesn't have free quota available.
+ self.req.environ['cinder.context'].project_id = self.A.id
+ body = make_body(gigabytes=2000, snapshots=15,
+ volumes=5, backups=5, tenant_id=None)
+ self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
+ self.req, self.C.id, body)
+ # Successfully update the quota of D.
+ self.req.environ['cinder.context'].project_id = self.A.id
+ body = make_body(gigabytes=1000, snapshots=7,
+ volumes=3, backups=3, tenant_id=None)
+ result = self.controller.update(self.req, self.D.id, body)
self.assertDictMatch(result, body)
+ # An admin of B can also update the quota of D, since D is its an
+ # immediate child.
+ self.req.environ['cinder.context'].project_id = self.B.id
+ body = make_body(gigabytes=1500, snapshots=10,
+ volumes=4, backups=4, tenant_id=None)
+ result = self.controller.update(self.req, self.D.id, body)
+
+ def test_update_subproject_not_in_hierarchy(self):
+ self.controller._get_project = mock.Mock()
+ self.controller._get_project.side_effect = self._get_project
+
+ # Create another project hierarchy
+ E = self.FakeProject(id=uuid.uuid4().hex, parent_id=None)
+ F = self.FakeProject(id=uuid.uuid4().hex, parent_id=E.id)
+ E.subtree = {F.id: F.subtree}
+ self.project_by_id[E.id] = E
+ self.project_by_id[F.id] = F
+
+ # Update the project A quota.
+ self.req.environ['cinder.context'].project_id = self.A.id
+ body = make_body(gigabytes=2000, snapshots=15,
+ volumes=5, backups=5, tenant_id=None)
+ result = self.controller.update(self.req, self.A.id, body)
+ self.assertDictMatch(result, body)
+ # Try to update the quota of F, it will not be allowed, since the
+ # project E doesn't belongs to the project hierarchy of A.
+ self.req.environ['cinder.context'].project_id = self.A.id
+ body = make_body(gigabytes=2000, snapshots=15,
+ volumes=5, backups=5, tenant_id=None)
+ self.assertRaises(webob.exc.HTTPForbidden, self.controller.update,
+ self.req, F.id, body)
+
+ def test_update_subproject_with_not_root_context_project(self):
+ self.controller._get_project = mock.Mock()
+ self.controller._get_project.side_effect = self._get_project
+ # Update the project A quota.
+ self.req.environ['cinder.context'].project_id = self.A.id
+ body = make_body(gigabytes=2000, snapshots=15,
+ volumes=5, backups=5, tenant_id=None)
+ result = self.controller.update(self.req, self.A.id, body)
+ self.assertDictMatch(result, body)
+ # Try to update the quota of B, it will not be allowed, since the
+ # project in the context (B) is not a root project.
+ self.req.environ['cinder.context'].project_id = self.B.id
+ body = make_body(gigabytes=2000, snapshots=15,
+ volumes=5, backups=5, tenant_id=None)
+ self.assertRaises(webob.exc.HTTPForbidden, self.controller.update,
+ self.req, self.B.id, body)
@mock.patch(
'cinder.api.openstack.wsgi.Controller.validate_string_length')
@mock.patch(
'cinder.api.openstack.wsgi.Controller.validate_integer')
def test_update_limit(self, mock_validate_integer, mock_validate):
+ self.controller._get_project = mock.Mock()
+ self.controller._get_project.side_effect = self._get_project
mock_validate_integer.return_value = 10
body = {'quota_set': {'volumes': 10}}
db.quota_usage_get_all_by_project(ctxt, 'foo'))
def test_update_lower_than_existing_resources_when_skip_false(self):
+ self.controller._get_project = mock.Mock()
+ self.controller._get_project.side_effect = self._get_project
self._commit_quota_reservation()
body = {'quota_set': {'volumes': 0},
'skip_validation': 'false'}
self.req, 'foo', body)
def test_update_lower_than_existing_resources_when_skip_true(self):
+ self.controller._get_project = mock.Mock()
+ self.controller._get_project.side_effect = self._get_project
self._commit_quota_reservation()
body = {'quota_set': {'volumes': 0},
'skip_validation': 'true'}
result['quota_set']['volumes'])
def test_update_lower_than_existing_resources_without_skip_argument(self):
+ self.controller._get_project = mock.Mock()
+ self.controller._get_project.side_effect = self._get_project
self._commit_quota_reservation()
body = {'quota_set': {'volumes': 0}}
result = self.controller.update(self.req, 'foo', body)
result_show_after = self.controller.show(self.req, 'foo')
self.assertDictMatch(result_show, result_show_after)
+ def test_subproject_delete(self):
+ self.controller._get_project = mock.Mock()
+ self.controller._get_project.side_effect = self._get_project
+ self.req.environ['cinder.context'].project_id = self.A.id
+
+ body = make_body(gigabytes=2000, snapshots=15,
+ volumes=5, backups=5,
+ backup_gigabytes=1000, tenant_id=None)
+ result_update = self.controller.update(self.req, self.A.id, body)
+ self.assertDictMatch(result_update, body)
+
+ # Set usage param to True in order to see get allocated values.
+ self.req.params = {'usage': 'True'}
+ result_show = self.controller.show(self.req, self.A.id)
+
+ result_update = self.controller.update(self.req, self.B.id, body)
+ self.assertDictMatch(result_update, body)
+
+ self.controller.delete(self.req, self.B.id)
+
+ result_show_after = self.controller.show(self.req, self.A.id)
+ self.assertDictMatch(result_show, result_show_after)
+
+ def test_delete_with_allocated_quota_different_from_zero(self):
+ self.controller._get_project = mock.Mock()
+ self.controller._get_project.side_effect = self._get_project
+ self.req.environ['cinder.context'].project_id = self.A.id
+
+ body = make_body(gigabytes=2000, snapshots=15,
+ volumes=5, backups=5,
+ backup_gigabytes=1000, tenant_id=None)
+ result_update = self.controller.update(self.req, self.A.id, body)
+ self.assertDictMatch(result_update, body)
+
+ # Set usage param to True in order to see get allocated values.
+ self.req.params = {'usage': 'True'}
+ result_show = self.controller.show(self.req, self.A.id)
+
+ result_update = self.controller.update(self.req, self.B.id, body)
+ self.assertDictMatch(result_update, body)
+
+ self.controller.delete(self.req, self.B.id)
+
+ result_show_after = self.controller.show(self.req, self.A.id)
+ self.assertDictMatch(result_show, result_show_after)
+
def test_delete_no_admin(self):
self.controller._get_project = mock.Mock()
self.controller._get_project.side_effect = self._get_project
self._stub_quota_class_get_all_by_name()
+ def _stub_allocated_get_all_by_project(self, allocated_quota=False):
+ def fake_qagabp(context, project_id):
+ self.calls.append('quota_allocated_get_all_by_project')
+ self.assertEqual('test_project', project_id)
+ if allocated_quota:
+ return dict(project_id=project_id, volumes=3)
+ return dict(project_id=project_id)
+
+ self.stubs.Set(db, 'quota_allocated_get_all_by_project', fake_qagabp)
+
def test_get_project_quotas(self):
self._stub_get_by_project()
self._stub_volume_type_get_all()
+ self._stub_allocated_get_all_by_project()
result = self.driver.get_project_quotas(
FakeContext('test_project', 'test_class'),
quota.QUOTAS.resources, 'test_project')
self.assertEqual(['quota_get_all_by_project',
'quota_usage_get_all_by_project',
+ 'quota_allocated_get_all_by_project',
'quota_class_get_all_by_name',
'quota_class_get_default', ], self.calls)
self.assertEqual(dict(volumes=dict(limit=10,
reserved= 0)
), result)
+ def test_get_root_project_with_subprojects_quotas(self):
+ self._stub_get_by_project()
+ self._stub_volume_type_get_all()
+ self._stub_allocated_get_all_by_project(allocated_quota=True)
+ result = self.driver.get_project_quotas(
+ FakeContext('test_project', None),
+ quota.QUOTAS.resources, 'test_project')
+
+ self.assertEqual(['quota_get_all_by_project',
+ 'quota_usage_get_all_by_project',
+ 'quota_allocated_get_all_by_project',
+ 'quota_class_get_default', ], self.calls)
+ self.assertEqual(dict(volumes=dict(limit=10,
+ in_use=2,
+ reserved=0,
+ allocated=3, ),
+ snapshots=dict(limit=10,
+ in_use=2,
+ reserved=0,
+ allocated=0, ),
+ gigabytes=dict(limit=50,
+ in_use=10,
+ reserved=0,
+ allocated=0, ),
+ backups=dict(limit=10,
+ in_use=2,
+ reserved=0,
+ allocated=0, ),
+ backup_gigabytes=dict(limit=50,
+ in_use=10,
+ reserved=0,
+ allocated=0, ),
+ per_volume_gigabytes=dict(in_use=0,
+ limit=-1,
+ reserved=0,
+ allocated=0)
+ ), result)
+
def test_get_subproject_quotas(self):
self._stub_get_by_subproject()
self._stub_volume_type_get_all()
+ self._stub_allocated_get_all_by_project(allocated_quota=True)
parent_project_id = 'test_parent_project_id'
result = self.driver.get_project_quotas(
FakeContext('test_project', None),
parent_project_id=parent_project_id)
self.assertEqual(['quota_get_all_by_project',
- 'quota_usage_get_all_by_project', ], self.calls)
+ 'quota_usage_get_all_by_project',
+ 'quota_allocated_get_all_by_project', ], self.calls)
self.assertEqual(dict(volumes=dict(limit=10,
in_use=2,
- reserved=0, ),
+ reserved=0,
+ allocated=3, ),
snapshots=dict(limit=0,
in_use=0,
- reserved=0, ),
+ reserved=0,
+ allocated=0, ),
gigabytes=dict(limit=50,
in_use=10,
- reserved=0, ),
+ reserved=0,
+ allocated=0, ),
backups=dict(limit=0,
in_use=0,
- reserved=0, ),
+ reserved=0,
+ allocated=0, ),
backup_gigabytes=dict(limit=0,
in_use=0,
- reserved=0, ),
+ reserved=0,
+ allocated=0, ),
per_volume_gigabytes=dict(in_use=0,
limit=0,
- reserved= 0)
+ reserved=0,
+ allocated=0)
), result)
def test_get_project_quotas_alt_context_no_class(self):
Code Review / openstack-build / cinder-build.git / commitdiff