Merge "Ensure the zone records quota is enforced" into stable/liberty

designate/central/service.py

@@ -658,19 +658,37 @@ def _enforce_recordset_quota(self, context, domain):
             context, domain.tenant_id, domain_recordsets=count)
 
     def _enforce_record_quota(self, context, domain, recordset):
+        # Quotas don't apply to managed records.
+        if recordset.managed:
+            return
+
         # Ensure the records per domain quota is OK
-        criterion = {'domain_id': domain.id}
-        count = self.storage.count_records(context, criterion)
+        domain_criterion = {
+            'domain_id': domain.id,
+            'managed': False,  # only include non-managed records
+        }
+
+        domain_records = self.storage.count_records(context, domain_criterion)
+
+        recordset_criterion = {
+            'recordset_id': recordset.id,
+            'managed': False,  # only include non-managed records
+        }
+        recordset_records = self.storage.count_records(
+            context, recordset_criterion)
+
+        # We need to check the current number of domains + the
+        # changes that add, so lets get +/- from our recordset
+        # records based on the action
+        adjusted_domain_records = (
+            domain_records - recordset_records + len(recordset.records))
 
         self.quota.limit_check(context, domain.tenant_id,
-                               domain_records=count)
+                               domain_records=adjusted_domain_records)
 
         # Ensure the records per recordset quota is OK
-        criterion = {'recordset_id': recordset.id}
-        count = self.storage.count_records(context, criterion)
-
         self.quota.limit_check(context, domain.tenant_id,
-                               recordset_records=count)
+                               recordset_records=recordset_records)
 
     # Misc Methods
     def get_absolute_limits(self, context):
@@ -1250,6 +1268,11 @@ def _create_recordset_in_storage(self, context, domain, recordset,
         self._is_valid_recordset_records(recordset)
 
         if recordset.obj_attr_is_set('records') and len(recordset.records) > 0:
+
+            # Ensure the tenant has enough zone record quotas to
+            # create new records
+            self._enforce_record_quota(context, domain, recordset)
+
             if increment_serial:
                 # update the zone's status and increment the serial
                 domain = self._update_domain_in_storage(
@@ -1391,6 +1414,10 @@ def _update_recordset_in_storage(self, context, domain, recordset,
                     record.status = 'PENDING'
                     record.serial = domain.serial
 
+            # Ensure the tenant has enough zone record quotas to
+            # create new records
+            self._enforce_record_quota(context, domain, recordset)
+
         # Update the recordset
         recordset = self.storage.update_recordset(context, recordset)
 

designate/tests/test_central/test_service.py

@@ -1944,9 +1944,9 @@ def test_create_record(self):
         self.assertEqual(record['data'], values['data'])
         self.assertIn('status', record)
 
-    def test_create_record_over_domain_quota(self):
+    def test_create_record_and_update_over_zone_quota(self):
         # SOA and NS Records exist
-        self.config(quota_domain_records=3)
+        self.config(quota_domain_records=1)
 
         # Creating the domain automatically creates SOA & NS records
         domain = self.create_domain()
@@ -1957,6 +1957,26 @@ def test_create_record_over_domain_quota(self):
         with testtools.ExpectedException(exceptions.OverQuota):
             self.create_record(domain, recordset)
 
+    def test_create_record_over_zone_quota(self):
+        self.config(quota_domain_records=1)
+
+        # Creating the domain automatically creates SOA & NS records
+        domain = self.create_domain()
+
+        recordset = objects.RecordSet(
+            name='www.%s' % domain.name,
+            type='A',
+            records=objects.RecordList(objects=[
+                objects.Record(data='192.3.3.15'),
+                objects.Record(data='192.3.3.16'),
+            ])
+        )
+
+        with testtools.ExpectedException(exceptions.OverQuota):
+            # Persist the Object
+            recordset = self.central_service.create_recordset(
+                self.admin_context, domain.id, recordset=recordset)
+
     def test_create_record_over_recordset_quota(self):
         self.config(quota_recordset_records=1)
 

designate/tests/unit/test_central/test_basic.py

@@ -384,6 +384,30 @@ def test_create_recordset_in_storage(self):
         self.assertEqual(rs, 'rs')
         self.assertFalse(self.service._update_domain_in_storage.called)
 
+    def test_create_recordset_with_records_in_storage(self):
+        self.service._enforce_recordset_quota = mock.Mock()
+        self.service._enforce_record_quota = mock.Mock()
+        self.service._is_valid_recordset_name = mock.Mock()
+        self.service._is_valid_recordset_placement = mock.Mock()
+        self.service._is_valid_recordset_placement_subzone = mock.Mock()
+        self.service._is_valid_ttl = mock.Mock()
+
+        self.service.storage.create_recordset = mock.Mock(return_value='rs')
+
+        self.service.storage.find_domains = mock.Mock(return_value=[])
+        self.service._update_domain_in_storage = mock.Mock()
+
+        recordset = Mock()
+        recordset.obj_attr_is_set.return_value = True
+        recordset.records = [MockRecord()]
+
+        rs, zone = self.service._create_recordset_in_storage(
+            self.context, MockDomain(), recordset
+        )
+
+        assert self.service._enforce_record_quota.called
+        assert self.service._update_domain_in_storage.called
+
     def test__create_soa(self):
         self.service._create_recordset_in_storage = Mock(
             return_value=(None, None)
@@ -1179,6 +1203,7 @@ def test__update_recordset_in_storage_2(self):
         self.service._is_valid_recordset_placement_subdomain = Mock()
         self.service._is_valid_ttl = Mock()
         self.service._update_domain_in_storage = Mock()
+        self.service._enforce_record_quota = mock.Mock()
 
         self.service._update_recordset_in_storage(
             self.context,
@@ -1203,6 +1228,7 @@ def test__update_recordset_in_storage_2(self):
         assert not self.service._is_valid_ttl.called
         assert not self.service._update_domain_in_storage.called
         assert self.service.storage.update_recordset.called
+        assert self.service._enforce_record_quota.called
 
     def test_delete_recordset_not_found(self):
         self.service.storage.get_domain.return_value = RoObject(
@@ -1927,3 +1953,40 @@ def test__update_domain_or_record_status_no_domain(self):
 
         self.assertEqual(dom.action, 'CREATE')
         self.assertEqual(dom.status, 'ERROR')
+
+
+class CentralQuotaTest(unittest.TestCase):
+
+    def setUp(self):
+        self.context = mock.Mock()
+        self.domain = mock.Mock()
+
+    @patch('designate.central.service.storage')
+    @patch('designate.central.service.quota')
+    def test_domain_record_quota_allows_lowering_value(self, quota, storage):
+        service = Service()
+        service.storage.count_records.return_value = 10
+
+        recordset = mock.Mock()
+        recordset.managed = False
+        recordset.records = ['1.1.1.%i' % (i + 1) for i in range(5)]
+
+        service._enforce_record_quota(
+            self.context, self.domain, recordset
+        )
+
+        # Ensure we check against the number of records that will
+        # result in the API call. The 5 value is as if there were 10
+        # unmanaged records unders a single recordset. We find 10
+        # total - 10 for the recordset being passed in and add the 5
+        # from the new recordset.
+        check_domain_records = mock.call(
+            self.context, self.domain.tenant_id, domain_records=10 - 10 + 5
+        )
+        assert check_domain_records in service.quota.limit_check.mock_calls
+
+        # Check the recordset limit as well
+        check_recordset_records = mock.call(
+            self.context, self.domain.tenant_id, recordset_records=10
+        )
+        assert check_recordset_records in service.quota.limit_check.mock_calls