num_rrsets = len(self.__req_message.get_section(SECTION_PREREQUISITE))
self.assertEqual(2, num_rrsets)
- def check_session_msg(self, result, expect_recv=1, notify_auth=False):
+ def check_session_msg(self, result, expect_recv=2):
'''Check post update communication with other modules.'''
# iff the update succeeds, b10-ddns should tell interested other
# modules the information about the update zone. Possible modules
# 'zone_class', <updated_zone_class>}]}
# for auth, it should be:
# {'command': ['loadzone', {'origin': <updated_zone_name>,
- # 'class', <updated_zone_class>,
- # 'datasrc', <datasrc type, should be
- # "memory" in practice>}]}
+ # 'class', <updated_zone_class>}]}
# and expect an answer by calling group_recvmsg().
#
# expect_recv indicates the expected number of calls to
- # group_recvmsg(), which is normally 1, but can be 0 if send fails;
+ # group_recvmsg(), which is normally 2, but can be 0 if send fails;
# if the message is to be sent
if result == UPDATE_SUCCESS:
- expected_sentmsg = 2 if notify_auth else 1
+ expected_sentmsg = 2
self.assertEqual(expected_sentmsg,
len(self.__cc_session._sent_msg))
self.assertEqual(expect_recv, self.__cc_session._recvmsg_called)
msg_cnt = 0
- if notify_auth:
- sent_msg, sent_group = self.__cc_session._sent_msg[msg_cnt]
- sent_cmd = sent_msg['command']
- self.assertEqual('Auth', sent_group)
- self.assertEqual('loadzone', sent_cmd[0])
- self.assertEqual(3, len(sent_cmd[1]))
- self.assertEqual(TEST_ZONE_NAME.to_text(),
- sent_cmd[1]['origin'])
- self.assertEqual(TEST_RRCLASS.to_text(),
- sent_cmd[1]['class'])
- self.assertEqual('memory', sent_cmd[1]['datasrc'])
- msg_cnt += 1
+ sent_msg, sent_group = self.__cc_session._sent_msg[msg_cnt]
+ sent_cmd = sent_msg['command']
+ self.assertEqual('Auth', sent_group)
+ self.assertEqual('loadzone', sent_cmd[0])
+ self.assertEqual(2, len(sent_cmd[1]))
+ self.assertEqual(TEST_ZONE_NAME.to_text(),
+ sent_cmd[1]['origin'])
+ self.assertEqual(TEST_RRCLASS.to_text(),
+ sent_cmd[1]['class'])
+ msg_cnt += 1
sent_msg, sent_group = self.__cc_session._sent_msg[msg_cnt]
sent_cmd = sent_msg['command']
self.assertEqual('Xfrout', sent_group)
[{'type': 'memory', 'class': 'IN', 'zones': [
{'origin': TEST_ZONE_NAME_STR, 'filetype': 'sqlite3'}]}]
self.check_session()
- self.check_session_msg(UPDATE_SUCCESS, expect_recv=2, notify_auth=True)
+ self.check_session_msg(UPDATE_SUCCESS)
# Let sendmsg() raise an exception. The first exception shouldn't
# stop sending the second message. There's just no recv calls.
self.__cc_session.clear_msg()
self.__cc_session._sendmsg_exception = SessionError('send error')
self.check_session()
- self.check_session_msg(UPDATE_SUCCESS, expect_recv=0, notify_auth=True)
+ self.check_session_msg(UPDATE_SUCCESS, expect_recv=0)
# Likewise, in the case recvmsg() raises (and there should be recv
# calls in this case)
self.__cc_session.clear_msg()
self.__cc_session._recvmsg_exception = SessionError('recv error')
self.check_session()
- self.check_session_msg(UPDATE_SUCCESS, expect_recv=2, notify_auth=True)
+ self.check_session_msg(UPDATE_SUCCESS)
def test_session_with_config(self):
'''Check a session with more realistic config setups.
class TestXfrinProcessMockCC:
def __init__(self):
- self.get_called = False
- self.get_called_correctly = False
self.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
# Create a connection for each attempt
self.assertEqual(len(transfers), self.__created_connections)
self.assertEqual([published], self.__published)
- if published == XFRIN_OK:
- self.assertTrue(self._module_cc.get_called)
- self.assertTrue(self._module_cc.get_called_correctly)
- else:
- self.assertFalse(self._module_cc.get_called)
- self.assertFalse(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.assertFalse(self._send_cc_session.send_called)
- self.assertFalse(self._send_cc_session.send_called_correctly)
- self.assertFalse(self._send_cc_session.recv_called)
- self.assertFalse(self._send_cc_session.recv_called_correctly)
+ # Check there was loadzone command
+ self.assertTrue(self._send_cc_session.send_called)
+ self.assertTrue(self._send_cc_session.send_called_correctly)
+ self.assertTrue(self._send_cc_session.recv_called)
+ self.assertTrue(self._send_cc_session.recv_called_correctly)
def test_axfr_ok(self):
"""
self.__do_test([XFRIN_FAIL, XFRIN_FAIL],
[RRType.IXFR(), RRType.AXFR()], RRType.IXFR())
- def test_inmem_ok(self):
+ def test_send_loadzone(self):
"""
- Inmem configuration where all the configuration is just right
- for loadzone to be sent to b10-auth (origin is the name received
- by xfrin, filetype is sqlite3, type is memory and class is the
- one received by xfrin).
+ Check the loadzone command is sent after successful transfer.
"""
- 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.assertTrue(self._send_cc_session.send_called)
self.assertTrue(self._send_cc_session.send_called_correctly)
self.assertTrue(self._send_cc_session.recv_called)
self.assertTrue(self._send_cc_session.recv_called_correctly)
- def test_inmem_datasource_type_not_memory(self):
- """
- Inmem configuration where the datasource type is not memory. In
- this case, loadzone should not be sent to b10-auth.
- """
- 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.assertFalse(self._send_cc_session.send_called)
- self.assertFalse(self._send_cc_session.send_called_correctly)
- self.assertFalse(self._send_cc_session.recv_called)
- self.assertFalse(self._send_cc_session.recv_called_correctly)
-
- def test_inmem_datasource_type_is_missing(self):
- """
- Inmem configuration where the datasource type is missing. In
- this case, loadzone should not be sent to b10-auth.
- """
- self._module_cc.config = [{'zones': [{'origin': 'example.org', 'filetype': 'sqlite3',
- 'file': 'data/inmem-xfrin.sqlite3'}],
- 'class': 'IN'}]
- self.__do_test([XFRIN_OK], [RRType.IXFR()], RRType.IXFR())
- self.assertFalse(self._send_cc_session.send_called)
- self.assertFalse(self._send_cc_session.send_called_correctly)
- self.assertFalse(self._send_cc_session.recv_called)
- self.assertFalse(self._send_cc_session.recv_called_correctly)
-
- def test_inmem_backend_type_not_sqlite3(self):
- """
- Inmem configuration where the datasource backing file is not of
- type sqlite3. In this case, loadzone should not be sent to
- b10-auth.
- """
- self._module_cc.config = [{'zones': [{'origin': 'example.org', 'filetype': 'postgresql',
- 'file': 'data/inmem-xfrin.db'}],
- 'type': 'memory', 'class': 'IN'}]
- self.__do_test([XFRIN_OK], [RRType.IXFR()], RRType.IXFR())
- self.assertFalse(self._send_cc_session.send_called)
- self.assertFalse(self._send_cc_session.send_called_correctly)
- self.assertFalse(self._send_cc_session.recv_called)
- self.assertFalse(self._send_cc_session.recv_called_correctly)
-
- def test_inmem_backend_type_is_missing(self):
- """
- Inmem configuration where the datasource backing file type is
- not set. In this case, loadzone should not be sent to b10-auth.
- """
- self._module_cc.config = [{'zones': [{'origin': 'example.org',
- 'file': 'data/inmem-xfrin'}],
- 'type': 'memory', 'class': 'IN'}]
- self.__do_test([XFRIN_OK], [RRType.IXFR()], RRType.IXFR())
- self.assertFalse(self._send_cc_session.send_called)
- self.assertFalse(self._send_cc_session.send_called_correctly)
- self.assertFalse(self._send_cc_session.recv_called)
- self.assertFalse(self._send_cc_session.recv_called_correctly)
-
- def test_inmem_class_is_different(self):
- """
- Inmem configuration where the datasource class does not match
- the received class. In this case, loadzone should not be sent to
- b10-auth.
- """
- 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.assertFalse(self._send_cc_session.send_called)
- self.assertFalse(self._send_cc_session.send_called_correctly)
- self.assertFalse(self._send_cc_session.recv_called)
- self.assertFalse(self._send_cc_session.recv_called_correctly)
-
- def test_inmem_class_is_missing(self):
- """
- Inmem configuration where the datasource class is missing. In
- this case, we assume the IN class and loadzone may be sent to
- b10-auth if everything else matches.
- """
- self._module_cc.config = [{'zones': [{'origin': 'example.org', 'filetype': 'sqlite3',
- 'file': 'data/inmem-xfrin.sqlite3'}],
- 'type': 'memory'}]
- self.__do_test([XFRIN_OK], [RRType.IXFR()], RRType.IXFR())
- self.assertTrue(self._send_cc_session.send_called)
- self.assertTrue(self._send_cc_session.send_called_correctly)
- self.assertTrue(self._send_cc_session.recv_called)
- self.assertTrue(self._send_cc_session.recv_called_correctly)
-
- def test_inmem_name_doesnt_match(self):
- """
- Inmem configuration where the origin does not match the received
- name. In this case, loadzone should not be sent to b10-auth.
- """
- 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.assertFalse(self._send_cc_session.send_called)
- self.assertFalse(self._send_cc_session.send_called_correctly)
- self.assertFalse(self._send_cc_session.recv_called)
- self.assertFalse(self._send_cc_session.recv_called_correctly)
-
- def test_inmem_name_is_missing(self):
- """
- Inmem configuration where the origin is missing. In this case,
- loadzone should not be sent to b10-auth.
- """
- self._module_cc.config = [{'zones': [{'filetype': 'sqlite3',
- 'file': 'data/inmem-xfrin.sqlite3'}],
- 'type': 'memory', 'class': 'IN'}]
- self.__do_test([XFRIN_OK], [RRType.IXFR()], RRType.IXFR())
- self.assertFalse(self._send_cc_session.send_called)
- self.assertFalse(self._send_cc_session.send_called_correctly)
- self.assertFalse(self._send_cc_session.recv_called)
- self.assertFalse(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.
def auth_loadzone_command(module_cc, zone_name, zone_class):
'''Create a 'loadzone' command with a given zone for Auth server.
- This function checks the Auth module configuration to see if it
- servers a given zone via an in-memory data source on top of SQLite3
- data source, and, if so, generate an inter-module command for Auth
+ This function generates an inter-module command for Auth
to force it to reload the zone.
Parameters:
zone_class (isc.dns.RRClass): the RR class of the zone to be possibly
reloaded.
- Return: a CC command message for the reload if the zone is found;
- otherwise None.
-
+ Return: a CC command message for the reload.
'''
# Note: this function was originally a dedicated subroutine of xfrin,
# but was moved here so it can be shared by some other modules
# deprecated (which is a more likely scenario). For this reason, the
# corresponding tests were still kept in xfrin.
- datasources, is_default =\
- module_cc.get_remote_config_value(AUTH_MODULE_NAME, "datasources")
- if is_default:
- return None
- for d in datasources:
- if "type" not in d:
- continue
- try:
- if "class" in d:
- dclass = RRClass(d["class"])
- else:
- dclass = RRClass("IN")
- except InvalidRRClass as err:
- logger.info(PYSERVER_COMMON_AUTH_CONFIG_RRCLASS_ERROR, err)
- continue
-
- if d["type"].lower() == "memory" and dclass == zone_class:
- for zone in d["zones"]:
- if "filetype" not in zone:
- continue
- if "origin" not in zone:
- continue
- if "filetype" not in zone:
- continue
- try:
- name = Name(zone["origin"])
- except (EmptyLabel, TooLongLabel, BadLabelType, BadEscape,
- TooLongName, IncompleteName):
- logger.info(PYSERVER_COMMON_AUTH_CONFIG_NAME_PARSER_ERROR,
- err)
- continue
+ # Note: The function got very simplified by #1976. There's plan to move
+ # to notification-driven approach, at which point the function would
+ # be changed a lot.
- if zone["filetype"].lower() == "sqlite3" and name == zone_name:
- param = {"origin": zone_name.to_text(),
- "class": zone_class.to_text(),
- "datasrc": d["type"]}
- return create_command("loadzone", param)
- return None
+ param = {
+ "origin": zone_name.to_text(),
+ "class": zone_class.to_text()
+ }
+ return create_command("loadzone", param)
projects / thirdparty / kea.git / commitdiff
