openstack/designate
Code Issues Proposed changes

Merge "Modernized backend tests"

Browse Source
This commit is contained in:
Zuul 2019-05-25 20:06:38 +00:00 committed by Gerrit Code Review
commit 4e49b5581f
10 changed files with 569 additions and 566 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

138
designate/tests/test_backend/test_infoblox.py
View File

@ -1,138 +0,0 @@
# Copyright 2015 Infoblox Inc.
# All Rights Reserved.
#
# 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.
import six
from mock import MagicMock
from designate import objects
from designate.tests.test_backend import BackendTestCase
from designate.backend.impl_infoblox import InfobloxBackend
from designate.exceptions import ConfigurationError
from designate.backend.impl_infoblox import ibexceptions
class InfobloxBackendTestCase(BackendTestCase):
def get_zone_fixture(self):
return super(InfobloxBackendTestCase, self).get_zone_fixture(
values={
'name': 'test.example.com.'
}
)
def setUp(self):
super(InfobloxBackendTestCase, self).setUp()
self.config(group='backend:infoblox',
wapi_url=None,
username=None,
password=None,
ns_group=None)
def get_target_fixture(self, masters=None, options=None):
if not masters:
masters = [{'host': '1.1.1.1', 'port': 53}]
if not options:
options = [{'key': 'wapi_url', 'value': 'test'},
{'key': 'username', 'value': 'test'},
{'key': 'password', 'value': 'test'},
{'key': 'ns_group', 'value': 'test'}]
return objects.PoolTarget.from_dict({
'id': '4588652b-50e7-46b9-b688-a9bad40a873e',
'type': 'infoblox',
'masters': masters,
'options': options
})
def set_up_backend(self, target=None):
if not target:
target = self.get_target_fixture()
self.backend = InfobloxBackend(target)
self.backend.start()
self.backend.infoblox = MagicMock()
def test_create_zone(self):
self.set_up_backend()
context = self.get_context()
zone = self.get_zone_fixture()
self.backend.infoblox.get_dns_view = MagicMock(return_value='default')
self.backend.create_zone(context, zone)
self.backend.infoblox.create_zone_auth.assert_called_once_with(
fqdn='test.example.com',
dns_view='default')
def test_update_zone(self):
self.set_up_backend()
context = self.get_context()
zone = objects.Zone().from_dict(self.get_zone_fixture())
self.backend.update_zone(context, zone)
def test_delete_zone(self):
self.set_up_backend()
context = self.get_context()
zone = self.get_zone_fixture()
self.backend.create_zone(context, zone)
self.backend.delete_zone(context, zone)
self.backend.infoblox.delete_zone_auth.assert_called_once_with(
'test.example.com')
def test_missing_wapi_url(self):
options = [{'key': 'username', 'value': 'test'},
{'key': 'password', 'value': 'test'},
{'key': 'ns_group', 'value': 'test'}]
target = self.get_target_fixture(options=options)
six.assertRaisesRegex(self, ibexceptions.InfobloxIsMisconfigured,
"wapi_url",
self.set_up_backend, target)
def test_missing_username(self):
options = [{'key': 'wapi_url', 'value': 'test'},
{'key': 'password', 'value': 'test'},
{'key': 'ns_group', 'value': 'test'}]
target = self.get_target_fixture(options=options)
six.assertRaisesRegex(self, ibexceptions.InfobloxIsMisconfigured,
"username",
self.set_up_backend, target)
def test_missing_password(self):
options = [{'key': 'wapi_url', 'value': 'test'},
{'key': 'username', 'value': 'test'},
{'key': 'ns_group', 'value': 'test'}]
target = self.get_target_fixture(options=options)
six.assertRaisesRegex(self, ibexceptions.InfobloxIsMisconfigured,
"password",
self.set_up_backend, target)
def test_missing_ns_group(self):
options = [{'key': 'wapi_url', 'value': 'test'},
{'key': 'username', 'value': 'test'},
{'key': 'password', 'value': 'test'}]
target = self.get_target_fixture(options=options)
six.assertRaisesRegex(self, ibexceptions.InfobloxIsMisconfigured,
"ns_group",
self.set_up_backend, target)
def test_wrong_port(self):
masters = [{'host': '1.1.1.1', 'port': 100}]
target = self.get_target_fixture(masters=masters)
six.assertRaisesRegex(self, ConfigurationError,
"port 53",
self.set_up_backend, target)

0
designate/tests/unit/test_backend/__init__.py → designate/tests/unit/backend/__init__.py
View File

224
designate/tests/unit/backend/test_agent.py Normal file
View File

