openstack
/
neutron
Code Issues Proposed changes

Scope get_tenant_quotas by tenant_id 

Using model_query in the operation for retrieving tenant limits
will spare the need for explicit authorization check in the
quota controller. This is particularly relevant for the pecan
framework where every Neutron API call undergoes authZ checks
in the same pecan hook.

This patch will automatically adapt by eventuals changes
introducing "un-scoped" contexts.

Closes-bug: #1505406

Change-Id: I6952f5c85cd7fb0263789f768d23de3fe80b8183
This commit is contained in:
Salvatore Orlando 2015-10-12 15:47:03 -07:00
parent 6576b7061e
commit 24b482ac15
2 changed files with 19 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
neutron/db/quota/driver.py
View File

@ -18,6 +18,7 @@ from oslo_log import log
from neutron.common import exceptions
from neutron.db import api as db_api
from neutron.db import common_db_mixin as common_db
from neutron.db.quota import api as quota_api
from neutron.db.quota import models as quota_models
@ -34,7 +35,8 @@ class DbQuotaDriver(object):
@staticmethod
def get_tenant_quotas(context, resources, tenant_id):
"""Given a list of resources, retrieve the quotas for the given
tenant.
tenant. If no limits are found for the specified tenant, the operation
returns the default limits.
:param context: The request context, for access checks.
:param resources: A dictionary of the registered resource keys.
@ -47,7 +49,7 @@ class DbQuotaDriver(object):
for key, resource in resources.items())
# update with tenant specific limits
q_qry = context.session.query(quota_models.Quota).filter_by(
q_qry = common_db.model_query(context, quota_models.Quota).filter_by(
tenant_id=tenant_id)
for item in q_qry:
tenant_quota[item['resource']] = item['limit']

15
neutron/tests/unit/db/quota/test_driver.py
View File

@ -74,6 +74,21 @@ class TestDbQuotaDriver(testlib_api.SqlTestCase):
quotas = self.plugin.get_tenant_quotas(self.context, defaults, PROJECT)
self.assertEqual(4, quotas[RESOURCE])
def test_get_tenant_quotas(self):
user_ctx = context.Context(user_id=PROJECT, tenant_id=PROJECT)
self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
quotas = self.plugin.get_tenant_quotas(user_ctx, {}, PROJECT)
self.assertEqual(2, quotas[RESOURCE])
def test_get_tenant_quotas_different_tenant(self):
user_ctx = context.Context(user_id=PROJECT,
tenant_id='another_project')
self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
# It is appropriate to use assertFalse here as the expected return
# value is an empty dict (the defaults passed in the statement below
# after the request context)
self.assertFalse(self.plugin.get_tenant_quotas(user_ctx, {}, PROJECT))
def test_get_all_quotas(self):
project_1 = 'prj_test_1'
project_2 = 'prj_test_2'