openstack
/
designate
Code Issues Proposed changes

Improve quota API validations 

This patch improves the quotas API validations and test coverage. Invalid quota settings will now be caught at the API/Central layer and not down in the storage layer(DB).

Closes-Bug: #1934596
Change-Id: I474bdd988a6cc3a9bcce1b65c2f49216dd85addf
(cherry picked from commit 38178c079a)
This commit is contained in:
Michael Johnson 2022-05-17 19:37:55 +00:00
parent 5d348de3e8
commit 2e55c3e24b
5 changed files with 167 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
designate/common/constants.py
View File

@ -15,3 +15,14 @@
# RBAC related constants
RBAC_PROJECT_ID = 'project_id'
RBAC_TARGET_PROJECT_ID = 'target_project_id'
# Quotas
MIN_QUOTA = -1
MAX_QUOTA = 2147483647
QUOTA_API_EXPORT_SIZE = 'api_export_size'
QUOTA_RECORDSET_RECORDS = 'recordset_records'
QUOTA_ZONE_RECORDS = 'zone_records'
QUOTA_ZONE_RECORDSETS = 'zone_recordsets'
QUOTA_ZONES = 'zones'
VALID_QUOTAS = [QUOTA_API_EXPORT_SIZE, QUOTA_RECORDSET_RECORDS,
QUOTA_ZONE_RECORDS, QUOTA_ZONE_RECORDSETS, QUOTA_ZONES]

8
designate/objects/quota.py
View File

@ -12,6 +12,7 @@
# 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 designate.common import constants
from designate.objects import base
from designate.objects import fields
@ -21,8 +22,11 @@ class Quota(base.DictObjectMixin, base.PersistentObjectMixin,
base.DesignateObject):
fields = {
'tenant_id': fields.AnyField(nullable=True),
'resource': fields.AnyField(nullable=True),
'hard_limit': fields.AnyField(nullable=True)
'resource': fields.EnumField(nullable=True,
valid_values=constants.VALID_QUOTAS),
'hard_limit': fields.IntegerFields(nullable=True,
minimum=constants.MIN_QUOTA,
maximum=constants.MAX_QUOTA)
}
STRING_KEYS = [

5
designate/tests/__init__.py
View File

@ -28,6 +28,7 @@ from oslo_messaging import conffixture as messaging_fixture
from oslotest import base
from testtools import testcase
from designate.common import constants
import designate.conf
from designate import exceptions
from designate import objects
@ -81,10 +82,10 @@ class TestCase(base.BaseTestCase):
}]
quota_fixtures = [{
'resource': 'zones',
'resource': constants.QUOTA_ZONES,
'hard_limit': 5,
}, {
'resource': 'records',
'resource': constants.QUOTA_ZONE_RECORDS,
'hard_limit': 50,
}]

81
designate/tests/test_quota/test_storage.py
View File

@ -15,6 +15,7 @@
# under the License.
from oslo_log import log as logging
from designate.common import constants
from designate import quota
from designate import tests
@ -27,49 +28,75 @@ class StorageQuotaTest(tests.TestCase):
self.config(quota_driver='storage')
self.quota = quota.get_quota()
def test_set_quota_create(self):
def test_set_quota_create_min(self):
context = self.get_admin_context()
context.all_tenants = True
quota = self.quota.set_quota(context, 'tenant_id', 'zones', 1500)
for current_quota in constants.VALID_QUOTAS:
quota = self.quota.set_quota(context, 'tenant_id',
current_quota, constants.MIN_QUOTA)
self.assertEqual({'zones': 1500}, quota)
self.assertEqual({current_quota: constants.MIN_QUOTA}, quota)
# Drop into the storage layer directly to ensure the quota was created
# successfully
criterion = {
'tenant_id': 'tenant_id',
'resource': 'zones'
}
# Drop into the storage layer directly to ensure the quota was
# created successfully
criterion = {
'tenant_id': 'tenant_id',
'resource': current_quota
}
quota = self.quota.storage.find_quota(context, criterion)
quota = self.quota.storage.find_quota(context, criterion)
self.assertEqual('tenant_id', quota['tenant_id'])
self.assertEqual('zones', quota['resource'])
self.assertEqual(1500, quota['hard_limit'])
self.assertEqual('tenant_id', quota['tenant_id'])
self.assertEqual(current_quota, quota['resource'])
self.assertEqual(constants.MIN_QUOTA, quota['hard_limit'])
def test_set_quota_create_max(self):
context = self.get_admin_context()
context.all_tenants = True
for current_quota in constants.VALID_QUOTAS:
quota = self.quota.set_quota(context, 'tenant_id',
current_quota, constants.MAX_QUOTA)
self.assertEqual({current_quota: constants.MAX_QUOTA}, quota)
# Drop into the storage layer directly to ensure the quota was
# created successfully
criterion = {
'tenant_id': 'tenant_id',
'resource': current_quota
}
quota = self.quota.storage.find_quota(context, criterion)
self.assertEqual('tenant_id', quota['tenant_id'])
self.assertEqual(current_quota, quota['resource'])
self.assertEqual(constants.MAX_QUOTA, quota['hard_limit'])
def test_set_quota_update(self):
context = self.get_admin_context()
context.all_tenants = True
# First up, Create the quota
self.quota.set_quota(context, 'tenant_id', 'zones', 1500)
for current_quota in constants.VALID_QUOTAS:
# First up, Create the quota
self.quota.set_quota(context, 'tenant_id', current_quota, 1500)
# Next, update the quota
self.quota.set_quota(context, 'tenant_id', 'zones', 1234)
# Next, update the quota
self.quota.set_quota(context, 'tenant_id', current_quota, 1234)
# Drop into the storage layer directly to ensure the quota was updated
# successfully
criterion = {
'tenant_id': 'tenant_id',
'resource': 'zones'
}
# Drop into the storage layer directly to ensure the quota was
# updated successfully
criterion = {
'tenant_id': 'tenant_id',
'resource': current_quota
}
quota = self.quota.storage.find_quota(context, criterion)
quota = self.quota.storage.find_quota(context, criterion)
self.assertEqual('tenant_id', quota['tenant_id'])
self.assertEqual('zones', quota['resource'])
self.assertEqual(1234, quota['hard_limit'])
self.assertEqual('tenant_id', quota['tenant_id'])
self.assertEqual(current_quota, quota['resource'])
self.assertEqual(1234, quota['hard_limit'])
def test_reset_quotas(self):
context = self.get_admin_context()

