openstack
/
designate
Code Issues Proposed changes

Modernized agent tests 

* Moved all tests from test_agent to unit.agent.
* Cleaned up tests and standardized them.

Change-Id: I048980a74129c603af779ca3f0f9a5ddf750f118
This commit is contained in:
Erik Olof Gunnar Andersson 2019-05-25 15:01:29 -07:00
parent b81755750a
commit d60bd5369a
21 changed files with 961 additions and 1122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
designate/tests/test_agent/test_backends/__init__.py
View File

@ -1,26 +0,0 @@
# Copyright 2014 Rackspace Inc.
#
# Author: Tim Simmons <tim.simmons@rackspace.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.
from oslo_config import cfg
from designate.backend import agent_backend
from designate.agent import service
class BackendTestMixin(object):
def get_backend_driver(self):
return agent_backend.get_backend(
cfg.CONF['service:agent'].backend_driver,
agent_service=service.Service())

69
designate/tests/test_agent/test_backends/test_bind9.py
View File

@ -1,69 +0,0 @@
# Copyright 2014 Rackspace Inc.
#
# Author: Tim Simmons <tim.simmons@rackspace.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 mock
import dns.zone
from designate.agent import service
from designate.backend import agent_backend
from designate.tests import TestCase
from designate.tests.test_agent.test_backends import BackendTestMixin
class Bind9AgentBackendTestCase(TestCase, BackendTestMixin):
def setUp(self):
super(Bind9AgentBackendTestCase, self).setUp()
# Use a random port
self.config(port=0, group='service:agent')
self.backend = agent_backend.get_backend('bind9',
agent_service=service.Service())
self.backend.start()
def tearDown(self):
super(Bind9AgentBackendTestCase, self).tearDown()
self.backend.agent_service.stop()
self.backend.stop()
def test_find_zone_serial(self):
self.backend.find_zone_serial('example.org.')
@mock.patch('designate.utils.execute')
@mock.patch(('designate.backend.agent_backend.impl_bind9.Bind9Backend'
'._sync_zone'))
def test_create_zone(self, execute, sync):
zone = self._create_dnspy_zone('example.org')
self.backend.create_zone(zone)
@mock.patch('designate.utils.execute')
@mock.patch(('designate.backend.agent_backend.impl_bind9.Bind9Backend'
'._sync_zone'))
def test_update_zone(self, execute, sync):
zone = self._create_dnspy_zone('example.org')
self.backend.update_zone(zone)
@mock.patch('designate.utils.execute')
@mock.patch(('designate.backend.agent_backend.impl_bind9.Bind9Backend'
'._sync_zone'))
def test_delete_zone(self, execute, sync):
self.backend.delete_zone('example.org.')
# Helper
def _create_dnspy_zone(self, name):
zone_text = ('$ORIGIN %(name)s\n%(name)s 3600 IN SOA %(ns)s '
'email.email.com. 1421777854 3600 600 86400 3600\n%(name)s 3600 IN NS '
'%(ns)s\n') % {'name': name, 'ns': 'ns1.designate.com'}
return dns.zone.from_text(zone_text, check_origin=False)

129
designate/tests/test_agent/test_backends/test_denominator.py
View File

@ -1,129 +0,0 @@
# Copyright 2015 Dyn Inc.
#
# Author: Yasha Bubnov <ybubnov@dyn.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 mock
import dns.zone
from designate.agent import service
from designate.backend import agent_backend
from designate.tests import TestCase
from designate.tests.test_agent.test_backends import BackendTestMixin
class DenominatorAgentBackendTestCase(TestCase, BackendTestMixin):
def setUp(self):
super(DenominatorAgentBackendTestCase, self).setUp()
self.config(port=0, group='service:agent')
self.backend = agent_backend.get_backend('denominator',
agent_service=service.Service())
self.backend.start()
def tearDown(self):
super(DenominatorAgentBackendTestCase, self).tearDown()
self.backend.agent_service.stop()
self.backend.stop()
@mock.patch('designate.utils.execute', return_value=(
'example.org SOA 86400 ns1.designate.com. '
'hostmaster@example.org. 475 3600 600 604800 1800', None))
def test_find_zone_serial(self, execute):
serial = self.backend.find_zone_serial('example.org.')
# Ensure returned right serial number
self.assertEqual(475, serial)
# Ensure called "denominator zone add"
self.assertIn('record', execute.call_args[0])
self.assertIn('get', execute.call_args[0])
@mock.patch('designate.utils.execute', return_value=('', None))
def test_find_zone_serial_fail(self, execute):
serial = self.backend.find_zone_serial('example.org.')
self.assertIsNone(serial)
@mock.patch('designate.utils.execute', return_value=(None, None))
def test_create_zone(self, execute):
zone = self._create_dnspy_zone('example.org.')
self.backend.create_zone(zone)
# Ensure denominator called for each record (except SOA)
# plus one to update zone data
self.assertEqual(len(list(zone.iterate_rdatas())),
execute.call_count)
@mock.patch('designate.utils.execute')
def test_update_zone(self, execute):
# Output from 'designate record list' command
records = ('example.org SOA 86400 ns1.designate.com. '
'hostmaster@example.org. 475 3600 600 604800 1800\n'
'example.org NS 86400 ns1.designator.net.\n'
'example.org NS 86400 ns2.designator.net.\n'
'example.org MX 86400 10 mx1.designator.net.')
# That should force update_zone to delete A and AAAA records
# from the zone and create a new MX record.
execute.return_value = (records, None)
zone = self._create_dnspy_zone('example.org.')
self.backend.update_zone(zone)
# Ensure denominator called to:
# *update zone info
# *fetch list of zone records
# *delete one MX record
# *replace one NS record
# *create two A and two AAAA records
# total: 8 calls
self.assertEqual(8, execute.call_count)
self.backend.denominator = mock.MagicMock
methods = ['update_zone',
'get_records',
'create_record', 'update_record', 'delete_record']
for method in methods:
setattr(self.backend.denominator, method, mock.Mock(
return_value=records))
self.backend.update_zone(zone)
self.assertEqual(1, self.backend.denominator.update_zone.call_count)
self.assertEqual(1, self.backend.denominator.get_records.call_count)
self.assertEqual(4, self.backend.denominator.create_record.call_count)
self.assertEqual(1, self.backend.denominator.update_record.call_count)
self.assertEqual(1, self.backend.denominator.delete_record.call_count)
@mock.patch('designate.utils.execute', return_value=(None, None))
def test_delete_zone(self, execute):
self.backend.delete_zone('example.org.')
# Ensure called 'denominator zone delete'
self.assertEqual(1, execute.call_count)
self.assertIn('zone', execute.call_args[0])
self.assertIn('delete', execute.call_args[0])
# Returns dns.zone test object
def _create_dnspy_zone(self, name):
zone_text = ('$ORIGIN %(name)s\n'
'@ 3600 IN SOA %(ns)s email.%(name)s 1421777854 3600 600 86400 3600\n'
' 3600 IN NS %(ns)s\n'
' 1800 IN A 173.194.123.30\n'
' 1800 IN A 173.194.123.31\n'
's 2400 IN AAAA 2001:db8:cafe::1\n'
's 2400 IN AAAA 2001:db8:cafe::2\n'
% {'name': name, 'ns': 'ns1.designate.net.'})
return dns.zone.from_text(zone_text, check_origin=False)

127
designate/tests/test_agent/test_backends/test_djbdns.py
View File