@ -0,0 +1,224 @@
# Author: Federico Ceratto <federico.ceratto@hpe.com>
#
# 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.
import dns
import dns.rdataclass
import dns.rdatatype
import mock
import designate.backend.agent as agent
import designate.backend.private_codes as pcodes
from designate import exceptions
from designate import objects
from designate import tests
from designate.mdns import rpcapi as mdns_api
from designate.tests.unit import RoObject
class AgentBackendTestCase(tests.TestCase):
def setUp(self):
super(AgentBackendTestCase, self).setUp()
self.CONF.set_override('poll_timeout', 1, 'service:pool_manager')
self.CONF.set_override('poll_retry_interval', 4,
'service:pool_manager')
self.CONF.set_override('poll_max_retries', 5, 'service:pool_manager')
self.CONF.set_override('poll_delay', 6, 'service:pool_manager')
self.context = self.get_context()
self.zone = objects.Zone(
id='e2bed4dc-9d01-11e4-89d3-123b93f75cba',
name='example.com.',
email='example@example.com',
)
self.target = {
'id': '4588652b-50e7-46b9-b688-a9bad40a873e',
'type': 'agent',
'masters': [],
'options': [
{'key': 'host', 'value': 2},
{'key': 'port', 'value': 3},
],
}
self.backend = agent.AgentPoolBackend(
objects.PoolTarget.from_dict(self.target)
)
@mock.patch.object(mdns_api.MdnsAPI, 'get_instance')
def test_mdns_api(self, mock_get_instance):
self.assertIsInstance(self.backend.mdns_api, mock.Mock)
@mock.patch.object(mdns_api.MdnsAPI, 'get_instance')
def test_create_zone(self, mock_get_instance):
self.backend._make_and_send_dns_message = mock.Mock(
return_value=(1, 2))
out = self.backend.create_zone(self.context, self.zone)
self.backend._make_and_send_dns_message.assert_called_with(
self.zone.name, 1, 14, pcodes.CREATE, pcodes.SUCCESS, 2, 3)
self.assertIsNone(out)
@mock.patch.object(mdns_api.MdnsAPI, 'get_instance')
def test_create_zone_exception(self, mock_get_instance):
self.backend._make_and_send_dns_message = mock.Mock(
return_value=(None, 2))
self.assertRaisesRegex(
exceptions.Backend, 'create_zone.* failed',
self.backend.create_zone, self.context, self.zone,
)
self.backend._make_and_send_dns_message.assert_called_with(
self.zone.name, 1, 14, pcodes.CREATE, pcodes.SUCCESS, 2, 3)
@mock.patch.object(mdns_api.MdnsAPI, 'get_instance')
def test_update_zone(self, mock_get_instance):
self.backend.mdns_api.notify_zone_changed = mock.Mock()
out = self.backend.update_zone(self.context, self.zone)
self.backend.mdns_api.notify_zone_changed.assert_called_with(
self.context, self.zone, 2, 3, 1, 4, 5, 6)
self.assertIsNone(out)
@mock.patch.object(mdns_api.MdnsAPI, 'get_instance')
def test_delete_zone(self, mock_get_instance):
self.backend._make_and_send_dns_message = mock.Mock(
return_value=(1, 2))
out = self.backend.delete_zone(self.context, self.zone)
self.backend._make_and_send_dns_message.assert_called_with(
self.zone.name, 1, 14, pcodes.DELETE, pcodes.SUCCESS, 2, 3)
self.assertIsNone(out)
@mock.patch.object(mdns_api.MdnsAPI, 'get_instance')
def test_delete_zone_exception(self, mock_get_instance):
self.backend._make_and_send_dns_message = mock.Mock(
return_value=(None, 2))
self.assertRaisesRegex(
exceptions.Backend, 'failed delete_zone',
self.backend.delete_zone, self.context, self.zone,
)
self.backend._make_and_send_dns_message.assert_called_with(
self.zone.name, 1, 14, pcodes.DELETE, pcodes.SUCCESS, 2, 3)
def test_make_and_send_dns_message_timeout(self):
self.backend._make_dns_message = mock.Mock(return_value='')
self.backend._send_dns_message = mock.Mock(
return_value=dns.exception.Timeout())
out = self.backend._make_and_send_dns_message('h', 123, 1, 2, 3, 4, 5)
self.assertEqual((None, 0), out)
def test_make_and_send_dns_message_bad_response(self):
self.backend._make_dns_message = mock.Mock(return_value='')
self.backend._send_dns_message = mock.Mock(
return_value=agent.dns_query.BadResponse())
out = self.backend._make_and_send_dns_message('h', 123, 1, 2, 3, 4, 5)
self.assertEqual((None, 0), out)
def test_make_and_send_dns_message_missing_AA_flags(self):
self.backend._make_dns_message = mock.Mock(return_value='')
response = RoObject(
rcode=mock.Mock(return_value=dns.rcode.NOERROR),
# rcode is NOERROR but (flags & dns.flags.AA) gives 0
flags=0,
)
self.backend._send_dns_message = mock.Mock(return_value=response)
out = self.backend._make_and_send_dns_message('h', 123, 1, 2, 3, 4, 5)
self.assertEqual((None, 0), out)
def test_make_and_send_dns_message_error_flags(self):
self.backend._make_dns_message = mock.Mock(return_value='')
response = RoObject(
rcode=mock.Mock(return_value=dns.rcode.NOERROR),
# rcode is NOERROR but flags are not NOERROR
flags=123,
ednsflags=321
)
self.backend._send_dns_message = mock.Mock(return_value=response)
out = self.backend._make_and_send_dns_message('h', 123, 1, 2, 3, 4, 5)
self.assertEqual((None, 0), out)
def test_make_and_send_dns_message(self):
self.backend._make_dns_message = mock.Mock(return_value='')
response = RoObject(
rcode=mock.Mock(return_value=dns.rcode.NOERROR),
flags=agent.dns.flags.AA,
ednsflags=321
)
self.backend._send_dns_message = mock.Mock(return_value=response)
out = self.backend._make_and_send_dns_message('h', 123, 1, 2, 3, 4, 5)
self.assertEqual((response, 0), out)
@mock.patch.object(agent.dns_query, 'tcp')
@mock.patch.object(agent.dns_query, 'udp')
def test_send_dns_message(self, mock_udp, mock_tcp):
mock_udp.return_value = 'mock udp resp'
out = self.backend._send_dns_message('msg', 'host', 123, 1)
self.assertFalse(agent.dns_query.tcp.called)
agent.dns_query.udp.assert_called_with('msg', 'host', port=123,
timeout=1)
self.assertEqual('mock udp resp', out)
@mock.patch.object(agent.dns_query, 'tcp')
@mock.patch.object(agent.dns_query, 'udp')
def test_send_dns_message_timeout(self, mock_udp, mock_tcp):
mock_udp.side_effect = dns.exception.Timeout
out = self.backend._send_dns_message('msg', 'host', 123, 1)
agent.dns_query.udp.assert_called_with('msg', 'host', port=123,
timeout=1)
self.assertIsInstance(out, dns.exception.Timeout)
@mock.patch.object(agent.dns_query, 'tcp')
@mock.patch.object(agent.dns_query, 'udp')
def test_send_dns_message_bad_response(self, mock_udp, mock_tcp):
mock_udp.side_effect = agent.dns_query.BadResponse
out = self.backend._send_dns_message('msg', 'host', 123, 1)
agent.dns_query.udp.assert_called_with('msg', 'host', port=123,
timeout=1)
self.assertIsInstance(out, agent.dns_query.BadResponse)
@mock.patch.object(agent.dns_query, 'tcp')
@mock.patch.object(agent.dns_query, 'udp')
def test_send_dns_message_tcp(self, mock_udp, mock_tcp):
self.CONF.set_override('all_tcp', True, 'service:mdns')
mock_tcp.return_value = 'mock tcp resp'
out = self.backend._send_dns_message('msg', 'host', 123, 1)
self.assertFalse(agent.dns_query.udp.called)
agent.dns_query.tcp.assert_called_with('msg', 'host', port=123,
timeout=1)
self.assertEqual('mock tcp resp', out)