93
designate/tests/unit/objects/test_quota.py Normal file
View File

@ -0,0 +1,93 @@
# 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 oslo_log import log as logging
import oslotest.base
from designate.common import constants
from designate import objects
LOG = logging.getLogger(__name__)
class QuotaTest(oslotest.base.BaseTestCase):
def test_quota_min(self):
for current_quota in constants.VALID_QUOTAS:
quota = objects.Quota(tenant_id='123', resource=current_quota,
hard_limit=constants.MIN_QUOTA)
self.assertEqual('123', quota.tenant_id)
self.assertEqual(current_quota, quota.resource)
self.assertEqual(constants.MIN_QUOTA, quota.hard_limit)
def test_quota_max(self):
for current_quota in constants.VALID_QUOTAS:
quota = objects.Quota(tenant_id='123', resource=current_quota,
hard_limit=constants.MAX_QUOTA)
self.assertEqual('123', quota.tenant_id)
self.assertEqual(current_quota, quota.resource)
self.assertEqual(constants.MAX_QUOTA, quota.hard_limit)
def test_quota_too_small(self):
for current_quota in constants.VALID_QUOTAS:
self.assertRaises(ValueError, objects.Quota, tenant_id='123',
resource=current_quota,
hard_limit=constants.MIN_QUOTA - 1)
def test_quota_too_large(self):
for current_quota in constants.VALID_QUOTAS:
self.assertRaises(ValueError, objects.Quota, tenant_id='123',
resource=current_quota,
hard_limit=constants.MAX_QUOTA + 1)
def test_quota_invalid(self):
for current_quota in constants.VALID_QUOTAS:
self.assertRaises(ValueError, objects.Quota, tenant_id='123',
resource=current_quota,
hard_limit='bogus')
def test_quota_list(self):
quotas = objects.QuotaList()
quotas.append(objects.Quota(
tenant_id='123', resource=constants.QUOTA_RECORDSET_RECORDS))
quotas.append(objects.Quota(tenant_id='123',
resource=constants.QUOTA_ZONE_RECORDS))
quotas.append(objects.Quota(tenant_id='123',
resource=constants.QUOTA_ZONE_RECORDSETS))
self.assertEqual(constants.QUOTA_RECORDSET_RECORDS, quotas[0].resource)
self.assertEqual(constants.QUOTA_ZONE_RECORDS, quotas[1].resource)
self.assertEqual(constants.QUOTA_ZONE_RECORDSETS, quotas[2].resource)
def test_quota_list_from_dict(self):
quotas = objects.QuotaList().from_dict({
constants.QUOTA_ZONES: 100,
constants.QUOTA_ZONE_RECORDSETS: 101,
constants.QUOTA_ZONE_RECORDS: 102,
constants.QUOTA_RECORDSET_RECORDS: 103,
constants.QUOTA_API_EXPORT_SIZE: 104,
})
self.assertEqual(constants.QUOTA_ZONES, quotas[0].resource)
self.assertEqual(100, quotas[0].hard_limit)
self.assertEqual(constants.QUOTA_API_EXPORT_SIZE, quotas[4].resource)
self.assertEqual(104, quotas[4].hard_limit)
def test_quota_list_to_dict(self):
quotas = objects.QuotaList().from_dict({
constants.QUOTA_ZONES: 100,
constants.QUOTA_ZONE_RECORDSETS: 101,
})
self.assertEqual(100, quotas.to_dict()[constants.QUOTA_ZONES])
self.assertEqual(101,
quotas.to_dict()[constants.QUOTA_ZONE_RECORDSETS])