@ -1,127 +0,0 @@
# Copyright 2016 Hewlett Packard Enterprise Development Company LP
#
# 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.
"""
Test the Djbdns agent backend
These tests *do* rely on creating directories and files or running
executables from the djbdns suite
If djbdns is not available some tests are skipped.
"""
import os
import tempfile
import unittest
import fixtures
import mock
from designate import exceptions
from designate.backend.agent_backend.impl_djbdns import DjbdnsBackend
from designate.tests import TestCase
import designate.backend.agent_backend.impl_djbdns
TINYDNSDATA_PATH = '/usr/bin/tinydns-data'
class DjbdnsAgentBackendSimpleTestCase(TestCase):
def test__check_dirs(self):
DjbdnsBackend._check_dirs('/tmp')
def test__check_dirs_not_found(self):
self.assertRaises(
exceptions.Backend,
DjbdnsBackend._check_dirs,
'/nonexistent_dir_name'
)
class DjbdnsAgentBackendTestCase(TestCase):
def setUp(self):
super(DjbdnsAgentBackendTestCase, self).setUp()
self.CONF.set_override('masters', ['127.0.0.1:5354'], 'service:agent')
tmp_datafiles_dir = tempfile.mkdtemp()
os.mkdir(os.path.join(tmp_datafiles_dir, 'datafiles'))
self.CONF.set_override(
'tinydns_datadir',
tmp_datafiles_dir,
designate.backend.agent_backend.impl_djbdns.CFG_GROUP_NAME,
)
self.useFixture(fixtures.MockPatchObject(
DjbdnsBackend, '_check_dirs'
))
self.backend = DjbdnsBackend('foo')
self.patch_ob(self.backend._resolver, 'query')
def tearDown(self):
super(DjbdnsAgentBackendTestCase, self).tearDown()
def patch_ob(self, *a, **kw):
self.useFixture(fixtures.MockPatchObject(*a, **kw))
@mock.patch('designate.backend.agent_backend.impl_djbdns.os.remove')
@mock.patch('designate.backend.agent_backend.impl_djbdns.execute')
def test__perform_axfr_from_minidns(self, mock_exe, mock_rm):
mock_exe.return_value = (None, None)
self.backend._perform_axfr_from_minidns('foo')
mock_exe.assert_called_once_with(
'tcpclient', '127.0.0.1', '5354', 'axfr-get', 'foo',
os.path.join(self.backend._datafiles_dir, 'foo.zonedata'),
os.path.join(self.backend._datafiles_dir, 'foo.ztmp')
)
def test_delete_zone_no_file(self):
self.patch_ob(self.backend, '_rebuild_data_cdb')
# Should not raise exceptions
self.backend.delete_zone('non_existent_zone_file')
@unittest.skipIf(not os.path.isfile(TINYDNSDATA_PATH),
"tinydns-data not installed")
def test__rebuild_data_cdb_empty(self):
# Check that tinydns-data can be run and the required files are
# generated / renamed as needed
self.CONF.set_override('root_helper', ' ') # disable rootwrap
self.backend._tinydns_cdb_filename = tempfile.mkstemp()[1]
self.backend._rebuild_data_cdb()
assert os.path.isfile(self.backend._tinydns_cdb_filename)
os.remove(self.backend._tinydns_cdb_filename)
@unittest.skipIf(not os.path.isfile(TINYDNSDATA_PATH),
"tinydns-data not installed")
def test__rebuild_data_cdb(self):
# Check that tinydns-data can be run and the required files are
# generated / renamed as needed
self.CONF.set_override('root_helper', ' ') # disable rootwrap
self.backend._tinydns_cdb_filename = tempfile.mkstemp()[1]
fn = os.path.join(self.backend._datafiles_dir, 'example.org.zonedata')
with open(fn, 'w') as f:
f.write(""".example.org::ns1.example.org
+ns1.example.org:127.0.0.1
+www.example.org:127.0.0.1
""")
self.backend._rebuild_data_cdb()
assert os.path.isfile(self.backend._tinydns_cdb_filename)
os.remove(self.backend._tinydns_cdb_filename)

156
designate/tests/test_agent/test_backends/test_gdnsd.py
View File

@ -1,156 +0,0 @@
# Copyright 2016 Hewlett Packard Enterprise Development Company LP
#
# 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.
"""
Functional-test the Gdnsd 2 agent backend
These tests rely on creating directories and files and running
gdnsd.
gdnsd must be installed
"""
from textwrap import dedent as de
import glob
import os
import tempfile
import unittest
from oslo_concurrency.processutils import ProcessExecutionError
from oslo_config import cfg
from oslo_config import fixture as cfg_fixture
import dns
import fixtures
import mock
import testtools
from designate import exceptions
from designate.backend.agent_backend.impl_gdnsd import CFG_GROUP_NAME
from designate.backend.agent_backend.impl_gdnsd import GdnsdBackend
from designate.tests import TestCase
TMPFS_DIR = "/dev/shm"
ROOT_TMP_DIR = TMPFS_DIR if os.path.isdir(TMPFS_DIR) else "/tmp"
GDNSD_BIN_PATH = "/usr/sbin/gdnsd"
GDNSD_NOT_AVAILABLE = not os.path.isfile(GDNSD_BIN_PATH)
ZONE_TPL = """
$ORIGIN %(name)s
%(name)s 3600 IN SOA ns.%(name)s email.%(name)s. 1421777854 3600 600 86400 3600
%(name)s 3600 IN NS ns.%(name)s
ns 300 IN A 127.0.0.1
""" # noqa
class GdnsdAgentBackendTestCase(TestCase):
def setUp(self):
super(GdnsdAgentBackendTestCase, self).setUp()
self.conf_dir_path = tempfile.mkdtemp(dir=ROOT_TMP_DIR)
self.zones_dir_path = os.path.join(self.conf_dir_path, 'zones')
os.mkdir(self.zones_dir_path)
self.CONF = self.useFixture(cfg_fixture.Config(cfg.CONF)).conf
cfg.CONF.set_override('confdir_path', self.conf_dir_path,
CFG_GROUP_NAME)
cfg.CONF.set_override('gdnsd_cmd_name', GDNSD_BIN_PATH,
CFG_GROUP_NAME)
self.backend = GdnsdBackend('foo')
def tearDown(self):
super(GdnsdAgentBackendTestCase, self).tearDown()
for zone_fn in glob.glob(os.path.join(self.zones_dir_path, "*.org")):
os.remove(zone_fn)
os.rmdir(self.zones_dir_path)
os.rmdir(self.conf_dir_path)
def _patch_ob(self, *a, **kw):
self.useFixture(fixtures.MockPatchObject(*a, **kw))
def _create_dnspy_zone(self, name):
name = name.rstrip('.')
zone_text = ZONE_TPL % {'name': name}
return dns.zone.from_text(zone_text, check_origin=False)
def _create_dnspy_zone_with_records(self, name):
zone_text = (
'$ORIGIN %(name)s\n'
'@ 3600 IN SOA %(ns)s email.%(name)s 1421777854 3600 600 86400 3600\n' # noqa
' 3600 IN NS %(ns)s\n'
' 1800 IN A 173.194.123.30\n'
' 1800 IN A 173.194.123.31\n'
's 2400 IN AAAA 2001:db8:cafe::1\n'
's 2400 IN AAAA 2001:db8:cafe::2\n'
% {'name': name, 'ns': 'ns1.example.net.'}
)
return dns.zone.from_text(zone_text, check_origin=False)
@mock.patch('designate.utils.execute', return_value=("", ""))
def test_start(self, *mocks):
self.backend.start()
@mock.patch('designate.utils.execute', side_effect=ProcessExecutionError)
def test_exec_error(self, *mocks):
with testtools.ExpectedException(exceptions.Backend):
self.backend._check_conf()
def test_create_zone(self, *mocks):
zone = self._create_dnspy_zone('example.org')
self.backend.create_zone(zone)
zone_fn = os.path.join(self.zones_dir_path, "example.org")
expected = de("""\
ns 300 IN A 127.0.0.1
@ 3600 IN SOA ns email.example.org. 1421777854 3600 600 86400 3600
@ 3600 IN NS ns
""")
with open(zone_fn) as f:
self.assertEqual(expected, f.read())
@unittest.skipIf(GDNSD_NOT_AVAILABLE, "gdnsd binary not installed")
def test_create_zone_and_check(self):
zone = self._create_dnspy_zone('example.org')
self.backend.create_zone(zone)
self.backend._check_conf()
def test_update_zone(self):
zone = self._create_dnspy_zone_with_records('example.org')
self.backend.update_zone(zone)
zone_fn = os.path.join(self.zones_dir_path, "example.org")
expected = de("""\
@ 3600 IN SOA ns1.example.net. email 1421777854 3600 600 86400 3600
@ 3600 IN NS ns1.example.net.
@ 1800 IN A 173.194.123.30
@ 1800 IN A 173.194.123.31
s 2400 IN AAAA 2001:db8:cafe::1
s 2400 IN AAAA 2001:db8:cafe::2
""") # noqa
with open(zone_fn) as f:
self.assertEqual(expected, f.read())
@unittest.skipIf(GDNSD_NOT_AVAILABLE, "gdnsd binary not installed")
def test_update_zone_and_check(self):
zone = self._create_dnspy_zone_with_records('example.org')
self.backend.update_zone(zone)
self.backend._check_conf()
def test_delete_zone(self):
foo_fn = os.path.join(self.zones_dir_path, 'foo')
with open(foo_fn, 'w') as f:
f.write("42")
self.backend.delete_zone('foo')
self.assertFalse(os.path.isfile(foo_fn))
self.backend.delete_zone('foo')

30
designate/tests/test_agent/test_service.py
View File

@ -1,30 +0,0 @@
# Copyright 2014 Rackspace Inc.
#
# Author: Tim Simmons <tim.simmons@rackspace.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.
from designate.tests.test_agent import AgentTestCase
class AgentServiceTest(AgentTestCase):
def setUp(self):
super(AgentServiceTest, self).setUp()
# Use a random port
self.config(port=0, group='service:agent')
self.service = self.start_service('agent')
def test_stop(self):
# NOTE: Start is already done by the fixture in start_service()
self.service.stop()

0
designate/tests/unit/test_agent/__init__.py → designate/tests/unit/agent/__init__.py
View File

19
designate/tests/test_agent/__init__.py → designate/tests/unit/agent/backends/__init__.py
View File

@ -1,7 +1,3 @@
# Copyright 2014 Rackspace Inc.
#
# Author: Tim Simmons <tim.simmons@rackspace.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
@ -13,8 +9,17 @@
# 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 designate.tests import TestCase
import dns.zone
class AgentTestCase(TestCase):
pass
def create_dnspy_zone(name):
zone_text = (
'$ORIGIN %(name)s\n%(name)s 3600 IN SOA %(ns)s email.email.com. '
'1421777854 3600 600 86400 3600\n%(name)s 3600 IN NS %(ns)s\n'
)
return dns.zone.from_text(
zone_text % {'name': name, 'ns': 'ns1.designate.com'},
check_origin=False
)

124
designate/tests/unit/agent/backends/test_bind9.py Normal file
View File

