openstack/designate
Code Issues Proposed changes

Improve central unit test coverage

Browse Source 
- Minor changes to fix code that wasn't unit testable.

Change-Id: I1b00300118340faecc81c33f3a77e46e151e0a1c
This commit is contained in:
Erik Olof Gunnar Andersson
2023-05-26 19:54:24 -07:00
parent 661f0dd517
commit 9ba81579ac
5 changed files with 545 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
designate/central/service.py
+4 -9
View File
@@ -899,7 +899,6 @@ class Service(service.RPCService):
@rpc.expected_exceptions()
def get_zone_ns_records(self, context, zone_id=None, criterion=None):
if zone_id is None:
policy.check('get_zone_ns_records', context)
pool_id = cfg.CONF['service:central'].default_pool_id
@@ -1386,14 +1385,8 @@ class Service(service.RPCService):
return recordset
def _validate_recordset(self, context, zone, recordset):
# See if we're validating an existing or new recordset
recordset_id = None
if hasattr(recordset, 'id'):
recordset_id = recordset.id
# Ensure TTL is above the minimum
if not recordset_id:
if not recordset.id:
ttl = getattr(recordset, 'ttl', None)
else:
changes = recordset.obj_get_changes()
@@ -1405,7 +1398,7 @@ class Service(service.RPCService):
self._is_valid_recordset_name(context, zone, recordset.name)
self._is_valid_recordset_placement(
context, zone, recordset.name, recordset.type, recordset_id)
context, zone, recordset.name, recordset.type, recordset.id)
self._is_valid_recordset_placement_subzone(
context, zone, recordset.name)
@@ -1989,6 +1982,7 @@ class Service(service.RPCService):
context, fip, recordset.records[0], zone=zone, recordset=recordset
)
@rpc.expected_exceptions()
def _create_ptr_zone(self, elevated_context, zone_name):
zone_values = {
'type': 'PRIMARY',
@@ -2039,6 +2033,7 @@ class Service(service.RPCService):
record['id']
)
@rpc.expected_exceptions()
def _create_floating_ip(self, context, fip, record,
zone=None, recordset=None):
"""
designate/tests/test_central/test_service.py
+173 -2
View File
@@ -35,6 +35,7 @@ from oslo_utils import timeutils
from oslo_versionedobjects import exception as ovo_exc
import testtools
from designate.common import constants
from designate import exceptions
from designate import objects
from designate.storage import sql
@@ -886,6 +887,12 @@ class CentralServiceTest(designate.tests.TestCase):
self.assertGreater(len(servers), 0)
def test_get_zone_ns_records_no_zone_provided(self):
result = self.central_service.get_zone_ns_records(
self.admin_context, None
)
self.assertEqual(1, len(result))
@mock.patch.object(notifier.Notifier, "info")
def test_update_zone(self, mock_notifier):
# Create a zone
@@ -917,6 +924,14 @@ class CentralServiceTest(designate.tests.TestCase):
self.assertIsInstance(notified_zone, objects.Zone)
self.assertEqual(zone.id, notified_zone.id)
def test_update_zone_do_not_allow_tenant_id_update(self):
zone = self.create_zone(email='info@example.org')
zone.tenant_id = '1'
self.assertRaises(
rpc_dispatcher.ExpectedException,
self.central_service.update_zone, self.admin_context, zone
)
def test_update_zone_without_id(self):
# Create a zone
zone = self.create_zone(email='info@example.org')
@@ -924,9 +939,13 @@ class CentralServiceTest(designate.tests.TestCase):
# Update the object
zone.email = 'info@example.net'
zone.id = None
# Perform the update
with testtools.ExpectedException(Exception):
self.central_service.update_zone(self.admin_context, zone)
self.assertRaisesRegex(
Exception,
'Failed to determine zone id for synchronized operation',
self.central_service.update_zone, self.admin_context, zone
)
def test_update_zone_without_incrementing_serial(self):
# Create a zone
@@ -3783,6 +3802,158 @@ class CentralServiceTest(designate.tests.TestCase):
self.assertEqual(exceptions.ZoneImportNotFound, exc.exc_info[0])
def test_update_zone_export(self):
self.central_service.tg = mock.Mock()
new_zone_export = self.central_service.create_zone_import(
self.admin_context, objects.ZoneExport()
)
self.assertIsNone(new_zone_export.message)
new_zone_export.message = 'foo'
result = self.central_service.update_zone_export(
self.admin_context, new_zone_export
)
self.assertEqual('foo', result.message)
def test_create_ptr_zone(self):
cfg.CONF.set_override(
'managed_resource_tenant_id',
self.admin_context.project_id,
'service:central'
)
self.central_service._create_ptr_zone(
self.admin_context, 'example.org.'
)
zones = self.central_service.find_zones(self.admin_context)
self.assertEqual(1, len(zones))
self.assertEqual('example.org.', zones[0]['name'])
def test_create_duplicate_ptr_zone(self):
cfg.CONF.set_override(
'managed_resource_tenant_id',
self.admin_context.project_id,
'service:central'
)
self.central_service._create_ptr_zone(
self.admin_context, 'example.org.'
)
self.central_service._create_ptr_zone(
self.admin_context, 'example.org.'
)
zones = self.central_service.find_zones(self.admin_context)
self.assertEqual(1, len(zones))
self.assertEqual('example.org.', zones[0]['name'])
def test_create_floating_ip_with_record(self):
zone = self.create_zone()
record = objects.PTR(data='srv1.example.com.')
recordset = objects.RecordSet()
record.action = constants.CREATE
recordset.records = [record]
recordset.ttl = 3600
fip = {
'address': '127.0.0.1',
'id': '1',
'region': 'region',
}
result = self.central_service._create_floating_ip(
self.admin_context, fip, record, zone, recordset
)
self.assertEqual('srv1.example.com.', result.ptrdname)
self.assertEqual('127.0.0.1', result.address)
def test_create_floating_ip_with_no_record(self):
fip = {
'address': '127.0.0.1',
'id': '1',
'region': 'region',
}
self.central_service._create_floating_ip(self.admin_context, fip, None)
def test_create_floating_ip_zone_not_found(self):
fip = {
'address': '127.0.0.1',
'id': '1',
'region': 'region',
}
record = objects.PTR(data='srv1.example.com.')
recordset = objects.RecordSet()
recordset.records = [record]
self.central_service._create_floating_ip(
self.admin_context, fip, record=record, recordset=recordset
)
def test_delete_or_update_managed_recordset(self):
zone = self.create_zone()
recordset = self.create_recordset(zone)
record = recordset.records[0]
recordsets = self.central_service.find_recordsets(
self.admin_context,
criterion={'zone_id': zone.id, 'type': 'A'}
)
self.assertEqual(1, len(recordsets))
self.central_service._delete_or_update_managed_recordset(
self.admin_context, zone.id, recordset.id, record.id
)
recordsets = self.central_service.find_recordsets(
self.admin_context,
criterion={'zone_id': zone.id, 'type': 'A'}
)
self.assertEqual(0, len(recordsets), recordsets)
def test_delete_or_update_managed_recordset_with_multiple_records(self):
zone = self.create_zone()
recordset = self.create_recordset(
zone,
records=[objects.A(data='192.0.2.1'), objects.A(data='192.0.2.2')]
)
record = recordset.records[1]
records = self.central_service.find_records(
self.admin_context,
criterion={'zone_id': zone.id, 'recordset_id': recordset.id}
)
self.assertEqual(2, len(records))
self.central_service._delete_or_update_managed_recordset(
self.admin_context, zone.id, recordset.id, record.id
)
records = self.central_service.find_records(
self.admin_context,
criterion={'zone_id': zone.id, 'recordset_id': recordset.id}
)
self.assertEqual(1, len(records), records)
def test_delete_or_update_managed_recordset_record_not_found(self):
zone = self.create_zone()
recordset = self.create_recordset(zone)
self.central_service._delete_or_update_managed_recordset(
self.admin_context, zone.id, recordset.id, '1'
)
def test_delete_or_update_managed_recordset_not_found(self):
zone = self.create_zone()
record = objects.A()
recordset = objects.RecordSet()
recordset.records = [record]
self.central_service._delete_or_update_managed_recordset(
self.admin_context, zone.id, recordset.id, record.id
)
def test_share_zone(self):
# Create a Shared Zone
context = self.get_context(project_id='1', roles=['member', 'reader'])
designate/tests/unit/central/__init__.py
View File
designate/tests/unit/central/test_central.py
+329
View File
@@ -0,0 +1,329 @@
# 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 unittest import mock
from oslo_config import cfg
import oslotest.base
from designate.central import service
from designate.common import profiler
from designate import exceptions
from designate.objects import record
from designate.objects import zone
from designate import policy
from designate import rpc
CONF = cfg.CONF
class CentralTestCase(oslotest.base.BaseTestCase):
@mock.patch.object(policy, 'init')
@mock.patch.object(rpc, 'init')
@mock.patch.object(profiler, 'setup_profiler')
def setUp(self, mock_setup_profiler, mock_rpc_init, mock_policy_init):
super(CentralTestCase, self).setUp()
self.storage = mock.Mock()
self.service = service.Service()
self.service.coordination = mock.Mock()
self.service._storage = self.storage
self.context = mock.Mock()
mock_setup_profiler.assert_called()
mock_rpc_init.assert_called()
mock_policy_init.assert_called()
@mock.patch.object(rpc, 'get_server')
@mock.patch.object(rpc, 'get_notifier')
def test_service_start(self, mock_get_notifier, mock_get_server):
self.service.start()
mock_get_server.assert_called()
mock_get_notifier.assert_called_with('central')
self.service.coordination.start.assert_called()
@mock.patch.object(rpc, 'get_server')
@mock.patch.object(rpc, 'get_notifier')
def test_service_start_with_managed_tenant_id_set(self, mock_get_notifier,
mock_get_server):
CONF.set_override(
'managed_resource_tenant_id',
'24d1c4be-eb2d-44bb-bc49-fc5eced95e3d',
'service:central'
)
self.service.start()
mock_get_server.assert_called()
mock_get_notifier.assert_called_with('central')
self.service.coordination.start.assert_called()
def test_is_valid_project_id(self):
self.assertIsNone(self.service._is_valid_project_id('1'))
def test_is_valid_project_id_missing_project_id(self):
self.assertRaisesRegex(
exceptions.MissingProjectID,
'A project ID must be specified when not using a project '
'scoped token.',
self.service._is_valid_project_id, None
)
def test_is_valid_zone_name_invalid_object(self):
self.assertRaisesRegex(
exceptions.InvalidObject,
'',
self.service._is_valid_zone_name, self.context, None
)
def test_list_to_dict(self):
result = self.service._list_to_dict([{'region': 'foo', 'id': '1'}])
self.assertEqual(('1',), result.popitem()[0])
def test_check_zone_share_permission_raise_when_not_shared(self):
new_zone = zone.Zone()
new_zone.tenant_id = '2'
self.context.project_id = '1'
self.context.all_tenants = False
self.storage.is_zone_shared_with_project.return_value = False
self.assertRaisesRegex(
exceptions.ZoneNotFound, 'Could not find Zone',
self.service._check_zone_share_permission, self.context, new_zone
)
class ZoneAndRecordStatusTestCase(oslotest.base.BaseTestCase):
@mock.patch.object(policy, 'init')
@mock.patch.object(rpc, 'init')
@mock.patch.object(profiler, 'setup_profiler')
def setUp(self, mock_setup_profiler, mock_rpc_init, mock_policy_init):
super(ZoneAndRecordStatusTestCase, self).setUp()
self.service = service.Service()
mock_setup_profiler.assert_called()
mock_rpc_init.assert_called()
mock_policy_init.assert_called()
def test_success_pending(self):
new_zone = zone.Zone(action='CREATE', status='PENDING', serial=100)
result = self.service._update_zone_or_record_status(
new_zone, 'SUCCESS', 101
)
self.assertEqual('ACTIVE', result.status)
self.assertEqual('NONE', result.action)
new_record = record.Record(action='CREATE', status='PENDING',
serial=100)
result = self.service._update_zone_or_record_status(
new_record, 'SUCCESS', 101
)
self.assertEqual('ACTIVE', result.status)
self.assertEqual('NONE', result.action)
def test_success_error(self):
new_zone = zone.Zone(action='CREATE', status='ERROR', serial=100)
result = self.service._update_zone_or_record_status(
new_zone, 'SUCCESS', 101
)
self.assertEqual('ACTIVE', result.status)
self.assertEqual('NONE', result.action)
new_record = record.Record(action='CREATE', status='ERROR', serial=100)
result = self.service._update_zone_or_record_status(
new_record, 'SUCCESS', 101
)
self.assertEqual('ACTIVE', result.status)
self.assertEqual('NONE', result.action)
def test_success_delete(self):
new_zone = zone.Zone(action='DELETE', status='PENDING', serial=100)
result = self.service._update_zone_or_record_status(
new_zone, 'SUCCESS', 101
)
self.assertEqual('DELETED', result.status)
self.assertEqual('NONE', result.action)
new_record = record.Record(action='DELETE', status='PENDING',
serial=100)
result = self.service._update_zone_or_record_status(
new_record, 'SUCCESS', 101
)
self.assertEqual('DELETED', result.status)
self.assertEqual('NONE', result.action)
def test_success_do_nothing(self):
new_zone = zone.Zone(action='CREATE', status='PENDING', serial=2)
result = self.service._update_zone_or_record_status(
new_zone, 'SUCCESS', 1
)
self.assertEqual('PENDING', result.status)
self.assertEqual('CREATE', result.action)
new_record = record.Record(action='NONE', status='PENDING', serial=100)
result = self.service._update_zone_or_record_status(
new_record, 'SUCCESS', 101
)
self.assertEqual('PENDING', result.status)
self.assertEqual('NONE', result.action)
def test_error_pending(self):
new_zone = zone.Zone(action='CREATE', status='PENDING', serial=1)
result = self.service._update_zone_or_record_status(
new_zone, 'ERROR', 2
)
self.assertEqual('ERROR', result.status)
self.assertEqual('CREATE', result.action)
new_record = record.Record(action='DELETE', status='PENDING', serial=1)
result = self.service._update_zone_or_record_status(
new_record, 'ERROR', 2
)
self.assertEqual('ERROR', result.status)
self.assertEqual('DELETE', result.action)
def test_error_pending_do_nothing(self):
new_zone = zone.Zone(action='CREATE', status='PENDING', serial=100)
result = self.service._update_zone_or_record_status(
new_zone, 'ERROR', 0
)
self.assertEqual('ERROR', result.status)
self.assertEqual('CREATE', result.action)
new_record = record.Record(action='CREATE', status='PENDING',
serial=100)
result = self.service._update_zone_or_record_status(
new_record, 'ERROR', 0
)
self.assertEqual('ERROR', result.status)
self.assertEqual('CREATE', result.action)
def test_error_create_do_nothing(self):
new_zone = zone.Zone(action='CREATE', status='ACTIVE', serial=1)
result = self.service._update_zone_or_record_status(
new_zone, 'ERROR', 2
)
self.assertEqual('ACTIVE', result.status)
self.assertEqual('CREATE', result.action)
new_record = record.Record(action='CREATE', status='ACTIVE', serial=1)
result = self.service._update_zone_or_record_status(
new_record, 'ERROR', 2
)
self.assertEqual('ACTIVE', result.status)
self.assertEqual('CREATE', result.action)
def test_no_zone_create(self):
new_zone = zone.Zone(action='CREATE', status='PENDING', serial=100)
result = self.service._update_zone_or_record_status(
new_zone, 'NO_ZONE', 1
)
self.assertEqual('ERROR', result.status)
self.assertEqual('CREATE', result.action)
new_record = record.Record(action='CREATE', status='PENDING',
serial=100)
result = self.service._update_zone_or_record_status(
new_record, 'NO_ZONE', 1
)
self.assertEqual('ERROR', result.status)
self.assertEqual('CREATE', result.action)
def test_no_zone_update(self):
new_zone = zone.Zone(action='UPDATE', status='PENDING', serial=100)
result = self.service._update_zone_or_record_status(
new_zone, 'NO_ZONE', 1
)
self.assertEqual('ERROR', result.status)
self.assertEqual('CREATE', result.action)
new_record = record.Record(action='UPDATE', status='PENDING',
serial=100)
result = self.service._update_zone_or_record_status(
new_record, 'NO_ZONE', 1
)
self.assertEqual('ERROR', result.status)
self.assertEqual('CREATE', result.action)
def test_no_zone_delete(self):
new_zone = zone.Zone(action='DELETE', status='PENDING', serial=100)
result = self.service._update_zone_or_record_status(
new_zone, 'NO_ZONE', 1
)
self.assertEqual('DELETED', result.status)
self.assertEqual('NONE', result.action)
new_record = record.Record(action='DELETE', status='PENDING',
serial=100)
result = self.service._update_zone_or_record_status(
new_record, 'NO_ZONE', 1
)
self.assertEqual('DELETED', result.status)
self.assertEqual('NONE', result.action)
def test_no_zone_do_nothing(self):
new_zone = zone.Zone(action='NONE', status='SUCCESS', serial=100)
result = self.service._update_zone_or_record_status(
new_zone, 'NO_ZONE', 1
)
self.assertEqual('SUCCESS', result.status)
self.assertEqual('NONE', result.action)
new_record = record.Record(action='NONE', status='ACTIVE', serial=100)
result = self.service._update_zone_or_record_status(
new_record, 'NO_ZONE', 1
)
self.assertEqual('ACTIVE', result.status)
self.assertEqual('NONE', result.action)
def test_do_nothing(self):
new_zone = zone.Zone(action='NONE', status='SUCCESS', serial=100)
result = self.service._update_zone_or_record_status(
new_zone, None, None
)
self.assertEqual('SUCCESS', result.status)
self.assertEqual('NONE', result.action)
new_record = record.Record(action='NONE', status='ACTIVE', serial=100)
result = self.service._update_zone_or_record_status(
new_record, None, None
)
self.assertEqual('ACTIVE', result.status)
self.assertEqual('NONE', result.action)
designate/tests/unit/test_central/test_basic.py
+39 -13
View File
@@ -565,32 +565,44 @@ class CentralZoneTestCase(CentralBasic):
def test_is_valid_zone_name_invalid(self):
self.service._is_blacklisted_zone_name = mock.Mock()
with testtools.ExpectedException(exceptions.InvalidZoneName):
self.service._is_valid_zone_name(self.context,
'example^org.')
self.assertRaisesRegex(
exceptions.InvalidZoneName,
'More than one label is required',
self.service._is_valid_zone_name, self.context, 'example^org.'
)
def test_is_valid_zone_name_invalid_2(self):
self.service._is_blacklisted_zone_name = mock.Mock()
with testtools.ExpectedException(exceptions.InvalidZoneName):
self.service._is_valid_zone_name(self.context,
'example.tld.')
self.assertRaisesRegex(
exceptions.InvalidZoneName,
'Invalid TLD',
self.service._is_valid_zone_name, self.context, 'example.tld.'
)
def test_is_valid_zone_name_invalid_same_as_tld(self):
self.service._is_blacklisted_zone_name = mock.Mock()
with testtools.ExpectedException(exceptions.InvalidZoneName):
self.service._is_valid_zone_name(self.context, 'com.com.')
self.assertRaisesRegex(
exceptions.InvalidZoneName,
'Invalid TLD',
self.service._is_valid_zone_name, self.context, 'com.com.'
)
def test_is_valid_zone_name_invalid_tld(self):
self.service._is_blacklisted_zone_name = mock.Mock()
with testtools.ExpectedException(exceptions.InvalidZoneName):
self.service._is_valid_zone_name(self.context, 'tld.')
self.assertRaisesRegex(
exceptions.InvalidZoneName,
'More than one label is required',
self.service._is_valid_zone_name, self.context, 'tld.'
)
def test_is_valid_zone_name_blacklisted(self):
self.service._is_blacklisted_zone_name = mock.Mock(
side_effect=exceptions.InvalidZoneName)
with testtools.ExpectedException(exceptions.InvalidZoneName):
self.service._is_valid_zone_name(self.context,
'valid.com.')
self.assertRaisesRegex(
exceptions.InvalidZoneName,
'Invalid TLD',
self.service._is_valid_zone_name, self.context, 'valid.com.'
)
def test_is_blacklisted_zone_name(self):
self.service.storage.find_blacklists.return_value = [
@@ -1769,6 +1781,20 @@ class CentralZoneExportTests(CentralBasic):
designate.central.service.policy.check.call_args[0])
self.assertEqual('find_zone_exports', pcheck)
def test_find_zone_exports_with_custom_criterion(self):
self.context = mock.Mock()
self.context.project_id = 't'
self.service.storage.find_zone_exports = mock.Mock()
self.service.find_zone_exports(
self.context, criterion={'project_id': 't'}
)
self.assertTrue(self.service.storage.find_zone_exports.called)
pcheck, ctx, target = (
designate.central.service.policy.check.call_args[0])
self.assertEqual('find_zone_exports', pcheck)
def test_delete_zone_export(self):
self.context = mock.Mock()
self.context.project_id = 't'