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 38178c079a1a66229f06becbd7b60a749879c4a6)

5 files changed, 167 insertions, 31 deletions

diff --git a/designate/common/constants.py b/designate/common/constants.py
index 1275c44c..1479cf7f 100644
--- a/designate/common/constants.py
+++ b/designate/common/constants.py
@@ -31,3 +31,14 @@ UPDATE = 'UPDATE'
 # Floating IP constants
 FLOATING_IP_ACTIONS = [CREATE, DELETE, UPDATE, NONE]
 FLOATING_IP_STATUSES = [ACTIVE, ERROR, INACTIVE, PENDING]
+
+# 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]
diff --git a/designate/objects/quota.py b/designate/objects/quota.py
index 0ad30094..1d35f9d0 100644
--- a/designate/objects/quota.py
+++ b/designate/objects/quota.py
@@ -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 = [
diff --git a/designate/tests/__init__.py b/designate/tests/__init__.py
index 86ca1381..4a30744e 100644
--- a/designate/tests/__init__.py
+++ b/designate/tests/__init__.py
@@ -29,6 +29,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.context import DesignateContext
 from designate import exceptions
@@ -82,10 +83,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,
     }]
 
diff --git a/designate/tests/test_quota/test_storage.py b/designate/tests/test_quota/test_storage.py
index 08dc178a..92c4461f 100644
--- a/designate/tests/test_quota/test_storage.py
+++ b/designate/tests/test_quota/test_storage.py
@@ -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)
+
+            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
 
-        quota = self.quota.storage.find_quota(context, criterion)
+        for current_quota in constants.VALID_QUOTAS:
+            quota = self.quota.set_quota(context, 'tenant_id',
+                                         current_quota, constants.MAX_QUOTA)
 
-        self.assertEqual('tenant_id', quota['tenant_id'])
-        self.assertEqual('zones', quota['resource'])
-        self.assertEqual(1500, quota['hard_limit'])
+            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()
diff --git a/designate/tests/unit/objects/test_quota.py b/designate/tests/unit/objects/test_quota.py
new file mode 100644
index 00000000..17a3de98
--- /dev/null
+++ b/designate/tests/unit/objects/test_quota.py
@@ -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])