@ -0,0 +1,124 @@
# Copyright 2014 Rackspace Inc.
#
# Author: Tim Simmons <tim.simmons@rackspace.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.resolver
import mock
import designate.tests
from designate import exceptions
from designate import utils
from designate.backend.agent_backend import impl_bind9
from designate.tests.unit.agent import backends
class Bind9AgentBackendTestCase(designate.tests.TestCase):
def setUp(self):
super(Bind9AgentBackendTestCase, self).setUp()
self.CONF.set_override('port', 0, 'service:agent')
self.backend = impl_bind9.Bind9Backend('foo')
def test_start_backend(self):
self.backend.start()
def test_stop_backend(self):
self.backend.stop()
@mock.patch.object(dns.resolver.Resolver, 'query')
def test_find_zone_serial(self, mock_query):
self.assertIsNotNone(self.backend.find_zone_serial('example.org.'))
@mock.patch.object(dns.resolver.Resolver, 'query')
def test_find_zone_serial_query_raises(self, mock_query):
mock_query.side_effect = Exception()
self.assertIsNone(self.backend.find_zone_serial('example.org.'))
@mock.patch('designate.utils.execute')
@mock.patch('designate.backend.agent_backend.impl_bind9.Bind9Backend'
'._sync_zone')
def test_create_zone(self, mock_execute, mock_sync_zone):
zone = backends.create_dnspy_zone('example.org')
self.backend.create_zone(zone)
@mock.patch('designate.utils.execute')
@mock.patch('designate.backend.agent_backend.impl_bind9.Bind9Backend'
'._sync_zone')
def test_update_zone(self, mock_execute, mock_sync_zone):
zone = backends.create_dnspy_zone('example.org')
self.backend.update_zone(zone)
@mock.patch('designate.utils.execute')
@mock.patch('designate.backend.agent_backend.impl_bind9.Bind9Backend'
'._sync_zone')
def test_delete_zone(self, mock_execute, mock_sync_zone):
self.backend.delete_zone('example.org.')
@mock.patch('designate.utils.execute')
def test_execute_rndc(self, mock_execute):
self.CONF.set_override(
'rndc_config_file', 'config_file', 'backend:agent:bind9'
)
self.CONF.set_override(
'rndc_key_file', 'key_file', 'backend:agent:bind9'
)
self.backend._execute_rndc(self.backend._rndc_base())
mock_execute.assert_called_once_with(
'rndc', '-s', '127.0.0.1', '-p', '953',
'-c', 'config_file', '-k', 'key_file'
)
@mock.patch('designate.utils.execute')
def test_execute_rndc_raises(self, mock_execute):
mock_execute.side_effect = utils.processutils.ProcessExecutionError()
self.assertRaises(
exceptions.Backend,
self.backend._execute_rndc, self.backend._rndc_base()
)
@mock.patch('designate.utils.execute')
@mock.patch.object(dns.zone.Zone, 'to_file')
def test_sync_zone(self, mock_to_file, mock_execute):
zone = backends.create_dnspy_zone('example.org')
self.backend._sync_zone(zone)
mock_to_file.assert_called_once_with(
'/var/lib/designate/zones/example.org.zone', relativize=False
)
mock_execute.assert_called_once_with(
'rndc', '-s', '127.0.0.1', '-p', '953', 'reload', 'example.org'
)
@mock.patch('designate.utils.execute')
@mock.patch.object(dns.zone.Zone, 'to_file')
def test_sync_zone_with_new_zone(self, mock_to_file, mock_execute):
zone = backends.create_dnspy_zone('example.org')
self.backend._sync_zone(zone, new_zone_flag=True)
mock_to_file.assert_called_once_with(
'/var/lib/designate/zones/example.org.zone', relativize=False
)
mock_execute.assert_called_once_with(
'rndc', '-s', '127.0.0.1', '-p', '953', 'addzone',
'example.org { type master; '
'file "/var/lib/designate/zones/example.org.zone"; };'
)

168
designate/tests/unit/agent/backends/test_denominator.py Normal file
View File

@ -0,0 +1,168 @@
# Copyright 2015 Dyn Inc.
#
# Author: Yasha Bubnov <ybubnov@dyn.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 mock
from oslo_config import cfg
import designate.tests
from designate import exceptions
from designate import tests
from designate import utils
from designate.backend.agent_backend import impl_denominator
from designate.tests.unit.agent import backends
class DenominatorAgentBackendTestCase(designate.tests.TestCase):
def setUp(self):
super(DenominatorAgentBackendTestCase, self).setUp()
self.CONF.set_override('port', 0, 'service:agent')
self.backend = impl_denominator.DenominatorBackend('foo')
def test_start_backend(self):
self.backend.start()
def test_stop_backend(self):
self.backend.stop()
@mock.patch('designate.utils.execute', return_value=(
'example.org SOA 86400 ns1.designate.com. '
'hostmaster@example.org. 475 3600 600 604800 1800', None))
def test_find_zone_serial(self, mock_execute):
serial = self.backend.find_zone_serial('example.org.')
# Ensure returned right serial number
self.assertEqual(475, serial)
# Ensure called "denominator zone add"
self.assertIn('record', mock_execute.call_args[0])
self.assertIn('get', mock_execute.call_args[0])
@mock.patch('designate.utils.execute', return_value=('', None))
def test_find_zone_serial_fail(self, mock_execute):
serial = self.backend.find_zone_serial('example.org.')
self.assertIsNone(serial)
@mock.patch('designate.utils.execute', return_value=(None, None))
def test_create_zone(self, mock_execute):
zone = backends.create_dnspy_zone('example.org.')
self.backend.create_zone(zone)
# Ensure denominator called for each record (except SOA)
# plus one to update zone data
self.assertEqual(
mock_execute.call_count, len(list(zone.iterate_rdatas()))
)
@mock.patch('designate.utils.execute')
def test_update_zone(self, mock_execute):
# Output from 'designate record list' command
records = ('example.org SOA 86400 ns1.designate.com. '
'hostmaster@example.org. 475 3600 600 604800 1800\n'
'example.org NS 86400 ns1.designator.net.\n'
'example.org NS 86400 ns2.designator.net.\n'
'example.org MX 86400 10 mx1.designator.net.')
# That should force update_zone to delete A and AAAA records
# from the zone and create a new MX record.
mock_execute.return_value = (records, None)
zone = backends.create_dnspy_zone('example.org.')
self.backend.update_zone(zone)
# Ensure denominator called to:
# *update zone info
# *fetch list of zone records
# *delete one MX record
# *replace one NS record
# *create 0 records
# total: 4 calls
self.assertEqual(4, mock_execute.call_count)
methods = ['update_zone',
'get_records',
'create_record', 'update_record', 'delete_record']
for method in methods:
setattr(self.backend.denominator, method, mock.Mock(
return_value=records))
self.backend.update_zone(zone)
self.assertEqual(1, self.backend.denominator.update_zone.call_count)
self.assertEqual(1, self.backend.denominator.get_records.call_count)
self.assertEqual(0, self.backend.denominator.create_record.call_count)
self.assertEqual(1, self.backend.denominator.update_record.call_count)
self.assertEqual(1, self.backend.denominator.delete_record.call_count)
@mock.patch('designate.utils.execute', return_value=(None, None))
def test_delete_zone(self, mock_execute):
self.backend.delete_zone('example.org.')
# Ensure called 'denominator zone delete'
self.assertEqual(1, mock_execute.call_count)
self.assertIn('zone', mock_execute.call_args[0])
self.assertIn('delete', mock_execute.call_args[0])
class DenominatorAgentBaseTestCase(tests.TestCase):
def setUp(self):
super(DenominatorAgentBaseTestCase, self).setUp()
self.backend = impl_denominator.Denominator(
cfg.CONF['backend:agent:denominator']
)
def test_base(self):
self.assertEqual(
['denominator', '-q', '-n', 'fake', '-C', '/etc/denominator.conf'],
self.backend._base()
)
def test_base_without_config_file(self):
self.CONF.set_override(
'config_file', '', 'backend:agent:denominator'
)
self.assertEqual(
['denominator', '-q', '-n', 'fake'],
self.backend._base()
)
@mock.patch('designate.utils.execute')
def test_execute(self, mock_execute):
mock_execute.return_value = ('stdout', None,)
self.assertEqual(
'stdout',
self.backend._execute(
['record', '-z', 'example.org', 'add'], {'name': 'example.org'}
)
)
mock_execute.assert_called_once_with(
'denominator', '-q', '-n', 'fake', '-C', '/etc/denominator.conf',
'record', '-z', 'example.org', 'add', '--name', 'example.org'
)
@mock.patch('designate.utils.execute')
def test_execute_raises(self, mock_execute):
mock_execute.side_effect = utils.processutils.ProcessExecutionError()
self.assertRaises(
exceptions.DesignateException,
self.backend._execute, ['record', '-z', 'example.org', 'add'], {}
)

125
designate/tests/unit/agent/backends/test_djbdns.py Normal file
View File