91
designate/tests/unit/backend/test_designate.py Normal file
View File

@ -0,0 +1,91 @@
# Copyright 2015 Hewlett-Packard Development Company, L.P.
#
# Author: Endre Karlson <endre.karlson@hpe.com>
#
# 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.
import testtools
from designateclient import exceptions
from mock import NonCallableMagicMock
from mock import patch
from oslo_log import log as logging
from designate import objects
from designate import tests
from designate.backend import impl_designate
LOG = logging.getLogger(__name__)
class DesignateBackendTestCase(tests.TestCase):
def setUp(self):
super(DesignateBackendTestCase, self).setUp()
self.zone = objects.Zone(
id='e2bed4dc-9d01-11e4-89d3-123b93f75cba',
name='example.com.',
email='example@example.com',
)
self.target = {
'id': '4588652b-50e7-46b9-b688-a9bad40a873e',
'type': 'designate',
'masters': [
{'host': '192.0.2.1', 'port': 53},
],
'options': [
{'key': 'username', 'value': 'user'},
{'key': 'password', 'value': 'secret'},
{'key': 'project_name', 'value': 'project'},
{'key': 'project_zone_name', 'value': 'project_zone'},
{'key': 'user_zone_name', 'value': 'user_zone'},
],
}
self.backend = impl_designate.DesignateBackend(
objects.PoolTarget.from_dict(self.target)
)
# Mock client
self.client = NonCallableMagicMock()
zones = NonCallableMagicMock(spec_set=[
'create', 'delete'])
self.client.configure_mock(zones=zones)
def test_create_zone(self):
masters = ["%(host)s:%(port)s" % self.target['masters'][0]]
with patch.object(self.backend, '_get_client',
return_value=self.client):
self.backend.create_zone(self.admin_context, self.zone)
self.client.zones.create.assert_called_once_with(
self.zone.name, 'SECONDARY', masters=masters)
def test_delete_zone(self):
with patch.object(self.backend, '_get_client',
return_value=self.client):
self.backend.delete_zone(self.admin_context, self.zone)
self.client.zones.delete.assert_called_once_with(self.zone.name)
def test_delete_zone_notfound(self):
self.client.delete.side_effect = exceptions.NotFound
with patch.object(self.backend, '_get_client',
return_value=self.client):
self.backend.delete_zone(self.admin_context, self.zone)
self.client.zones.delete.assert_called_once_with(self.zone.name)
def test_delete_zone_exc(self):
self.client.zones.delete.side_effect = Exception
with testtools.ExpectedException(Exception):
with patch.object(self.backend, '_get_client',
return_value=self.client):
self.backend.delete_zone(self.admin_context, self.zone)
self.client.zones.delete.assert_called_once_with(self.zone.name)

