Merge "Ensure RecordSet quotas are enforced"
This commit is contained in:
commit
65694bc2f0
@ -595,18 +595,27 @@ class Service(service.RPCService, service.Service):
|
|||||||
self.quota.limit_check(context, tenant_id, domains=count)
|
self.quota.limit_check(context, tenant_id, domains=count)
|
||||||
|
|
||||||
def _enforce_recordset_quota(self, context, domain):
|
def _enforce_recordset_quota(self, context, domain):
|
||||||
# TODO(kiall): Enforce RRSet Quotas
|
# Ensure the recordsets per domain quota is OK
|
||||||
pass
|
criterion = {'domain_id': domain.id}
|
||||||
|
count = self.storage.count_recordsets(context, criterion)
|
||||||
|
|
||||||
|
self.quota.limit_check(
|
||||||
|
context, domain.tenant_id, domain_recordsets=count)
|
||||||
|
|
||||||
def _enforce_record_quota(self, context, domain, recordset):
|
def _enforce_record_quota(self, context, domain, recordset):
|
||||||
# Ensure the records per domain quota is OK
|
# Ensure the records per domain quota is OK
|
||||||
criterion = {'domain_id': domain['id']}
|
criterion = {'domain_id': domain.id}
|
||||||
count = self.storage.count_records(context, criterion)
|
count = self.storage.count_records(context, criterion)
|
||||||
|
|
||||||
self.quota.limit_check(context, domain['tenant_id'],
|
self.quota.limit_check(context, domain.tenant_id,
|
||||||
domain_records=count)
|
domain_records=count)
|
||||||
|
|
||||||
# TODO(kiall): Enforce Records per RRSet Quotas
|
# 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)
|
||||||
|
|
||||||
# Misc Methods
|
# Misc Methods
|
||||||
def get_absolute_limits(self, context):
|
def get_absolute_limits(self, context):
|
||||||
|
@ -1063,15 +1063,16 @@ class CentralServiceTest(CentralTestCase):
|
|||||||
self.assertIsNotNone(recordset.records[1].id)
|
self.assertIsNotNone(recordset.records[1].id)
|
||||||
self.assertThat(new_serial, GreaterThan(original_serial))
|
self.assertThat(new_serial, GreaterThan(original_serial))
|
||||||
|
|
||||||
# def test_create_recordset_over_quota(self):
|
def test_create_recordset_over_quota(self):
|
||||||
# self.config(quota_domain_recordsets=1)
|
# SOA, NS recordsets exist by default.
|
||||||
|
self.config(quota_domain_recordsets=3)
|
||||||
|
|
||||||
# domain = self.create_domain()
|
domain = self.create_domain()
|
||||||
|
|
||||||
# self.create_recordset(domain)
|
self.create_recordset(domain)
|
||||||
|
|
||||||
# with testtools.ExpectedException(exceptions.OverQuota):
|
with testtools.ExpectedException(exceptions.OverQuota):
|
||||||
# self.create_recordset(domain)
|
self.create_recordset(domain)
|
||||||
|
|
||||||
def test_create_invalid_recordset_location_cname_at_apex(self):
|
def test_create_invalid_recordset_location_cname_at_apex(self):
|
||||||
domain = self.create_domain()
|
domain = self.create_domain()
|
||||||
@ -1595,7 +1596,8 @@ class CentralServiceTest(CentralTestCase):
|
|||||||
self.assertEqual(record['data'], values['data'])
|
self.assertEqual(record['data'], values['data'])
|
||||||
self.assertIn('status', record)
|
self.assertIn('status', record)
|
||||||
|
|
||||||
def test_create_record_over_quota(self):
|
def test_create_record_over_domain_quota(self):
|
||||||
|
# SOA and NS Records exist
|
||||||
self.config(quota_domain_records=3)
|
self.config(quota_domain_records=3)
|
||||||
|
|
||||||
# Creating the domain automatically creates SOA & NS records
|
# Creating the domain automatically creates SOA & NS records
|
||||||
@ -1607,6 +1609,18 @@ class CentralServiceTest(CentralTestCase):
|
|||||||
with testtools.ExpectedException(exceptions.OverQuota):
|
with testtools.ExpectedException(exceptions.OverQuota):
|
||||||
self.create_record(domain, recordset)
|
self.create_record(domain, recordset)
|
||||||
|
|
||||||
|
def test_create_record_over_recordset_quota(self):
|
||||||
|
self.config(quota_recordset_records=1)
|
||||||
|
|
||||||
|
# Creating the domain automatically creates SOA & NS records
|
||||||
|
domain = self.create_domain()
|
||||||
|
recordset = self.create_recordset(domain)
|
||||||
|
|
||||||
|
self.create_record(domain, recordset)
|
||||||
|
|
||||||
|
with testtools.ExpectedException(exceptions.OverQuota):
|
||||||
|
self.create_record(domain, recordset)
|
||||||
|
|
||||||
def test_create_record_without_incrementing_serial(self):
|
def test_create_record_without_incrementing_serial(self):
|
||||||
domain = self.create_domain()
|
domain = self.create_domain()
|
||||||
recordset = self.create_recordset(domain, type='A')
|
recordset = self.create_recordset(domain, type='A')
|
||||||
|
Loading…
Reference in New Issue
Block a user