@ -0,0 +1,125 @@
# Copyright 2016 Hewlett Packard Enterprise Development Company LP
#
# 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 mock
import designate.tests
from designate import exceptions
from designate.backend.agent_backend import impl_djbdns
from designate.tests.unit.agent import backends
class DjbdnsAgentBackendTestCase(designate.tests.TestCase):
@mock.patch.object(impl_djbdns.DjbdnsBackend, '_check_dirs')
def setUp(self, mock_check_dirs):
super(DjbdnsAgentBackendTestCase, self).setUp()
self.CONF.set_override('masters', ['127.0.0.1:5354'], 'service:agent')
self.backend = impl_djbdns.DjbdnsBackend('foo')
def test_start_backend(self):
self.backend.start()
def test_stop_backend(self):
self.backend.stop()
def test_init(self):
self.assertTrue(hasattr(self.backend, '_resolver'))
self.assertEqual(1, self.backend._resolver.timeout)
self.assertEqual(1, self.backend._resolver.lifetime)
self.assertEqual(['127.0.0.1'], self.backend._resolver.nameservers)
self.assertEqual(
'/var/lib/djbdns/root/data.cdb',
self.backend._tinydns_cdb_filename
)
self.assertEqual(
'/var/lib/djbdns/datafiles',
self.backend._datafiles_dir
)
self.assertEqual(
'/var/lib/djbdns/datafiles/%s.zonedata',
self.backend._datafiles_path_tpl
)
self.assertEqual([('127.0.0.1', 5354)], self.backend._masters)
@mock.patch.object(impl_djbdns.DjbdnsBackend, '_check_dirs')
def test_init_no_masters(self, mock_check_dirs):
self.CONF.set_override('masters', [], 'service:agent')
self.assertRaisesRegex(
exceptions.Backend,
'Missing agent AXFR masters',
impl_djbdns.DjbdnsBackend, 'foo'
)
def test_find_zone_serial(self):
class Data(object):
serial = 3
self.backend._resolver = mock.Mock()
self.backend._resolver.query.return_value = [Data(), ]
serial = self.backend.find_zone_serial('example.com')
self.assertEqual(3, serial)
def test_find_zone_serial_error(self):
self.backend._resolver = mock.Mock()
self.backend._resolver.query.side_effect = RuntimeError('foo')
serial = self.backend.find_zone_serial('example.com')
self.assertIsNone(serial)
@mock.patch('designate.backend.agent_backend.impl_djbdns.execute')
def test_create_zone(self, mock_execute):
self.backend._perform_axfr_from_minidns = mock.Mock()
self.backend._rebuild_data_cdb = mock.Mock()
zone = backends.create_dnspy_zone('example.org')
self.backend.create_zone(zone)
def test_update_zone(self):
self.backend._perform_axfr_from_minidns = mock.Mock()
self.backend._rebuild_data_cdb = mock.Mock()
zone = backends.create_dnspy_zone('example.org')
self.backend.update_zone(zone)
@mock.patch('designate.backend.agent_backend.impl_djbdns.os.remove')
def test_delete_zone(self, mock_rm):
self.backend._rebuild_data_cdb = mock.Mock()
self.backend.delete_zone('foo')
mock_rm.assert_called_once_with(
'/var/lib/djbdns/datafiles/foo.zonedata'
)
@mock.patch('designate.backend.agent_backend.impl_djbdns.os.remove')
def test_exception_filter(self, mock_os_remove):
self.backend._rebuild_data_cdb = mock.Mock()
self.assertRaises(
exceptions.Backend,
self.backend.delete_zone, None
)
@mock.patch('designate.backend.agent_backend.impl_djbdns.os.remove')
def test_exception_filter_pass_through(self, mock_os_remove):
self.backend._rebuild_data_cdb = mock.Mock()
mock_os_remove.side_effect = exceptions.Backend
self.assertRaises(
exceptions.Backend,
self.backend.delete_zone, 'foo'
)

38
designate/tests/test_agent/test_backends/test_fake.py → designate/tests/unit/agent/backends/test_fake.py
View File

@ -13,47 +13,35 @@
# 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.zone
from designate.agent import service
from designate.backend import agent_backend
from designate.tests import TestCase
from designate.tests.test_agent.test_backends import BackendTestMixin
import designate.tests
from designate.backend.agent_backend import impl_fake
from designate.tests.unit.agent import backends
class FakeAgentBackendTestCase(TestCase, BackendTestMixin):
class FakeAgentBackendTestCase(designate.tests.TestCase):
def setUp(self):
super(FakeAgentBackendTestCase, self).setUp()
# Use a random port
self.config(port=0, group='service:agent')
self.backend = agent_backend.get_backend('fake',
agent_service=service.Service())
self.CONF.set_override('port', 0, 'service:agent')
self.backend = impl_fake.FakeBackend('foo')
def test_start_backend(self):
self.backend.start()
def tearDown(self):
super(FakeAgentBackendTestCase, self).tearDown()
self.backend.agent_service.stop()
def test_stop_backend(self):
self.backend.stop()
def test_find_zone_serial(self):
self.backend.find_zone_serial('example.org.')
def test_create_zone(self):
zone = self._create_dnspy_zone('example.org')
zone = backends.create_dnspy_zone('example.org')
self.backend.create_zone(zone)
def test_update_zone(self):
zone = self._create_dnspy_zone('example.org')
zone = backends.create_dnspy_zone('example.org')
self.backend.update_zone(zone)
def test_delete_zone(self):
self.backend.delete_zone('example.org.')
# Helper
def _create_dnspy_zone(self, name):
zone_text = ('$ORIGIN %(name)s\n%(name)s 3600 IN SOA %(ns)s '
'email.email.com. 1421777854 3600 600 86400 3600\n%(name)s 3600 IN NS '
'%(ns)s\n') % {'name': name, 'ns': 'ns1.designate.com'}
return dns.zone.from_text(zone_text, check_origin=False)

62
designate/tests/unit/test_agent/test_backends/test_gdnsd.py → designate/tests/unit/agent/backends/test_gdnsd.py
View File