112
designate/tests/test_backend/test_dynect.py → designate/tests/unit/backend/test_dynect.py
View File

@ -13,19 +13,15 @@
# 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_serialization import jsonutils
from requests_mock.contrib import fixture as req_fixture
import testtools
import requests_mock
from designate import objects
from designate import tests
from designate.backend import impl_dynect
from designate.tests.test_backend import BackendTestCase
MASTERS = ["10.0.0.1"]
CONTACT = 'jdoe@myco.biz'
LOGIN_SUCCESS = {
"status": "success",
"data": {
@ -93,7 +89,6 @@ TARGET_EXISTS = {
]
}
ACTIVATE_SUCCESS = {
"status": "success",
"data": {
@ -115,73 +110,74 @@ ACTIVATE_SUCCESS = {
}
class DynECTTestsCase(BackendTestCase):
class DynECTTestsCase(tests.TestCase):
def setUp(self):
super(DynECTTestsCase, self).setUp()
self.target = objects.PoolTarget.from_dict({
self.base_address = 'https://api.dynect.net:443/REST'
self.context = self.get_context()
self.zone = objects.Zone(
id='e2bed4dc-9d01-11e4-89d3-123b93f75cba',
name='example.com.',
email='example@example.com',
)
self.target = {
'id': '4588652b-50e7-46b9-b688-a9bad40a873e',
'type': 'dyndns',
'masters': [{'host': '192.0.2.1', 'port': 53}],
'masters': [
{'host': '192.0.2.1', 'port': 53}
],
'options': [
{'key': 'username', 'value': 'example'},
{'key': 'password', 'value': 'secret'},
{'key': 'customer_name', 'value': 'customer'}],
})
}
self.backend = impl_dynect.DynECTBackend(self.target)
self.requests = self.useFixture(req_fixture.Fixture())
self.backend = impl_dynect.DynECTBackend(
objects.PoolTarget.from_dict(self.target)
)
def stub_url(self, method, parts=None, base_url=None, json=None, **kwargs):
if not base_url:
base_url = 'https://api.dynect.net:443/REST'
@requests_mock.mock()
def test_create_zone_raise_dynclienterror(self, req_mock):
# https://api.dynect.net:443/REST/Session
req_mock.post(
'%s/Session' % self.base_address,
json=LOGIN_SUCCESS,
)
if json:
kwargs['text'] = jsonutils.dumps(json)
headers = kwargs.setdefault('headers', {})
headers['Content-Type'] = 'application/json'
if parts:
url = '/'.join([p.strip('/') for p in [base_url] + parts])
else:
url = base_url
url = url.replace("/?", "?")
return self.requests.register_uri(method, url, **kwargs)
def _stub_login(self):
self.stub_url('POST', ['/Session'], json=LOGIN_SUCCESS)
self.stub_url('DELETE', ['/Session'], json=LOGIN_SUCCESS)
def test_create_zone_raise_dynclienterror(self):
context = self.get_context()
zone = self.create_zone()
self._stub_login()
self.stub_url(
'POST', ['/Secondary/example.com'],
req_mock.post(
'%s/Secondary/example.com' % self.base_address,
json=INVALID_MASTER_DATA,
status_code=400)
status_code=400,
)
with testtools.ExpectedException(impl_dynect.DynClientError):
self.backend.create_zone(context, zone)
self.assertRaisesRegex(
impl_dynect.DynClientError, 'Zone not created',
self.backend.create_zone, self.context, self.zone,
)
def test_create_zone_duplicate_updates_existing(self):
context = self.get_context()
zone = self.create_zone()
@requests_mock.mock()
def test_create_zone_duplicate_updates_existing(self, req_mock):
req_mock.post(
'%s/Session' % self.base_address,
json=LOGIN_SUCCESS,
)
self._stub_login()
req_mock.delete(
'%s/Session' % self.base_address,
json=LOGIN_SUCCESS,
)
parts = ['/Secondary', '/%s' % zone['name'].rstrip('.')]
self.stub_url(
'POST', parts,
req_mock.post(
'%s/Secondary/example.com' % self.base_address,
json=TARGET_EXISTS,
status_code=400)
status_code=400,
)
update = self.stub_url('PUT', parts, json=ACTIVATE_SUCCESS)
req_mock.put(
'%s/Secondary/example.com' % self.base_address,
json=ACTIVATE_SUCCESS,
)
self.backend.create_zone(context, zone)
self.assertTrue(update.called)
self.backend.create_zone(self.context, self.zone)

164
designate/tests/unit/backend/test_infoblox.py Normal file
View File

@ -0,0 +1,164 @@
# Copyright 2015 Infoblox Inc.
# All Rights Reserved.
#
# 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.
import mock
import requests_mock
from designate import exceptions
from designate import objects
from designate import tests
from designate.backend import impl_infoblox
from designate.backend.impl_infoblox import ibexceptions
from designate.mdns import rpcapi as mdns_rpcapi
class InfobloxBackendTestCase(tests.TestCase):
def setUp(self):
super(InfobloxBackendTestCase, self).setUp()
self.base_address = 'https://localhost/wapi'
self.context = self.get_context()
self.zone = objects.Zone(
id='e2bed4dc-9d01-11e4-89d3-123b93f75cba',
name='example.com.',
email='example@example.com',
)
self.target = {
'id': '4588652b-50e7-46b9-b688-a9bad40a873e',
'type': 'infoblox',
'masters': [
{'host': '1.1.1.1', 'port': 53},
],
'options': [
{'key': 'wapi_url', 'value': 'https://localhost/wapi/v2.0/'},
{'key': 'username', 'value': 'test'},
{'key': 'password', 'value': 'test'},
{'key': 'ns_group', 'value': 'test'},
]
}
self.backend = impl_infoblox.InfobloxBackend(
objects.PoolTarget.from_dict(self.target)
)
@requests_mock.mock()
def test_create_zone(self, req_mock):
req_mock.post(
'%s/v2.0/zone_auth' % self.base_address,
json={},
)
req_mock.get(
'%s/v2.0/zone_auth' % self.base_address,
json={},
)
self.backend.create_zone(self.context, self.zone)
@mock.patch.object(mdns_rpcapi.MdnsAPI, 'notify_zone_changed')
def test_update_zone(self, mock_notify_zone_changed):
self.backend.update_zone(self.context, self.zone)
mock_notify_zone_changed.assert_called_with(
self.context, self.zone, '127.0.0.1', 53, 30, 15, 10, 5)
@requests_mock.mock()
def test_delete_zone(self, req_mock):
req_mock.post(
'%s/v2.0/zone_auth' % self.base_address,
json={},
)
req_mock.get(
'%s/v2.0/zone_auth' % self.base_address,
json={},
)
self.backend.create_zone(self.context, self.zone)
self.backend.delete_zone(self.context, self.zone)
def test_missing_wapi_url(self):
target = dict(self.target)
target['options'] = [
{'key': 'username', 'value': 'test'},
{'key': 'password', 'value': 'test'},
{'key': 'ns_group', 'value': 'test'},
]
pool_target = objects.PoolTarget.from_dict(target)
self.assertRaisesRegex(
ibexceptions.InfobloxIsMisconfigured, "wapi_url",
impl_infoblox.InfobloxBackend, pool_target,
)
def test_missing_username(self):
target = dict(self.target)
target['options'] = [
{'key': 'wapi_url', 'value': 'test'},
{'key': 'password', 'value': 'test'},
{'key': 'ns_group', 'value': 'test'}
]
pool_target = objects.PoolTarget.from_dict(target)
self.assertRaisesRegex(
ibexceptions.InfobloxIsMisconfigured, "username",
impl_infoblox.InfobloxBackend, pool_target,
)
def test_missing_password(self):
target = dict(self.target)
target['options'] = [
{'key': 'wapi_url', 'value': 'test'},
{'key': 'username', 'value': 'test'},
{'key': 'ns_group', 'value': 'test'},
]
pool_target = objects.PoolTarget.from_dict(target)
self.assertRaisesRegex(
ibexceptions.InfobloxIsMisconfigured, "password",
impl_infoblox.InfobloxBackend, pool_target,
)
def test_missing_ns_group(self):
target = dict(self.target)
target['options'] = [
{'key': 'wapi_url', 'value': 'test'},
{'key': 'username', 'value': 'test'},
{'key': 'password', 'value': 'test'},
]
pool_target = objects.PoolTarget.from_dict(target)
self.assertRaisesRegex(
ibexceptions.InfobloxIsMisconfigured, "ns_group",
impl_infoblox.InfobloxBackend, pool_target,
)
def test_wrong_port(self):
target = dict(self.target)
target['masters'] = [
{'host': '1.1.1.1', 'port': 100},
]
pool_target = objects.PoolTarget.from_dict(target)
self.assertRaisesRegex(
exceptions.ConfigurationError,
'Infoblox only supports mDNS instances on port 53',
impl_infoblox.InfobloxBackend, pool_target,
)

57
designate/tests/test_backend/test_nsd4.py → designate/tests/unit/backend/test_nsd4.py
View File

@ -13,7 +13,6 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import socket
import ssl
@ -22,34 +21,43 @@ import mock
from designate import exceptions
from designate import objects
from designate.tests.test_backend import BackendTestCase
from designate import tests
from designate.backend import impl_nsd4
# NOTE: We'll only test the specifics to the nsd4 backend here.
# Rest is handled via scenarios
class NSD4BackendTestCase(BackendTestCase):
class NSD4BackendTestCase(tests.TestCase):
def setUp(self):
super(NSD4BackendTestCase, self).setUp()
# NOTE(hieulq): we mock out NSD4 back-end with random port
keyfile = mock.sentinel.key
certfile = mock.sentinel.cert
self.context = self.get_context()
self.zone = objects.Zone(
id='e2bed4dc-9d01-11e4-89d3-123b93f75cba',
name='example.com.',
email='example@example.com',
)
self.port = 6969
self.target = objects.PoolTarget.from_dict({
self.target = {
'id': '4588652b-50e7-46b9-b688-a9bad40a873e',
'type': 'nsd4',
'masters': [{'host': '192.0.2.1', 'port': 53},
{'host': '192.0.2.2', 'port': 35}],
'masters': [
{'host': '192.0.2.1', 'port': 53},
{'host': '192.0.2.2', 'port': 35},
],
'options': [
{'key': 'keyfile', 'value': keyfile.name},
{'key': 'certfile', 'value': certfile.name},
{'key': 'pattern', 'value': 'test-pattern'},
{'key': 'port', 'value': str(self.port)}
{'key': 'port', 'value': str(self.port)},
],
})
self.backend = impl_nsd4.NSD4Backend(self.target)
}
self.backend = impl_nsd4.NSD4Backend(
objects.PoolTarget.from_dict(self.target)
)
@mock.patch.object(eventlet, 'connect')
@mock.patch.object(eventlet, 'wrap_ssl')
@ -64,20 +72,17 @@ class NSD4BackendTestCase(BackendTestCase):
else:
stream.read.return_value = 'ok'
context = self.get_context()
zone = self.get_zone_fixture()
if command_context is 'create':
self.backend.create_zone(context, zone)
command = 'NSDCT1 addzone %s test-pattern\n' % zone['name']
self.backend.create_zone(self.context, self.zone)
command = 'NSDCT1 addzone %s test-pattern\n' % self.zone.name
elif command_context is 'delete':
self.backend.delete_zone(context, zone)
command = 'NSDCT1 delzone %s\n' % zone['name']
self.backend.delete_zone(self.context, self.zone)
command = 'NSDCT1 delzone %s\n' % self.zone.name
elif command_context is 'create_fail':
self.assertRaises(exceptions.Backend,
self.backend.create_zone,
context, zone)
command = 'NSDCT1 addzone %s test-pattern\n' % zone['name']
self.context, self.zone)
command = 'NSDCT1 addzone %s test-pattern\n' % self.zone.name
stream.write.assert_called_once_with(command)
mock_ssl.assert_called_once_with(mock.sentinel.client,
@ -101,16 +106,12 @@ class NSD4BackendTestCase(BackendTestCase):
def test_ssl_error(self):
self.backend._command = mock.MagicMock(side_effect=ssl.SSLError)
context = self.get_context()
zone = self.get_zone_fixture()
self.assertRaises(exceptions.Backend,
self.backend.create_zone,
context, zone)
self.context, self.zone)
def test_socket_error(self):
self.backend._command = mock.MagicMock(side_effect=socket.error)
context = self.get_context()
zone = self.get_zone_fixture()
self.assertRaises(exceptions.Backend,
self.backend.create_zone,
context, zone)
self.context, self.zone)

14
designate/tests/unit/test_backend/test_pdns4.py → designate/tests/unit/backend/test_pdns4.py
View File

@ -14,24 +14,25 @@ import requests_mock
from designate import exceptions
from designate import objects
from designate import tests
from designate.backend import impl_pdns4
from designate.mdns import rpcapi as mdns_rpcapi
from designate.tests import fixtures
from designate.tests.test_backend import BackendTestCase
class PDNS4BackendTestCase(BackendTestCase):
class PDNS4BackendTestCase(tests.TestCase):
def setUp(self):
super(PDNS4BackendTestCase, self).setUp()
self.stdlog = fixtures.StandardLogging()
self.useFixture(self.stdlog)
self.base_address = 'http://localhost:8081/api/v1/servers'
self.context = self.get_context()
self.zone = objects.Zone(id='e2bed4dc-9d01-11e4-89d3-123b93f75cba',
name='example.com.',
email='example@example.com')
self.zone = objects.Zone(
id='e2bed4dc-9d01-11e4-89d3-123b93f75cba',
name='example.com.',
email='example@example.com',
)
self.target = {
'id': '4588652b-50e7-46b9-b688-a9bad40a873e',
@ -53,7 +54,6 @@ class PDNS4BackendTestCase(BackendTestCase):
@requests_mock.mock()
@mock.patch.object(mdns_rpcapi.MdnsAPI, 'notify_zone_changed')
def test_create_zone_success(self, req_mock, mock_notify_zone_changed):
req_mock.post(
'%s/localhost/zones' % self.base_address,
)

212
designate/tests/unit/test_backend/test_agent.py
View File

@ -1,212 +0,0 @@
# Author: Federico Ceratto <federico.ceratto@hpe.com>
#
# 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.
"""Unit-test backend agent
"""
from mock import MagicMock
from mock import Mock
from mock import patch
from oslotest import base
import dns
import dns.rdataclass
import dns.rdatatype
import mock
import testtools
from designate import exceptions
from designate.tests.unit import RoObject
import designate.backend.agent as agent
import designate.backend.private_codes as pcodes
class SCAgentPoolBackend(agent.AgentPoolBackend):
def __init__(self):
pass
@mock.patch.object(agent.mdns_api.MdnsAPI, 'get_instance')
@patch.object(agent.base.Backend, '__init__')
class BackendAgentTest(base.BaseTestCase):
def setUp(self, *mocks):
super(BackendAgentTest, self).setUp()
agent.CONF = RoObject({
'service:mdns': RoObject(all_tcp=False)
})
self.agent = SCAgentPoolBackend()
self.agent.timeout = 1
self.agent.host = 2
self.agent.port = 3
self.agent.retry_interval = 4
self.agent.max_retries = 5
self.agent.delay = 6
def test_mdns_api(self, *mock):
assert isinstance(self.agent.mdns_api, MagicMock)
def test_create_zone(self, *mock):
self.agent._make_and_send_dns_message = Mock(return_value=(1, 2))
out = self.agent.create_zone('ctx', RoObject(name='zn'))
self.agent._make_and_send_dns_message.assert_called_with(
'zn', 1, 14, pcodes.CREATE, pcodes.SUCCESS, 2, 3)
self.assertIsNone(out)
def test_create_zone_exception(self, *mock):
self.agent._make_and_send_dns_message = Mock(return_value=(None, 2))
with testtools.ExpectedException(exceptions.Backend):
self.agent.create_zone('ctx', RoObject(name='zn'))
self.agent._make_and_send_dns_message.assert_called_with(
'zn', 1, 14, pcodes.CREATE, pcodes.SUCCESS, 2, 3)
def test_update_zone(self, *mock):
self.agent.mdns_api.notify_zone_changed = Mock()
zone = RoObject(name='zn')
out = self.agent.update_zone('ctx', zone)
self.agent.mdns_api.notify_zone_changed.assert_called_with(
'ctx', zone, 2, 3, 1, 4, 5, 6)
self.assertIsNone(out)
def test_delete_zone(self, *mock):
self.agent._make_and_send_dns_message = Mock(return_value=(1, 2))
out = self.agent.delete_zone('ctx', RoObject(name='zn'))
self.agent._make_and_send_dns_message.assert_called_with(
'zn', 1, 14, pcodes.DELETE, pcodes.SUCCESS, 2, 3)
self.assertIsNone(out)
def test_delete_zone_exception(self, *mock):
self.agent._make_and_send_dns_message = Mock(return_value=(None, 2))
with testtools.ExpectedException(exceptions.Backend):
self.agent.delete_zone('ctx', RoObject(name='zn'))
self.agent._make_and_send_dns_message.assert_called_with(
'zn', 1, 14, pcodes.DELETE, pcodes.SUCCESS, 2, 3)
def test_make_and_send_dns_message_timeout(self, *mocks):
self.agent._make_dns_message = Mock(return_value='')
self.agent._send_dns_message = Mock(
return_value=dns.exception.Timeout())
out = self.agent._make_and_send_dns_message('h', 123, 1, 2, 3, 4, 5)
self.assertEqual((None, 0), out)
def test_make_and_send_dns_message_bad_response(self, *mocks):
self.agent._make_dns_message = Mock(return_value='')
self.agent._send_dns_message = Mock(
return_value=agent.dns_query.BadResponse())
out = self.agent._make_and_send_dns_message('h', 123, 1, 2, 3, 4, 5)
self.assertEqual((None, 0), out)
def test_make_and_send_dns_message_missing_AA_flags(self, *mocks):
self.agent._make_dns_message = Mock(return_value='')
response = RoObject(
rcode=Mock(return_value=dns.rcode.NOERROR),
# rcode is NOERROR but (flags & dns.flags.AA) gives 0
flags=0,
)
self.agent._send_dns_message = Mock(return_value=response)
out = self.agent._make_and_send_dns_message('h', 123, 1, 2, 3, 4, 5)
self.assertEqual((None, 0), out)
def test_make_and_send_dns_message_error_flags(self, *mocks):
self.agent._make_dns_message = Mock(return_value='')
response = RoObject(
rcode=Mock(return_value=dns.rcode.NOERROR),
# rcode is NOERROR but flags are not NOERROR
flags=123,
ednsflags=321
)
self.agent._send_dns_message = Mock(return_value=response)
out = self.agent._make_and_send_dns_message('h', 123, 1, 2, 3, 4, 5)
self.assertEqual((None, 0), out)
def test_make_and_send_dns_message(self, *mock):
self.agent._make_dns_message = Mock(return_value='')
response = RoObject(
rcode=Mock(return_value=dns.rcode.NOERROR),
flags=agent.dns.flags.AA,
ednsflags=321
)
self.agent._send_dns_message = Mock(return_value=response)
out = self.agent._make_and_send_dns_message('h', 123, 1, 2, 3, 4, 5)
self.assertEqual((response, 0), out)
@mock.patch.object(agent.dns_query, 'tcp')
@mock.patch.object(agent.dns_query, 'udp')
def test_send_dns_message(self, *mocks):
mocks[0].return_value = 'mock udp resp'
out = self.agent._send_dns_message('msg', 'host', 123, 1)
assert not agent.dns_query.tcp.called
agent.dns_query.udp.assert_called_with('msg', 'host', port=123,
timeout=1)
self.assertEqual('mock udp resp', out)
@mock.patch.object(agent.dns_query, 'tcp')
@mock.patch.object(agent.dns_query, 'udp')
def test_send_dns_message_timeout(self, *mocks):
mocks[0].side_effect = dns.exception.Timeout
out = self.agent._send_dns_message('msg', 'host', 123, 1)
agent.dns_query.udp.assert_called_with('msg', 'host', port=123,
timeout=1)
assert isinstance(out, dns.exception.Timeout)
@mock.patch.object(agent.dns_query, 'tcp')
@mock.patch.object(agent.dns_query, 'udp')
def test_send_dns_message_bad_response(self, *mocks):
mocks[0].side_effect = agent.dns_query.BadResponse
out = self.agent._send_dns_message('msg', 'host', 123, 1)
agent.dns_query.udp.assert_called_with('msg', 'host', port=123,
timeout=1)
assert isinstance(out, agent.dns_query.BadResponse)
@mock.patch.object(agent.dns_query, 'tcp')
@mock.patch.object(agent.dns_query, 'udp')
def test_send_dns_message_tcp(self, *mocks):
agent.CONF = RoObject({
'service:mdns': RoObject(all_tcp=True)
})
mocks[1].return_value = 'mock tcp resp'
out = self.agent._send_dns_message('msg', 'host', 123, 1)
assert not agent.dns_query.udp.called
agent.dns_query.tcp.assert_called_with('msg', 'host', port=123,
timeout=1)
self.assertEqual('mock tcp resp', out)

123
designate/tests/unit/test_backend/test_designate.py
View File

@ -1,123 +0,0 @@
# Copyright 2015 Hewlett-Packard Development Company, L.P.
#
# Author: Endre Karlson <endre.karlson@hpe.com>
#
# 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.
"""
Unit test Backend
"""
from designateclient import exceptions
from mock import patch
from mock import NonCallableMagicMock
from mock import Mock
from oslo_log import log as logging
import fixtures
import oslotest.base
import testtools
from designate.utils import generate_uuid
from designate import objects
from designate.backend import impl_designate
LOG = logging.getLogger(__name__)
def create_zone():
id_ = generate_uuid()
return objects.Zone(
id=id_,
name='%s-example.com.' % id_,
email='root@example.com',
)
class RoObject(dict):
def __setitem__(self, *a):
raise NotImplementedError
def __setattr__(self, *a):
raise NotImplementedError
def __getattr__(self, k):
return self[k]
class DesignateBackendTest(oslotest.base.BaseTestCase):
def setUp(self):
super(DesignateBackendTest, self).setUp()
opts = RoObject(
username='user',
password='secret',
project_name='project',
project_zone_name='project_zone',
user_zone_name='user_zone'
)
self.target = RoObject({
'id': '4588652b-50e7-46b9-b688-a9bad40a873e',
'type': 'dyndns',
'masters': [RoObject({'host': '192.0.2.1', 'port': 53})],
'options': opts
})
# Backends blow up when trying to self.admin_context = ... due to
# policy not being initialized
self.admin_context = Mock()
self.useFixture(fixtures.MockPatch(
'designate.context.DesignateContext.get_admin_context',
return_value=self.admin_context
))
self.backend = impl_designate.DesignateBackend(self.target)
# Mock client
self.client = NonCallableMagicMock()
zones = NonCallableMagicMock(spec_set=[
'create', 'delete'])
self.client.configure_mock(zones=zones)
def test_create_zone(self):
zone = create_zone()
masters = ["%(host)s:%(port)s" % self.target.masters[0]]
with patch.object(
self.backend, '_get_client', return_value=self.client):
self.backend.create_zone(self.admin_context, zone)
self.client.zones.create.assert_called_once_with(
zone.name, 'SECONDARY', masters=masters)
def test_delete_zone(self):
zone = create_zone()
with patch.object(
self.backend, '_get_client', return_value=self.client):
self.backend.delete_zone(self.admin_context, zone)
self.client.zones.delete.assert_called_once_with(zone.name)
def test_delete_zone_notfound(self):
zone = create_zone()
self.client.delete.side_effect = exceptions.NotFound
with patch.object(
self.backend, '_get_client', return_value=self.client):
self.backend.delete_zone(self.admin_context, zone)
self.client.zones.delete.assert_called_once_with(zone.name)
def test_delete_zone_exc(self):
class Exc(Exception):
pass
zone = create_zone()
self.client.zones.delete.side_effect = Exc()
with testtools.ExpectedException(Exc):
with patch.object(
self.backend, '_get_client', return_value=self.client):
self.backend.delete_zone(self.admin_context, zone)
self.client.zones.delete.assert_called_once_with(zone.name)