openstack/neutron
Code Issues Proposed changes

Add initial tests for the Designate DNS backend

Browse Source 
This adds some basic unit tests for the Designate DNS backend.

Change-Id: I7846086aa07b96e2a2ba6296abeda07a2a4f2198
This commit is contained in:
Erik Olof Gunnar Andersson 2022-07-05 23:01:02 -07:00
parent 7b8c8104c6
commit a5ddebf070
4 changed files with 278 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
neutron/tests/unit/services/externaldns/__init__.py Normal file
View File

0
neutron/tests/unit/services/externaldns/drivers/__init__.py Normal file
View File

0
neutron/tests/unit/services/externaldns/drivers/designate/__init__.py Normal file
View File

278
neutron/tests/unit/services/externaldns/drivers/designate/test_driver.py Normal file
View File

@ -0,0 +1,278 @@
# 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 designateclient import exceptions as d_exc
from neutron_lib.exceptions import dns as dns_exc
from oslo_config import cfg
from neutron.services.externaldns.drivers.designate import driver
from neutron.tests import base
class TestDesignateDriver(base.BaseTestCase):
def setUp(self):
super(TestDesignateDriver, self).setUp()
self.context = mock.Mock()
self.client = mock.Mock()
self.admin_client = mock.Mock()
self.all_projects_client = mock.Mock()
mock.patch.object(driver, 'get_clients', return_value=(
self.client, self.admin_client)).start()
mock.patch.object(driver, 'get_all_projects_client',
return_value=self.all_projects_client).start()
self.driver = driver.Designate()
def test_create_record_set(self):
cfg.CONF.set_override(
'allow_reverse_dns_lookup', False, group='designate'
)
self.driver.create_record_set(
self.context, 'example.test.', 'test',
['192.168.0.10', '2001:db8:0:1::1']
)
self.client.recordsets.create.assert_has_calls(
[
mock.call('example.test.', 'test', 'A', ['192.168.0.10']),
mock.call('example.test.', 'test', 'AAAA', ['2001:db8:0:1::1'])
]
)
self.admin_client.recordsets.create.assert_not_called()
def test_create_record_set_with_reverse_dns(self):
self.driver.create_record_set(
self.context, 'example.test.', 'test',
['192.168.0.10', '2001:db8:0:1::1']
)
self.client.recordsets.create.assert_has_calls(
[
mock.call('example.test.', 'test', 'A', ['192.168.0.10']),
mock.call('example.test.', 'test', 'AAAA', ['2001:db8:0:1::1'])
]
)
self.admin_client.recordsets.create.assert_has_calls(
[
mock.call(
'0.168.192.in-addr.arpa.', '10.0.168.192.in-addr.arpa.',
'PTR', ['test.example.test.']
),
mock.call(
'0.0.0.0.0.0.0.0.0.0.0.0.0.0.1.0.0.0.0.0.0.0.8.b.d.0.1.0.'
'0.2.ip6.arpa.',
'1.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.1.0.0.0.0.0.0.0.8.b.d.0.'
'1.0.0.2.ip6.arpa.',
'PTR', ['test.example.test.']
),
]
)
def test_create_record_set_with_reverse_zone_conflict(self):
self.admin_client.recordsets.create.side_effect = [
d_exc.NotFound, None,
]
self.admin_client.zones.create.side_effect = d_exc.Conflict
self.driver.create_record_set(
self.context, 'example.test.', 'test',
['192.168.0.10']
)
self.assertEqual(1, self.client.recordsets.create.call_count)
self.assertEqual(1, self.admin_client.zones.create.call_count)
self.assertEqual(2, self.admin_client.recordsets.create.call_count)
def test_create_new_zone_when_missing(self):
self.admin_client.recordsets.create.side_effect = [
d_exc.NotFound, None, d_exc.NotFound, None
]
self.driver.create_record_set(
self.context, 'example.test.', 'test',
['192.168.0.10', '2001:db8:0:1::1']
)
self.admin_client.zones.create.assert_has_calls(
[
mock.call(
'0.168.192.in-addr.arpa.', email='admin@example.test',
description='An in-addr.arpa. zone for reverse lookups '
'set up by Neutron.'
),
mock.call(
'0.0.0.0.0.0.0.0.0.0.0.0.0.0.1.0.0.0.0.0.0.0.8.b.d.0.1.0.'
'0.2.ip6.arpa.',
email='admin@example.test',
description='An ip6.arpa. zone for reverse lookups set up '
'by Neutron.'
)
]
)
def test_create_new_zone_with_custom_email(self):
self.admin_client.recordsets.create.side_effect = [
d_exc.NotFound, None, d_exc.NotFound, None
]
cfg.CONF.set_override(
'ptr_zone_email', 'ptr@example.test', group='designate'
)
self.driver.create_record_set(
self.context, 'example.test.', 'test',
['192.168.0.10']
)
self.admin_client.zones.create.assert_has_calls(
[
mock.call(
'0.168.192.in-addr.arpa.', email='ptr@example.test',
description='An in-addr.arpa. zone for reverse lookups '
'set up by Neutron.'
),
]
)
def test_delete_record_set(self):
self.client.recordsets.list.return_value = [
{'id': 123, 'records': ['192.168.0.10']},
{'id': 456, 'records': ['2001:db8:0:1::1']}
]
cfg.CONF.set_override(
'allow_reverse_dns_lookup', False, group='designate'
)
self.driver.delete_record_set(
self.context, 'example.test.', 'test',
['192.168.0.10', '2001:db8:0:1::1']
)
self.client.recordsets.delete.assert_has_calls(
[
mock.call('example.test.', 123),
mock.call('example.test.', 456)
]
)
self.admin_client.recordsets.delete.assert_not_called()
def test_delete_record_set_with_reverse_dns(self):
self.client.recordsets.list.return_value = [
{'id': 123, 'records': ['192.168.0.10']},
{'id': 456, 'records': ['2001:db8:0:1::1']}
]
self.driver.delete_record_set(
self.context, 'example.test.', 'test',
['192.168.0.10', '2001:db8:0:1::1']
)
self.client.recordsets.delete.assert_has_calls(
[
mock.call('example.test.', 123),
mock.call('example.test.', 456)
]
)
self.admin_client.recordsets.delete.assert_has_calls(
[
mock.call(
'0.168.192.in-addr.arpa.', '10.0.168.192.in-addr.arpa.'
),
mock.call(
'0.0.0.0.0.0.0.0.0.0.0.0.0.0.1.0.0.0.0.0.0.0.8.b.d.0.1.0.'
'0.2.ip6.arpa.',
'1.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.1.0.0.0.0.0.0.0.8.b.d.0.'
'1.0.0.2.ip6.arpa.'
)
]
)
def test_create_record_set_zone_not_found(self):
self.client.recordsets.create.side_effect = d_exc.NotFound
self.assertRaisesRegex(
dns_exc.DNSDomainNotFound,
'Domain example.test. not found in the external DNS service',
self.driver.create_record_set, self.context, 'example.test.',
'test', ['192.168.0.10']
)
def test_create_record_set_duplicate_recordset(self):
self.client.recordsets.create.side_effect = d_exc.Conflict
self.assertRaisesRegex(
dns_exc.DuplicateRecordSet,
'Name test is duplicated in the external DNS service',
self.driver.create_record_set, self.context, 'example.test.',
'test', ['192.168.0.10']
)
def test_create_record_set_over_quota(self):
self.client.recordsets.create.side_effect = d_exc.OverQuota
self.assertRaisesRegex(
dns_exc.ExternalDNSOverQuota,
'External DNS Quota exceeded for resources: recordset.',
self.driver.create_record_set, self.context, 'example.test.',
'test', ['192.168.0.10']
)
def test_create_reverse_zone_over_quota(self):
self.admin_client.recordsets.create.side_effect = d_exc.NotFound
self.admin_client.zones.create.side_effect = d_exc.OverQuota
self.assertRaisesRegex(
dns_exc.ExternalDNSOverQuota,
'External DNS Quota exceeded for resources: zone.',
self.driver.create_record_set, self.context, 'example.test.',
'test', ['192.168.0.10']
)
def test_delete_record_set_zone_not_found(self):
self.client.recordsets.list.side_effect = d_exc.NotFound
self.all_projects_client.recordsets.list.side_effect = d_exc.NotFound
self.assertRaisesRegex(
dns_exc.DNSDomainNotFound,
'Domain example.test. not found in the external DNS service',
self.driver.delete_record_set, self.context, 'example.test.',
'test', ['192.168.0.10']
)
def test_ipv4_ptr_is_misconfigured(self):
cfg.CONF.set_override(
'ipv4_ptr_zone_prefix_size', -1, group='designate'
)
self.assertRaisesRegex(
dns_exc.InvalidPTRZoneConfiguration,
'Value of ipv4_ptr_zone_size has to be multiple of 8, with '
'maximum value of 24 and minimum value of 8',
driver.Designate
)
def test_ipv6_ptr_is_misconfigured(self):
cfg.CONF.set_override(
'ipv6_ptr_zone_prefix_size', -1, group='designate'
)
self.assertRaisesRegex(
dns_exc.InvalidPTRZoneConfiguration,
'Value of ipv6_ptr_zone_size has to be multiple of 4, with '
'maximum value of 124 and minimum value of 4',
driver.Designate
)