@ -13,66 +13,52 @@
# 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 the Gdnsd 2 agent backend
These tests do not rely on creating directories and files or running
executables from the gdnsd suite
"""
import dns.zone
import fixtures
import mock
from designate.backend.agent_backend.impl_gdnsd import GdnsdBackend
from designate.tests import TestCase
import designate.backend.agent_backend.impl_gdnsd # noqa
import designate.tests
from designate.backend.agent_backend import impl_gdnsd
class GdnsdAgentBackendUnitTestCase(TestCase):
class GdnsdAgentBackendTestCase(designate.tests.TestCase):
@mock.patch.object(impl_gdnsd.GdnsdBackend, '_check_dirs')
def setUp(self, mock_check_dirs):
super(GdnsdAgentBackendTestCase, self).setUp()
def setUp(self):
super(GdnsdAgentBackendUnitTestCase, self).setUp()
self.useFixture(fixtures.MockPatchObject(
GdnsdBackend, '_check_dirs'
))
self.backend = GdnsdBackend('foo')
self.useFixture(fixtures.MockPatchObject(self.backend._resolver,
'query'))
self.backend = impl_gdnsd.GdnsdBackend('foo')
def tearDown(self):
super(GdnsdAgentBackendUnitTestCase, self).tearDown()
@mock.patch.object(impl_gdnsd.GdnsdBackend, '_check_conf')
def test_start_backend(self, mock_check_conf):
self.backend.start()
self.assertTrue(mock_check_conf.called)
def _create_dnspy_zone(self, name):
zone_text = (
'$ORIGIN %(name)s\n%(name)s 3600 IN SOA %(ns)s '
'email.email.com. 1421777854 3600 600 86400 3600\n%(name)s '
'3600 IN NS %(ns)s\n') % {'name': name, 'ns': 'ns1.designate.com'}
return dns.zone.from_text(zone_text, check_origin=False)
def test_stop_backend(self):
self.backend.stop()
def test_init(self):
self.assertEqual(1, self.backend._resolver.timeout)
self.assertEqual(1, self.backend._resolver.lifetime)
self.assertEqual(['127.0.0.1'], self.backend._resolver.nameservers)
self.assertEqual('/etc/gdnsd/zones',
self.backend._zonedir_path)
self.assertEqual(
'/etc/gdnsd/zones',
self.backend._zonedir_path
)
self.assertEqual('gdnsd', self.backend._gdnsd_cmd_name)
def test__generate_zone_filename(self):
fn = self.backend._generate_zone_filename("A/bc-d_e.f")
self.assertEqual("a@bc-d_e.f", fn)
def test_generate_zone_filename(self):
zone_filename = self.backend._generate_zone_filename('A/bc-d_e.f')
self.assertEqual('a@bc-d_e.f', zone_filename)
def test_find_zone_serial(self):
class Data(object):
serial = 3
self.backend._resolver = mock.Mock()
self.backend._resolver.query.return_value = [Data(), ]
serial = self.backend.find_zone_serial('example.com')
self.assertEqual(3, serial)
def test_find_zone_serial_error(self):
self.backend._resolver = mock.Mock()
self.backend._resolver.query.side_effect = RuntimeError('foo')
serial = self.backend.find_zone_serial('example.com')
@ -81,5 +67,7 @@ class GdnsdAgentBackendUnitTestCase(TestCase):
@mock.patch('designate.backend.agent_backend.impl_gdnsd.os.remove')
def test_delete_zone(self, mock_osremove):
self.backend.delete_zone('foo-bar.example.org.')
mock_osremove.assert_called_once_with(
"/etc/gdnsd/zones/foo-bar.example.org")
'/etc/gdnsd/zones/foo-bar.example.org'
)

225
designate/tests/unit/agent/backends/test_knot2.py Normal file
View File

@ -0,0 +1,225 @@
# Copyright 2016 Hewlett Packard Enterprise Development Company LP
#
# 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 mock
from mock import call
from oslo_concurrency import processutils
import designate.tests
from designate import exceptions
from designate.backend.agent_backend import impl_knot2
from designate.tests.unit.agent import backends
class Knot2AgentBackendTestCase(designate.tests.TestCase):
def setUp(self):
super(Knot2AgentBackendTestCase, self).setUp()
self.backend = impl_knot2.Knot2Backend('foo')
self.backend._execute_knotc = mock.Mock()
def test_start_backend(self):
self.backend.start()
def test_stop_backend(self):
self.backend.stop()
def test_create_zone(self):
zone = backends.create_dnspy_zone('example.org')
self.backend.create_zone(zone)
self.backend._execute_knotc.assert_has_calls([
call('conf-begin'),
call('conf-set', 'zone[example.org]',
expected_error='duplicate identifier'),
call('conf-commit'),
call('zone-refresh', 'example.org')
])
def test_create_zone_already_there(self):
self.backend._execute_knotc.return_value = 'duplicate identifier'
zone = backends.create_dnspy_zone('example.org')
self.backend.create_zone(zone)
self.backend._execute_knotc.assert_has_calls([
call('conf-begin'),
call('conf-set', 'zone[example.org]',
expected_error='duplicate identifier'),
call('conf-commit'),
call('zone-refresh', 'example.org')
])
def test_start_minidns_to_knot_axfr(self):
self.backend._start_minidns_to_knot_axfr('foo')
self.backend._execute_knotc.assert_called_with('zone-refresh', 'foo')
@mock.patch('oslo_concurrency.lockutils.lock')
def test_modify_zone(self, mock_lock):
self.backend._modify_zone('blah', 'bar')
self.assertEqual(3, self.backend._execute_knotc.call_count)
self.backend._execute_knotc.assert_called_with('conf-commit')
@mock.patch('oslo_concurrency.lockutils.lock')
def test_modify_zone_exception(self, mock_lock):
# Raise an exception during the second call to _execute_knotc
self.backend._execute_knotc.side_effect = [None, exceptions.Backend,
None]
self.assertRaises(
exceptions.Backend,
self.backend._modify_zone, 'blah', 'bar'
)
self.assertEqual(3, self.backend._execute_knotc.call_count)
self.backend._execute_knotc.assert_has_calls([
call('conf-begin'),
call('blah', 'bar'),
call('conf-abort'),
])
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test_find_zone_serial(self, mock_execute):
result = (
'[example.com.] type: slave | serial: 20 | next-event: idle | '
'auto-dnssec: disabled]'
)
mock_execute.return_value = result, ''
serial = self.backend.find_zone_serial('example.com')
self.assertEqual(20, serial)
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test_find_zone_serial_zone_not_found(self, mock_execute):
mock_execute.side_effect = processutils.ProcessExecutionError(
'error: [example.com.] (no such zone found)'
)
serial = self.backend.find_zone_serial('example.com')
self.assertIsNone(serial)
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test_find_zone_serial_unexpected_output(self, mock_execute):
mock_execute.return_value = 'bogus output', ''
self.assertRaises(
exceptions.Backend,
self.backend.find_zone_serial, 'example.com'
)
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test_find_zone_serial_error(self, mock_execute):
mock_execute.side_effect = processutils.ProcessExecutionError('blah')
self.assertRaises(
exceptions.Backend,
self.backend.find_zone_serial, 'example.com'
)
def test_update_zone(self):
zone = backends.create_dnspy_zone('example.org')
self.backend.update_zone(zone)
self.backend._execute_knotc.assert_called_once_with(
'zone-refresh', 'example.org'
)
def test_delete_zone(self):
self.backend.delete_zone('example.org')
self.backend._execute_knotc.assert_has_calls([
call('conf-begin'),
call('conf-unset', 'zone[example.org]',
expected_error='invalid identifier'),
call('conf-commit'),
])
def test_delete_zone_already_gone(self):
self.backend._execute_knotc.return_value = 'duplicate identifier'
self.backend.delete_zone('example.org')
self.backend._execute_knotc.assert_has_calls([
call('conf-begin'),
call('conf-unset', 'zone[example.org]',
expected_error='invalid identifier'),
call('conf-commit'),
])
class Knot2AgentExecuteTestCase(designate.tests.TestCase):
def setUp(self):
super(Knot2AgentExecuteTestCase, self).setUp()
self.backend = impl_knot2.Knot2Backend('foo')
def test_init(self):
self.assertEqual('knotc', self.backend._knotc_cmd_name)
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test_execute_knotc_ok(self, mock_execute):
mock_execute.return_value = ('OK', '')
self.backend._execute_knotc('a1', 'a2')
mock_execute.assert_called_with('knotc', 'a1', 'a2')
self.assertEqual(1, mock_execute.call_count)
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test_execute_knotc_expected_error(self, mock_execute):
mock_execute.return_value = ('xyz', '')
self.backend._execute_knotc('a1', 'a2', expected_error='xyz')
mock_execute.assert_called_once_with('knotc', 'a1', 'a2')
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test_execute_knotc_expected_output(self, mock_execute):
mock_execute.return_value = ('xyz', '')
self.backend._execute_knotc('a1', 'a2', expected_output='xyz')
mock_execute.assert_called_once_with('knotc', 'a1', 'a2')
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test_execute_knotc_with_error(self, mock_execute):
mock_execute.return_value = ('xyz', '')
self.assertRaises(
exceptions.Backend,
self.backend._execute_knotc, 'a1', 'a2'
)
mock_execute.assert_called_once_with('knotc', 'a1', 'a2')
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test_execute_knotc_raising_exception(self, mock_execute):
mock_execute.side_effect = processutils.ProcessExecutionError
self.assertRaises(
exceptions.Backend,
self.backend._execute_knotc, 'a1', 'a2'
)
mock_execute.assert_called_once_with('knotc', 'a1', 'a2')

140
designate/tests/unit/agent/backends/test_msdns.py Normal file
View File

@ -0,0 +1,140 @@
# Copyright 2016 Cloudbase Solutions Srl
# All Rights Reserved.
#
# Author: Alin Balutoiu <abalutoiu@cloudbasesolutions.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 mock
from os_win import constants
from os_win import exceptions as os_win_exc
from os_win import utilsfactory
import designate.tests
from designate import exceptions
from designate.backend.agent_backend import impl_msdns
from designate.tests.unit.agent import backends
class MSDNSAgentBackendTestCase(designate.tests.TestCase):
@mock.patch.object(utilsfactory, 'get_dnsutils')
def setUp(self, mock_get_dnsutils):
super(MSDNSAgentBackendTestCase, self).setUp()
self.zone_name = 'example.com'
self.CONF.set_override('masters', ['127.0.0.1:5354'], 'service:agent')
self.backend = impl_msdns.MSDNSBackend('foo')
self.backend._dnsutils = mock.MagicMock()
def test_start_backend(self):
self.backend.start()
def test_stop_backend(self):
self.backend.stop()
def test_init(self):
self.assertEqual(['127.0.0.1'], self.backend._masters)
@mock.patch.object(utilsfactory, 'get_dnsutils')
def test_init_no_masters(self, mock_get_dnsutils):
self.CONF.set_override('masters', [], 'service:agent')
self.assertRaisesRegex(
exceptions.Backend,
'Missing agent AXFR masters',
impl_msdns.MSDNSBackend, 'foo'
)
def test_find_zone_serial(self):
serial = self.backend.find_zone_serial(self.zone_name)
expected_serial = self.backend._dnsutils.get_zone_serial.return_value
self.assertEqual(expected_serial, serial)
self.backend._dnsutils.get_zone_serial.assert_called_once_with(
self.zone_name
)
def test_find_zone_serial_error(self):
self.backend._dnsutils.get_zone_serial.side_effect = (
os_win_exc.DNSZoneNotFound(zone_name=self.zone_name))
serial = self.backend.find_zone_serial(self.zone_name)
self.assertIsNone(serial)
self.backend._dnsutils.get_zone_serial.assert_called_once_with(
self.zone_name
)
def test_create_zone(self):
zone = backends.create_dnspy_zone(self.zone_name)
self.backend.create_zone(zone)
self.backend._dnsutils.zone_create.assert_called_once_with(
zone_name=self.zone_name,
zone_type=constants.DNS_ZONE_TYPE_SECONDARY,
ds_integrated=False,
ip_addrs=self.backend._masters
)
def test_create_zone_already_existing_diff(self):
zone = backends.create_dnspy_zone(self.zone_name)
self.backend._dnsutils.zone_create.side_effect = (
os_win_exc.DNSZoneAlreadyExists(zone_name=self.zone_name))
self.assertRaises(
os_win_exc.DNSZoneAlreadyExists,
self.backend.create_zone, zone
)
self.backend._dnsutils.get_zone_properties.assert_called_once_with(
self.zone_name
)
def test_create_zone_already_existing_identical(self):
zone = backends.create_dnspy_zone(self.zone_name)
self.backend._dnsutils.zone_create.side_effect = (
os_win_exc.DNSZoneAlreadyExists(zone_name=self.zone_name)
)
mock_zone_properties = {
'zone_type': constants.DNS_ZONE_TYPE_SECONDARY,
'ds_integrated': False,
'master_servers': self.backend._masters
}
self.backend._dnsutils.get_zone_properties.return_value = (
mock_zone_properties
)
self.backend.create_zone(zone)
self.backend._dnsutils.get_zone_properties.assert_called_once_with(
self.zone_name
)
def test_update_zone(self):
zone = backends.create_dnspy_zone(self.zone_name)
self.backend.update_zone(zone)
self.backend._dnsutils.zone_update.assert_called_once_with(
self.zone_name
)
def test_delete_zone(self):
self.backend.delete_zone(self.zone_name)
self.backend._dnsutils.zone_delete.assert_called_once_with(
self.zone_name
)

129
designate/tests/test_agent/test_handler.py → designate/tests/unit/agent/test_handler.py
View File

@ -20,29 +20,44 @@ import dns.resolver
import mock
import designate
import designate.tests
from designate.agent import handler
from designate.tests.test_agent import AgentTestCase
class AgentRequestHandlerTest(AgentTestCase):
class AgentRequestHandlerTest(designate.tests.TestCase):
def setUp(self):
super(AgentRequestHandlerTest, self).setUp()
self.config(allow_notify=["0.0.0.0"],
backend_driver="fake",
transfer_source="1.2.3.4",
group='service:agent')
self.CONF.set_override('allow_notify', ['0.0.0.0'], 'service:agent')
self.CONF.set_override('backend_driver', 'fake', 'service:agent')
self.CONF.set_override('transfer_source', '1.2.3.4', 'service:agent')
self.handler = handler.RequestHandler()
self.addr = ["0.0.0.0", 5558]
self.addr = ['0.0.0.0', 5558]
def test_init(self):
self.CONF.set_override('masters', ['192.168.0.1', '192.168.0.2'],
'service:agent')
hndlr = handler.RequestHandler()
self.assertEqual(
[
{'host': '192.168.0.1', 'port': 53},
{'host': '192.168.0.2', 'port': 53}
],
hndlr.masters
)
@mock.patch.object(dns.resolver.Resolver, 'query')
@mock.patch('designate.dnsutils.do_axfr')
def test_receive_notify(self, doaxfr, query):
def test_receive_notify(self, mock_doaxfr, mock_query):
"""
Get a NOTIFY and ensure the response is right,
and an AXFR is triggered
"""
payload = ("1a7220000001000000000000076578616d706c6503636f6d000006"
"0001")
payload = ('1a7220000001000000000000076578616d706c6503636f6d000006'
'0001')
# expected response is NOERROR, other fields are
# opcode NOTIFY
# rcode NOERROR
@ -52,18 +67,15 @@ class AgentRequestHandlerTest(AgentTestCase):
# ;ANSWER
# ;AUTHORITY
# ;ADDITIONAL
expected_response = (b"1a72a4000001000000000000076578616d706c6503"
b"636f6d0000060001")
expected_response = (b'1a72a4000001000000000000076578616d706c6503'
b'636f6d0000060001')
request = dns.message.from_wire(binascii.a2b_hex(payload))
request.environ = {'addr': ["0.0.0.0", 1234]}
request.environ = {'addr': ['0.0.0.0', 1234]}
response = next(self.handler(request)).to_wire()
self.assertEqual(expected_response, binascii.b2a_hex(response))
def test_receive_notify_bad_notifier(self):
"""
Get a NOTIFY from a bad master and refuse it
"""
payload = "243520000001000000000000076578616d706c6503636f6d0000060001"
payload = '243520000001000000000000076578616d706c6503636f6d0000060001'
# expected response is REFUSED, other fields are
# opcode NOTIFY
# rcode REFUSED
@ -73,24 +85,19 @@ class AgentRequestHandlerTest(AgentTestCase):
# ;ANSWER
# ;AUTHORITY
# ;ADDITIONAL
expected_response = (b"2435a0050001000000000000076578616d706c6503636f"
b"6d0000060001")
expected_response = (b'2435a0050001000000000000076578616d706c6503636f'
b'6d0000060001')
request = dns.message.from_wire(binascii.a2b_hex(payload))
# Bad 'requester'
request.environ = {'addr': ["6.6.6.6", 1234]}
request.environ = {'addr': ['6.6.6.6', 1234]}
response = next(self.handler(request)).to_wire()
self.assertEqual(expected_response, binascii.b2a_hex(response))
@mock.patch.object(dns.resolver.Resolver, 'query')
@mock.patch('designate.dnsutils.do_axfr')
def test_receive_create(self, doaxfr, query):
"""
Get a CREATE and ensure the response is right,
and an AXFR is triggered, and the proper backend
call is made
"""
payload = "735d70000001000000000000076578616d706c6503636f6d00ff02ff00"
def test_receive_create(self, mock_doaxfr, mock_query):
payload = '735d70000001000000000000076578616d706c6503636f6d00ff02ff00'
# Expected NOERROR other fields are
# opcode 14
# rcode NOERROR
@ -100,22 +107,18 @@ class AgentRequestHandlerTest(AgentTestCase):
# ;ANSWER
# ;AUTHORITY
# ;ADDITIONAL
expected_response = (b"735df4000001000000000000076578616d706c6503636f"
b"6d00ff02ff00")
expected_response = (b'735df4000001000000000000076578616d706c6503636f'
b'6d00ff02ff00')
request = dns.message.from_wire(binascii.a2b_hex(payload))
request.environ = {'addr': ["0.0.0.0", 1234]}
request.environ = {'addr': ['0.0.0.0', 1234]}
with mock.patch.object(
designate.backend.agent_backend.impl_fake.FakeBackend,
designate.backend.agent_backend.impl_fake.FakeBackend,
'find_zone_serial', return_value=None):
response = next(self.handler(request)).to_wire()
self.assertEqual(expected_response, binascii.b2a_hex(response))
def test_receive_create_bad_notifier(self):
"""
Get a NOTIFY from a bad master and refuse it
"""
payload = "8dfd70000001000000000000076578616d706c6503636f6d00ff02ff00"
payload = '8dfd70000001000000000000076578616d706c6503636f6d00ff02ff00'
# expected response is REFUSED, other fields are
# opcode 14
# rcode REFUSED
@ -125,22 +128,18 @@ class AgentRequestHandlerTest(AgentTestCase):
# ;ANSWER
# ;AUTHORITY
# ;ADDITIONAL
expected_response = (b"8dfdf0050001000000000000076578616d706c6503636f"
b"6d00ff02ff00")
expected_response = (b'8dfdf0050001000000000000076578616d706c6503636f'
b'6d00ff02ff00')
request = dns.message.from_wire(binascii.a2b_hex(payload))
# Bad 'requester'
request.environ = {'addr': ["6.6.6.6", 1234]}
request.environ = {'addr': ['6.6.6.6', 1234]}
response = next(self.handler(request)).to_wire()
self.assertEqual(expected_response, binascii.b2a_hex(response))
self.assertEqual(binascii.b2a_hex(response), expected_response)
@mock.patch('designate.utils.execute')
def test_receive_delete(self, func):
"""
Get a DELETE and ensure the response is right,
and that the proper backend call is made
"""
payload = "3b9970000001000000000000076578616d706c6503636f6d00ff03ff00"
def test_receive_delete(self, mock_execute):
payload = '3b9970000001000000000000076578616d706c6503636f6d00ff03ff00'
# Expected NOERROR other fields are
# opcode 14
# rcode NOERROR
@ -150,20 +149,16 @@ class AgentRequestHandlerTest(AgentTestCase):
# ;ANSWER
# ;AUTHORITY
# ;ADDITIONAL
expected_response = (b"3b99f4000001000000000000076578616d706c6503636f"
b"6d00ff03ff00")
expected_response = (b'3b99f4000001000000000000076578616d706c6503636f'
b'6d00ff03ff00')
request = dns.message.from_wire(binascii.a2b_hex(payload))
request.environ = {'addr': ["0.0.0.0", 1234]}
request.environ = {'addr': ['0.0.0.0', 1234]}
response = next(self.handler(request)).to_wire()
self.assertEqual(expected_response, binascii.b2a_hex(response))
def test_receive_delete_bad_notifier(self):
"""
Get a message with an unsupported OPCODE and make
sure that it is refused
"""
payload = "e6da70000001000000000000076578616d706c6503636f6d00ff03ff00"
payload = 'e6da70000001000000000000076578616d706c6503636f6d00ff03ff00'
# expected response is REFUSED, other fields are
# opcode 14
# rcode REFUSED
@ -173,23 +168,19 @@ class AgentRequestHandlerTest(AgentTestCase):
# ;ANSWER
# ;AUTHORITY
# ;ADDITIONAL
expected_response = (b"e6daf0050001000000000000076578616d706c6503636f"
b"6d00ff03ff00")
expected_response = (b'e6daf0050001000000000000076578616d706c6503636f'
b'6d00ff03ff00')
request = dns.message.from_wire(binascii.a2b_hex(payload))
# Bad 'requester'
request.environ = {'addr': ["6.6.6.6", 1234]}
request.environ = {'addr': ['6.6.6.6', 1234]}
response = next(self.handler(request)).to_wire()
self.assertEqual(expected_response, binascii.b2a_hex(response))
@mock.patch.object(dns.resolver.Resolver, 'query')
@mock.patch.object(designate.dnsutils, 'do_axfr')
def test_transfer_source(self, doaxfr, query):
"""
Get a CREATE and ensure the response is right,
and an AXFR is triggered with the right source IP
"""
payload = "735d70000001000000000000076578616d706c6503636f6d00ff02ff00"
def test_transfer_source(self, mock_doaxfr, mock_query):
payload = '735d70000001000000000000076578616d706c6503636f6d00ff02ff00'
# Expected NOERROR other fields are
# opcode 14
# rcode NOERROR
@ -199,13 +190,15 @@ class AgentRequestHandlerTest(AgentTestCase):
# ;ANSWER
# ;AUTHORITY
# ;ADDITIONAL
expected_response = (b"735df4000001000000000000076578616d706c6503636f"
b"6d00ff02ff00")
expected_response = (b'735df4000001000000000000076578616d706c6503636f'
b'6d00ff02ff00')
request = dns.message.from_wire(binascii.a2b_hex(payload))
request.environ = {'addr': ["0.0.0.0", 1234]}
request.environ = {'addr': ['0.0.0.0', 1234]}
with mock.patch.object(
designate.backend.agent_backend.impl_fake.FakeBackend,
designate.backend.agent_backend.impl_fake.FakeBackend,
'find_zone_serial', return_value=None):
response = next(self.handler(request)).to_wire()
doaxfr.assert_called_with('example.com.', [], source="1.2.3.4")
mock_doaxfr.assert_called_with(
'example.com.', [], source='1.2.3.4'
)
self.assertEqual(expected_response, binascii.b2a_hex(response))

68
designate/tests/unit/agent/test_service.py Normal file
View File

@ -0,0 +1,68 @@
# Copyright 2014 Rackspace Inc.
#
# Author: Tim Simmons <tim.simmons@rackspace.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 mock
import designate.tests
from designate import dnsutils
from designate import utils
from designate.agent import service
from designate.backend import agent_backend
from designate.backend.agent_backend import impl_fake
class AgentServiceTest(designate.tests.TestCase):
def setUp(self):
super(AgentServiceTest, self).setUp()
self.CONF.set_override('port', 0, 'service:agent')
self.CONF.set_override('notify_delay', 0, 'service:agent')
self.service = service.Service()
self.service._start = mock.Mock()
self.service._rpc_server = mock.Mock()
def test_service_name(self):
self.assertEqual('agent', self.service.service_name)
def test_start(self):
self.service.start()
self.assertTrue(self.service._start.called)
def test_stop(self):
self.service.stop()
def test_get_backend(self):
backend = agent_backend.get_backend('fake', agent_service=self.service)
self.assertIsInstance(backend, impl_fake.FakeBackend)
@mock.patch.object(utils, 'cache_result')
def test_get_dns_application(self, mock_cache_result):
self.assertIsInstance(
self.service._dns_application, dnsutils.SerializationMiddleware
)
@mock.patch.object(utils, 'cache_result')
def test_get_dns_application_with_notify_delay(self, mock_cache_result):
self.service = service.Service()
self.service._start = mock.Mock()
self.service._rpc_server = mock.Mock()
self.CONF.set_override('notify_delay', 1.0, 'service:agent')
self.assertIsInstance(
self.service._dns_application, dnsutils.SerializationMiddleware
)

0
designate/tests/unit/test_agent/test_backends/__init__.py
View File

126
designate/tests/unit/test_agent/test_backends/test_djbdns.py
View File

@ -1,126 +0,0 @@
# Copyright 2016 Hewlett Packard Enterprise Development Company LP
#
# 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 the Djbdns agent backend
These tests do not rely on creating directories and files or running
executables from the djbdns suite
"""
import dns.zone
import fixtures
import mock
from designate import exceptions
from designate.backend.agent_backend.impl_djbdns import DjbdnsBackend
from designate.tests import TestCase
import designate.backend.agent_backend.impl_djbdns # noqa
class DjbdnsAgentBackendUnitTestCase(TestCase):
def setUp(self):
super(DjbdnsAgentBackendUnitTestCase, self).setUp()
self.CONF.set_override('masters', ['127.0.0.1:5354'], 'service:agent')
self.useFixture(fixtures.MockPatchObject(
DjbdnsBackend, '_check_dirs'
))
self.backend = DjbdnsBackend('foo')
self.patch_ob(self.backend._resolver, 'query')
def tearDown(self):
super(DjbdnsAgentBackendUnitTestCase, self).tearDown()
def _create_dnspy_zone(self, name):
zone_text = (
'$ORIGIN %(name)s\n%(name)s 3600 IN SOA %(ns)s '
'email.email.com. 1421777854 3600 600 86400 3600\n%(name)s '
'3600 IN NS %(ns)s\n') % {'name': name, 'ns': 'ns1.designate.com'}
return dns.zone.from_text(zone_text, check_origin=False)
def patch_ob(self, *a, **kw):
self.useFixture(fixtures.MockPatchObject(*a, **kw))
def test_init(self):
self.assertTrue(hasattr(self.backend, '_resolver'))
self.assertEqual(1, self.backend._resolver.timeout)
self.assertEqual(1, self.backend._resolver.lifetime)
self.assertEqual(['127.0.0.1'], self.backend._resolver.nameservers)
self.assertEqual('/var/lib/djbdns/root/data.cdb',
self.backend._tinydns_cdb_filename)
self.assertEqual('/var/lib/djbdns/datafiles',
self.backend._datafiles_dir)
self.assertEqual('/var/lib/djbdns/datafiles/%s.zonedata',
self.backend._datafiles_path_tpl)
self.assertEqual([('127.0.0.1', 5354)], self.backend._masters)
def test_find_zone_serial(self):
class Data(object):
serial = 3
self.backend._resolver.query.return_value = [Data(), ]
serial = self.backend.find_zone_serial('example.com')
self.assertEqual(3, serial)
def test_find_zone_serial_error(self):
self.backend._resolver.query.side_effect = RuntimeError('foo')
serial = self.backend.find_zone_serial('example.com')
self.assertIsNone(serial)
@mock.patch('designate.backend.agent_backend.impl_djbdns.execute')
def test_create_zone(self, mock_exe):
self.patch_ob(self.backend, '_perform_axfr_from_minidns')
self.patch_ob(self.backend, '_rebuild_data_cdb')
zone = self._create_dnspy_zone('example.org')
self.backend.create_zone(zone)
def test_update_zone(self):
self.patch_ob(self.backend, '_perform_axfr_from_minidns')
self.patch_ob(self.backend, '_rebuild_data_cdb')
zone = self._create_dnspy_zone('example.org')
self.backend.update_zone(zone)
@mock.patch('designate.backend.agent_backend.impl_djbdns.os.remove')
def test_delete_zone(self, mock_rm):
self.patch_ob(self.backend, '_rebuild_data_cdb')
self.backend.delete_zone('foo')
mock_rm.assert_called_once_with(
'/var/lib/djbdns/datafiles/foo.zonedata'
)
@mock.patch('designate.backend.agent_backend.impl_djbdns.os.remove')
def test_exception_filter(self, *mocks):
self.patch_ob(self.backend, '_rebuild_data_cdb')
self.assertRaises(
exceptions.Backend,
self.backend.delete_zone,
None
)
@mock.patch('designate.backend.agent_backend.impl_djbdns.os.remove')
def test_exception_filter_pass_through(self, mock_rm):
self.patch_ob(self.backend, '_rebuild_data_cdb')
mock_rm.side_effect = exceptions.Backend
self.assertRaises(
exceptions.Backend,
self.backend.delete_zone,
'foo'
)

