BIND 10 trac1790, updated. 9174cdfbe4f982ed671ce8650867207b74763553 [1790] Add unit tests using mock objects
BIND 10 source code commits
bind10-changes at lists.isc.org
Tue May 8 04:43:56 UTC 2012
The branch, trac1790 has been updated
via 9174cdfbe4f982ed671ce8650867207b74763553 (commit)
via efa6cdbe1d576cb6cbb12cc62c8573156c101a74 (commit)
from 2e94c1ca0ccd402a4707745722f2ff66ba2e5557 (commit)
Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.
- Log -----------------------------------------------------------------
commit 9174cdfbe4f982ed671ce8650867207b74763553
Author: Mukund Sivaraman <muks at isc.org>
Date: Tue May 8 10:05:57 2012 +0530
[1790] Add unit tests using mock objects
commit efa6cdbe1d576cb6cbb12cc62c8573156c101a74
Author: Mukund Sivaraman <muks at isc.org>
Date: Tue May 8 08:23:53 2012 +0530
[master] Fix test class name
-----------------------------------------------------------------------
Summary of changes:
src/bin/xfrin/tests/xfrin_test.py | 118 ++++++++++++++++++++++++++++++++++++-
src/bin/xfrin/xfrin.py.in | 82 +++++++++++++-------------
2 files changed, 158 insertions(+), 42 deletions(-)
-----------------------------------------------------------------------
diff --git a/src/bin/xfrin/tests/xfrin_test.py b/src/bin/xfrin/tests/xfrin_test.py
index b88d6a9..7aba2b6 100644
--- a/src/bin/xfrin/tests/xfrin_test.py
+++ b/src/bin/xfrin/tests/xfrin_test.py
@@ -2577,7 +2577,7 @@ class TestXfrin(unittest.TestCase):
self.common_ixfr_setup('refresh', False)
self.assertEqual(RRType.AXFR(), self.xfr.xfrin_started_request_type)
-class TextXfrinMemoryZones(unittest.TestCase):
+class TestXfrinMemoryZones(unittest.TestCase):
def setUp(self):
self.xfr = MockXfrin()
# Configuration snippet containing 2 memory datasources,
@@ -2736,6 +2736,44 @@ class TestMain(unittest.TestCase):
MockXfrin.check_command_hook = raise_exception
main(MockXfrin, False)
+class TestXfrinProcessMockCC:
+ def __init__(self, config = []):
+ self.get_called = False
+ self.get_called_correctly = False
+ self.config = config
+
+ def get_remote_config_value(self, module, identifier):
+ self.get_called = True
+ if module == 'Auth' and identifier == 'datasources':
+ self.get_called_correctly = True
+ return (self.config, False)
+ else:
+ return (None, True)
+
+class TestXfrinProcessMockCCSession:
+ def __init__(self):
+ self.send_called = False
+ self.send_called_correctly = False
+ self.recv_called = False
+ self.recv_called_correctly = False
+
+ def group_sendmsg(self, msg, module):
+ self.send_called = True
+ if module == 'Auth' and msg['command'][0] == 'loadzone':
+ self.send_called_correctly = True
+ seq = "random-e068c2de26d760f20cf10afc4b87ef0f"
+ else:
+ seq = None
+
+ return seq
+
+ def group_recvmsg(self, message, seq):
+ self.recv_called = True
+ if message == False and seq == "random-e068c2de26d760f20cf10afc4b87ef0f":
+ self.recv_called_correctly = True
+ # return values are ignored
+ return (None, None)
+
class TestXfrinProcess(unittest.TestCase):
"""
Some tests for the xfrin_process function. This replaces the
@@ -2751,6 +2789,8 @@ class TestXfrinProcess(unittest.TestCase):
Also sets up several internal variables to watch what happens.
"""
+ self._module_cc = TestXfrinProcessMockCC()
+ self._send_cc_session = TestXfrinProcessMockCCSession()
# This will hold a "log" of what transfers were attempted.
self.__transfers = []
# This will "log" if failures or successes happened.
@@ -2795,6 +2835,9 @@ class TestXfrinProcess(unittest.TestCase):
Part of pretending to be the server as well. This just logs the
success/failure of the previous operation.
"""
+ if ret == XFRIN_OK:
+ xfrin._do_auth_loadzone(self, zone_name, rrclass)
+
self.__published.append(ret)
def close(self):
@@ -2825,12 +2868,19 @@ class TestXfrinProcess(unittest.TestCase):
# Create a connection for each attempt
self.assertEqual(len(transfers), self.__created_connections)
self.assertEqual([published], self.__published)
+ if published == XFRIN_OK:
+ self.assertEqual(True, self._module_cc.get_called)
+ self.assertEqual(True, self._module_cc.get_called_correctly)
def test_ixfr_ok(self):
"""
Everything OK the first time, over IXFR.
"""
self.__do_test([XFRIN_OK], [RRType.IXFR()], RRType.IXFR())
+ self.assertEqual(False, self._send_cc_session.send_called)
+ self.assertEqual(False, self._send_cc_session.send_called_correctly)
+ self.assertEqual(False, self._send_cc_session.recv_called)
+ self.assertEqual(False, self._send_cc_session.recv_called_correctly)
def test_axfr_ok(self):
"""
@@ -2861,6 +2911,72 @@ class TestXfrinProcess(unittest.TestCase):
"""
self.__do_test([XFRIN_FAIL, XFRIN_FAIL],
[RRType.IXFR(), RRType.AXFR()], RRType.IXFR())
+
+ def test_inmem_ok(self):
+ """
+ Inmem configuration 1.
+ """
+ self._module_cc.config = [{'zones': [{'origin': 'example.org', 'filetype': 'sqlite3',
+ 'file': 'data/inmem-xfrin.sqlite3'}],
+ 'type': 'memory', 'class': 'IN'}]
+ self.__do_test([XFRIN_OK], [RRType.IXFR()], RRType.IXFR())
+ self.assertEqual(True, self._send_cc_session.send_called)
+ self.assertEqual(True, self._send_cc_session.send_called_correctly)
+ self.assertEqual(True, self._send_cc_session.recv_called)
+ self.assertEqual(True, self._send_cc_session.recv_called_correctly)
+
+ def test_inmem_not_memory(self):
+ """
+ Inmem configuration 2.
+ """
+ self._module_cc.config = [{'zones': [{'origin': 'example.org', 'filetype': 'sqlite3',
+ 'file': 'data/inmem-xfrin.sqlite3'}],
+ 'type': 'punched-card', 'class': 'IN'}]
+ self.__do_test([XFRIN_OK], [RRType.IXFR()], RRType.IXFR())
+ self.assertEqual(False, self._send_cc_session.send_called)
+ self.assertEqual(False, self._send_cc_session.send_called_correctly)
+ self.assertEqual(False, self._send_cc_session.recv_called)
+ self.assertEqual(False, self._send_cc_session.recv_called_correctly)
+
+ def test_inmem_not_sqlite3(self):
+ """
+ Inmem configuration 3.
+ """
+ self._module_cc.config = [{'zones': [{'origin': 'example.org', 'filetype': 'postgresql',
+ 'file': 'data/inmem-xfrin.sqlite3'}],
+ 'type': 'memory', 'class': 'IN'}]
+ self.__do_test([XFRIN_OK], [RRType.IXFR()], RRType.IXFR())
+ self.assertEqual(False, self._send_cc_session.send_called)
+ self.assertEqual(False, self._send_cc_session.send_called_correctly)
+ self.assertEqual(False, self._send_cc_session.recv_called)
+ self.assertEqual(False, self._send_cc_session.recv_called_correctly)
+
+ def test_inmem_not_of_same_class(self):
+ """
+ Inmem configuration 4.
+ """
+ self._module_cc.config = [{'zones': [{'origin': 'example.org', 'filetype': 'sqlite3',
+ 'file': 'data/inmem-xfrin.sqlite3'}],
+ 'type': 'memory', 'class': 'XX'}]
+ self.__do_test([XFRIN_OK], [RRType.IXFR()], RRType.IXFR())
+ self.assertEqual(False, self._send_cc_session.send_called)
+ self.assertEqual(False, self._send_cc_session.send_called_correctly)
+ self.assertEqual(False, self._send_cc_session.recv_called)
+ self.assertEqual(False, self._send_cc_session.recv_called_correctly)
+
+ def test_inmem_not_present(self):
+ """
+ Inmem configuration 5.
+ """
+ self._module_cc.config = [{'zones': [{'origin': 'isc.org', 'filetype': 'sqlite3',
+ 'file': 'data/inmem-xfrin.sqlite3'}],
+ 'type': 'memory', 'class': 'IN'}]
+ self.__do_test([XFRIN_OK], [RRType.IXFR()], RRType.IXFR())
+ self.assertEqual(False, self._send_cc_session.send_called)
+ self.assertEqual(False, self._send_cc_session.send_called_correctly)
+ self.assertEqual(False, self._send_cc_session.recv_called)
+ self.assertEqual(False, self._send_cc_session.recv_called_correctly)
+
class TestFormatting(unittest.TestCase):
# If the formatting functions are moved to a more general library
# (ticket #1379), these tests should be moved with them.
diff --git a/src/bin/xfrin/xfrin.py.in b/src/bin/xfrin/xfrin.py.in
index 7beb5a5..20769c5 100755
--- a/src/bin/xfrin/xfrin.py.in
+++ b/src/bin/xfrin/xfrin.py.in
@@ -1248,6 +1248,46 @@ class ZoneInfo:
return (self.master_addr.family, socket.SOCK_STREAM,
(str(self.master_addr), self.master_port))
+def _do_auth_loadzone(server, zone_name, zone_class):
+ # On a successful zone transfer, if the zone is served by
+ # b10-auth in the in-memory data source using sqlite3 as a
+ # backend, send the "loadzone" command for the zone to auth.
+ datasources, is_default =\
+ server._module_cc.get_remote_config_value(AUTH_MODULE_NAME, "datasources")
+ if is_default:
+ return
+ for d in datasources:
+ try:
+ if "class" in d:
+ dclass = RRClass(d["class"])
+ else:
+ dclass = RRClass("IN")
+ except InvalidRRClass as err:
+ logger.info(XFRIN_AUTH_CONFIG_RRCLASS_ERROR, str(err))
+ continue
+
+ if d["type"].lower() == "memory" and dclass == zone_class:
+ for zone in d["zones"]:
+ if "filetype" not in zone:
+ continue
+ try:
+ name = Name(zone["origin"])
+ except (EmptyLabel, TooLongLabel, BadLabelType, BadEscape, TooLongName, IncompleteName):
+ logger.info(XFRIN_AUTH_CONFIG_NAME_PARSER_ERROR, str(err))
+ continue
+
+ if zone["filetype"].lower() == "sqlite3" and name == zone_name:
+ param = {"origin": zone_name.to_text(),
+ "class": zone_class.to_text(),
+ "datasrc": d["type"]}
+
+ logger.debug(DBG_XFRIN_TRACE, XFRIN_AUTH_LOADZONE,
+ param["origin"], param["class"], param["datasrc"])
+
+ msg = create_command("loadzone", param)
+ seq = server._send_cc_session.group_sendmsg(msg, AUTH_MODULE_NAME)
+ answer, env = server._send_cc_session.group_recvmsg(False, seq)
+
class Xfrin:
def __init__(self):
self._max_transfers_in = 10
@@ -1541,46 +1581,6 @@ class Xfrin:
"bind10_zones.sqlite3"
self._db_file = db_file
- def _do_auth_loadzone(self, zone_name, zone_class):
- # On a successful zone transfer, if the zone is served by
- # b10-auth in the in-memory data source using sqlite3 as a
- # backend, send the "loadzone" command for the zone to auth.
- datasources, is_default =\
- self._module_cc.get_remote_config_value(AUTH_MODULE_NAME, "datasources")
- if is_default:
- return
- for d in datasources:
- try:
- if "class" in d:
- dclass = RRClass(d["class"])
- else:
- dclass = RRClass("IN")
- except InvalidRRClass as err:
- logger.info(XFRIN_AUTH_CONFIG_RRCLASS_ERROR, str(err))
- continue
-
- if d["type"].lower() == "memory" and dclass == zone_class:
- for zone in d["zones"]:
- if "filetype" not in zone:
- continue
- try:
- name = Name(zone["origin"])
- except (EmptyLabel, TooLongLabel, BadLabelType, BadEscape, TooLongName, IncompleteName):
- logger.info(XFRIN_AUTH_CONFIG_NAME_PARSER_ERROR, str(err))
- continue
-
- if zone["filetype"].lower() == "sqlite3" and name == zone_name:
- param = {"origin": zone_name.to_text(),
- "class": zone_class.to_text(),
- "datasrc": d["type"]}
-
- logger.debug(DBG_XFRIN_TRACE, XFRIN_AUTH_LOADZONE,
- param["origin"], param["class"], param["datasrc"])
-
- msg = create_command("loadzone", param)
- seq = self._send_cc_session.group_sendmsg(msg, AUTH_MODULE_NAME)
- answer, env = self._send_cc_session.group_recvmsg(False, seq)
-
def publish_xfrin_news(self, zone_name, zone_class, xfr_result):
'''Send command to xfrout/zone manager module.
If xfrin has finished successfully for one zone, tell the good
@@ -1590,7 +1590,7 @@ class Xfrin:
param = {'zone_name': zone_name.to_text(),
'zone_class': zone_class.to_text()}
if xfr_result == XFRIN_OK:
- self._do_auth_loadzone(zone_name, zone_class)
+ _do_auth_loadzone(self, zone_name, zone_class)
msg = create_command(notify_out.ZONE_NEW_DATA_READY_CMD, param)
# catch the exception, in case msgq has been killed.
try:
More information about the bind10-changes
mailing list