200
designate/tests/unit/test_agent/test_backends/test_knot2.py
View File

@ -1,200 +0,0 @@
# Copyright 2016 Hewlett Packard Enterprise Development Company LP
#
# 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 the Knot 2 agent backend
knotc is not being executed
"""
from mock import call
from oslo_concurrency.processutils import ProcessExecutionError
import dns.zone
import fixtures
import mock
from designate import exceptions
from designate.backend.agent_backend.impl_knot2 import Knot2Backend
from designate.tests import TestCase
import designate.backend.agent_backend.impl_knot2 # noqa
class Knot2AgentBackendBasicUnitTestCase(TestCase):
def test_init(self):
kb = Knot2Backend('foo')
self.assertEqual('knotc', kb._knotc_cmd_name)
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test__execute_knotc_ok(self, mock_exe):
mock_exe.return_value = ('OK', '')
kb = Knot2Backend('foo')
kb._execute_knotc('a1', 'a2')
mock_exe.assert_called_with('knotc', 'a1', 'a2')
self.assertEqual(1, mock_exe.call_count)
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test__execute_knotc_expected_error(self, mock_exe):
mock_exe.return_value = ('xyz', '')
kb = Knot2Backend('foo')
kb._execute_knotc('a1', 'a2', expected_error='xyz')
mock_exe.assert_called_once_with('knotc', 'a1', 'a2')
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test__execute_knotc_expected_output(self, mock_exe):
mock_exe.return_value = ('xyz', '')
kb = Knot2Backend('foo')
kb._execute_knotc('a1', 'a2', expected_output='xyz')
mock_exe.assert_called_once_with('knotc', 'a1', 'a2')
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test__execute_knotc_with_error(self, mock_exe):
mock_exe.return_value = ('xyz', '')
kb = Knot2Backend('foo')
self.assertRaises(
exceptions.Backend,
kb._execute_knotc, 'a1', 'a2')
mock_exe.assert_called_once_with('knotc', 'a1', 'a2')
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test__execute_knotc_raising_exception(self, mock_exe):
mock_exe.side_effect = ProcessExecutionError
kb = Knot2Backend('foo')
self.assertRaises(
exceptions.Backend,
kb._execute_knotc, 'a1', 'a2')
mock_exe.assert_called_once_with('knotc', 'a1', 'a2')
class Knot2AgentBackendUnitTestCase(TestCase):
def _create_dnspy_zone(self, name):
zone_text = (
'$ORIGIN %(name)s\n%(name)s 3600 IN SOA %(ns)s '
'email.email.com. 1421777854 3600 600 86400 3600\n%(name)s '
'3600 IN NS %(ns)s\n') % {'name': name, 'ns': 'ns1.designate.com'}
return dns.zone.from_text(zone_text, check_origin=False)
def setUp(self):
super(Knot2AgentBackendUnitTestCase, self).setUp()
self.kb = Knot2Backend('foo')
self.patch_ob(self.kb, '_execute_knotc')
def tearDown(self):
super(Knot2AgentBackendUnitTestCase, self).tearDown()
def patch_ob(self, *a, **kw):
self.useFixture(fixtures.MockPatchObject(*a, **kw))
def test_create_zone(self, *mocks):
zone = self._create_dnspy_zone('example.org')
self.kb.create_zone(zone)
self.kb._execute_knotc.assert_has_calls([
call('conf-begin'),
call('conf-set', 'zone[example.org]',
expected_error='duplicate identifier'),
call('conf-commit'),
call('zone-refresh', 'example.org')
])
def test_create_zone_already_there(self, *mocks):
self.kb._execute_knotc.return_value = 'duplicate identifier'
zone = self._create_dnspy_zone('example.org')
self.kb.create_zone(zone)
self.kb._execute_knotc.assert_has_calls([
call('conf-begin'),
call('conf-set', 'zone[example.org]',
expected_error='duplicate identifier'),
call('conf-commit'),
call('zone-refresh', 'example.org')
])
def test__start_minidns_to_knot_axfr(self):
self.kb._start_minidns_to_knot_axfr('foo')
self.kb._execute_knotc.assert_called_with('zone-refresh', 'foo')
@mock.patch('oslo_concurrency.lockutils.lock')
def test__modify_zone(self, *mocks):
self.kb._modify_zone('blah', 'bar')
self.assertEqual(3, self.kb._execute_knotc.call_count)
self.kb._execute_knotc.assert_called_with('conf-commit')
@mock.patch('oslo_concurrency.lockutils.lock')
def test__modify_zone_exception(self, *mocks):
# Raise an exception during the second call to _execute_knotc
self.kb._execute_knotc.side_effect = [None, exceptions.Backend, None]
self.assertRaises(
exceptions.Backend,
self.kb._modify_zone, 'blah', 'bar')
self.assertEqual(3, self.kb._execute_knotc.call_count)
self.kb._execute_knotc.assert_has_calls([
call('conf-begin'),
call('blah', 'bar'),
call('conf-abort'),
])
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test_find_zone_serial(self, mock_exe):
mock_exe.return_value = "[example.com.] type: slave | serial: 20 | " \
"next-event: idle | auto-dnssec: disabled]", ""
serial = self.kb.find_zone_serial('example.com')
self.assertEqual(20, serial)
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test_find_zone_serial__zone_not_found(self, mock_exe):
mock_exe.side_effect = ProcessExecutionError(
"error: [example.com.] (no such zone found)")
serial = self.kb.find_zone_serial('example.com')
self.assertIsNone(serial)
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test_find_zone_serial_unexpected_output(self, mock_exe):
mock_exe.return_value = "bogus output", ""
self.assertRaises(
exceptions.Backend,
self.kb.find_zone_serial, 'example.com')
@mock.patch('designate.backend.agent_backend.impl_knot2.execute')
def test_find_zone_serial_error(self, mock_exe):
mock_exe.side_effect = ProcessExecutionError("blah")
self.assertRaises(
exceptions.Backend,
self.kb.find_zone_serial, 'example.com')
def test_update_zone(self):
zone = self._create_dnspy_zone('example.org')
self.kb.update_zone(zone)
self.kb._execute_knotc.assert_called_once_with(
'zone-refresh', 'example.org')
def test_delete_zone(self):
self.kb.delete_zone('example.org')
self.kb._execute_knotc.assert_has_calls([
call('conf-begin'),
call('conf-unset', 'zone[example.org]',
expected_error='invalid identifier'),
call('conf-commit'),
])
def test_delete_zone_already_gone(self):
self.kb._execute_knotc.return_value = 'duplicate identifier'
self.kb.delete_zone('example.org')
self.kb._execute_knotc.assert_has_calls([
call('conf-begin'),
call('conf-unset', 'zone[example.org]',
expected_error='invalid identifier'),
call('conf-commit'),
])

122
designate/tests/unit/test_agent/test_backends/test_msdns.py
View File

@ -1,122 +0,0 @@
# Copyright 2016 Cloudbase Solutions Srl
# All Rights Reserved.
#
# Author: Alin Balutoiu <abalutoiu@cloudbasesolutions.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 the MSDNS agent backend
"""
import dns
import mock
from os_win import constants
from os_win import exceptions as os_win_exc
from designate.backend.agent_backend import impl_msdns
from designate.tests import TestCase
class MSDNSAgentBackendUnitTestCase(TestCase):
_FAKE_ZONE_NAME = 'example.com'
def setUp(self):
super(MSDNSAgentBackendUnitTestCase, self).setUp()
self.CONF.set_override('masters', ['127.0.0.1:5354'], 'service:agent')
patcher = mock.patch('os_win.utilsfactory.get_dnsutils')
self._dnsutils = patcher.start().return_value
self.addCleanup(patcher.stop)
self.backend = impl_msdns.MSDNSBackend('foo')
def _create_dnspy_zone(self, name):
zone_text = (
'$ORIGIN %(name)s\n%(name)s 3600 IN SOA %(ns)s '
'email.email.com. 1421777854 3600 600 86400 3600\n%(name)s '
'3600 IN NS %(ns)s\n') % {'name': name, 'ns': 'ns1.designate.com'}
return dns.zone.from_text(zone_text, check_origin=False)
def test_init(self):
self.assertEqual(['127.0.0.1'], self.backend._masters)
self.assertEqual(self._dnsutils, self.backend._dnsutils)
def test_find_zone_serial(self):
serial = self.backend.find_zone_serial(self._FAKE_ZONE_NAME)
expected_serial = self._dnsutils.get_zone_serial.return_value
self.assertEqual(expected_serial, serial)
self._dnsutils.get_zone_serial.assert_called_once_with(
self._FAKE_ZONE_NAME)
def test_find_zone_serial_error(self):
self._dnsutils.get_zone_serial.side_effect = (
os_win_exc.DNSZoneNotFound(zone_name=self._FAKE_ZONE_NAME))
serial = self.backend.find_zone_serial(self._FAKE_ZONE_NAME)
self.assertIsNone(serial)
self._dnsutils.get_zone_serial.assert_called_once_with(
self._FAKE_ZONE_NAME)
def test_create_zone(self):
zone = self._create_dnspy_zone(self._FAKE_ZONE_NAME)
self.backend.create_zone(zone)
self._dnsutils.zone_create.assert_called_once_with(
zone_name=self._FAKE_ZONE_NAME,
zone_type=constants.DNS_ZONE_TYPE_SECONDARY,
ds_integrated=False,
ip_addrs=self.backend._masters)
def test_create_zone_already_existing_diff(self):
zone = self._create_dnspy_zone(self._FAKE_ZONE_NAME)
self._dnsutils.zone_create.side_effect = (
os_win_exc.DNSZoneAlreadyExists(zone_name=self._FAKE_ZONE_NAME))
self.assertRaises(os_win_exc.DNSZoneAlreadyExists,
self.backend.create_zone,
zone)
self._dnsutils.get_zone_properties.assert_called_once_with(
self._FAKE_ZONE_NAME)
def test_create_zone_already_existing_identical(self):
zone = self._create_dnspy_zone(self._FAKE_ZONE_NAME)
self._dnsutils.zone_create.side_effect = (
os_win_exc.DNSZoneAlreadyExists(zone_name=self._FAKE_ZONE_NAME))
mock_zone_properties = {'zone_type': constants.DNS_ZONE_TYPE_SECONDARY,
'ds_integrated': False,
'master_servers': self.backend._masters}
self._dnsutils.get_zone_properties.return_value = mock_zone_properties
self.backend.create_zone(zone)
self._dnsutils.get_zone_properties.assert_called_once_with(
self._FAKE_ZONE_NAME)
def test_update_zone(self):
zone = self._create_dnspy_zone(self._FAKE_ZONE_NAME)
self.backend.update_zone(zone)
self._dnsutils.zone_update.assert_called_once_with(
self._FAKE_ZONE_NAME)
def test_delete_zone(self):
self.backend.delete_zone(self._FAKE_ZONE_NAME)
self._dnsutils.zone_delete.assert_called_once_with(
self._FAKE_ZONE_NAME)