# Now make sure the clear_socket really works
ddns.clear_socket()
self.assertFalse(os.path.exists(ddns.SOCKET_FILE))
+ # Let ddns object complete any necessary cleanup (not part of the test,
+ # but for suppressing any warnings from the Python interpreter)
+ ddnss.shutdown_cleanup()
+ def test_initial_config(self):
+ # right now, the only configuration is the zone configuration, whose
+ # default should be an empty map.
+ self.assertEqual({}, self.ddns_server._zone_config)
+
def test_config_handler(self):
- # Config handler does not do anything yet, but should at least
- # return 'ok' for now.
- new_config = {}
+ # Update with a simple zone configuration: including an accept-all ACL
+ new_config = { 'zones': [ { 'origin': TEST_ZONE_NAME_STR,
+ 'class': TEST_RRCLASS_STR,
+ 'update_acl': [{'action': 'ACCEPT'}] } ] }
answer = self.ddns_server.config_handler(new_config)
self.assertEqual((0, None), isc.config.parse_answer(answer))
+ acl = self.ddns_server._zone_config[(TEST_ZONE_NAME, TEST_RRCLASS)]
+ self.assertEqual(ACCEPT, acl.execute(TEST_ACL_CONTEXT))
+
+ # Slightly more complicated one: containing multiple ACLs
+ new_config = { 'zones': [ { 'origin': 'example.com',
+ 'class': 'CH',
+ 'update_acl': [{'action': 'REJECT',
+ 'from': '2001:db8::1'}] },
+ { 'origin': TEST_ZONE_NAME_STR,
+ 'class': TEST_RRCLASS_STR,
+ 'update_acl': [{'action': 'ACCEPT'}] },
+ { 'origin': 'example.org',
+ 'class': 'CH',
+ 'update_acl': [{'action': 'DROP'}] } ] }
+ answer = self.ddns_server.config_handler(new_config)
+ self.assertEqual((0, None), isc.config.parse_answer(answer))
+ self.assertEqual(3, len(self.ddns_server._zone_config))
+ acl = self.ddns_server._zone_config[(TEST_ZONE_NAME, TEST_RRCLASS)]
+ self.assertEqual(ACCEPT, acl.execute(TEST_ACL_CONTEXT))
+
+ # empty zone config
+ new_config = { 'zones': [] }
+ answer = self.ddns_server.config_handler(new_config)
+ self.assertEqual((0, None), isc.config.parse_answer(answer))
+ self.assertEqual({}, self.ddns_server._zone_config)
+
+ # bad zone config data: bad name. The previous config shouls be kept.
+ bad_config = { 'zones': [ { 'origin': 'bad..example',
+ 'class': TEST_RRCLASS_STR,
+ 'update_acl': [{'action': 'ACCEPT'}] } ] }
+ answer = self.ddns_server.config_handler(bad_config)
+ self.assertEqual(1, isc.config.parse_answer(answer)[0])
+ self.assertEqual({}, self.ddns_server._zone_config)
+
+ # bad zone config data: bad class.
+ bad_config = { 'zones': [ { 'origin': TEST_ZONE_NAME_STR,
+ 'class': 'badclass',
+ 'update_acl': [{'action': 'ACCEPT'}] } ] }
+ answer = self.ddns_server.config_handler(bad_config)
+ self.assertEqual(1, isc.config.parse_answer(answer)[0])
+ self.assertEqual({}, self.ddns_server._zone_config)
+
+ # bad zone config data: bad ACL.
+ bad_config = { 'zones': [ { 'origin': TEST_ZONE_NAME_STR,
+ 'class': TEST_RRCLASS_STR,
+ 'update_acl': [{'action': 'badaction'}]}]}
+ answer = self.ddns_server.config_handler(bad_config)
+ self.assertEqual(1, isc.config.parse_answer(answer)[0])
+ self.assertEqual({}, self.ddns_server._zone_config)
+
+ # the first zone cofig is valid, but not the second. the first one
+ # shouldn't be installed.
+ bad_config = { 'zones': [ { 'origin': TEST_ZONE_NAME_STR,
+ 'class': TEST_RRCLASS_STR,
+ 'update_acl': [{'action': 'ACCEPT'}] },
+ { 'origin': 'bad..example',
+ 'class': TEST_RRCLASS_STR,
+ 'update_acl': [{'action': 'ACCEPT'}] } ] }
+ answer = self.ddns_server.config_handler(bad_config)
+ self.assertEqual(1, isc.config.parse_answer(answer)[0])
+ self.assertEqual({}, self.ddns_server._zone_config)
+
+ # Half-broken case: 'origin, class' pair is duplicate. For now we
+ # we accept it (the latter one will win)
+ dup_config = { 'zones': [ { 'origin': TEST_ZONE_NAME_STR,
+ 'class': TEST_RRCLASS_STR,
+ 'update_acl': [{'action': 'REJECT'}] },
+ { 'origin': TEST_ZONE_NAME_STR,
+ 'class': TEST_RRCLASS_STR,
+ 'update_acl': [{'action': 'ACCEPT'}] } ] }
+ answer = self.ddns_server.config_handler(dup_config)
+ self.assertEqual((0, None), isc.config.parse_answer(answer))
+ acl = self.ddns_server._zone_config[(TEST_ZONE_NAME, TEST_RRCLASS)]
+ self.assertEqual(ACCEPT, acl.execute(TEST_ACL_CONTEXT))
+
+ def test_datasrc_config(self):
+ # By default (in our faked config) it should be derived from the
+ # test data source
+ rrclass, datasrc_client = self.ddns_server._datasrc_info
+ self.assertEqual(RRClass.IN(), rrclass)
+ self.assertEqual(DataSourceClient.SUCCESS,
+ datasrc_client.find_zone(Name('example.org'))[0])
+
+ # emulating an update. calling add_remote_config_by_name is a
+ # convenient faked way to invoke the callback. We set the db file
+ # to a bogus one; the current implementation will create an unusable
+ # data source client.
+ self.__cc_session.auth_db_file = './notexistentdir/somedb.sqlite3'
+ self.__cc_session._auth_config = \
+ {'database_file': './notexistentdir/somedb.sqlite3'}
+ self.__cc_session.add_remote_config_by_name('Auth')
+ rrclass, datasrc_client = self.ddns_server._datasrc_info
+ self.assertEqual(RRClass.IN(), rrclass)
+ self.assertRaises(isc.datasrc.Error,
+ datasrc_client.find_zone, Name('example.org'))
+
+ # Check the current info isn't changed if the new config doesn't
+ # update it.
+ info_orig = self.ddns_server._datasrc_info
+ self.ddns_server._datasrc_info = 42 # dummy value, should be kept.
+ self.__cc_session._auth_config = {'other_config': 'value'}
+ self.__cc_session.add_remote_config_by_name('Auth')
+ self.assertEqual(42, self.ddns_server._datasrc_info)
+ self.ddns_server._datasrc_info = info_orig
+
+ def test_secondary_zones_config(self):
+ # By default it should be an empty list
+ self.assertEqual(set(), self.ddns_server._secondary_zones)
+
+ # emulating an update.
+ self.__cc_session._zonemgr_config = {'secondary_zones': [
+ {'name': TEST_ZONE_NAME_STR, 'class': TEST_RRCLASS_STR}]}
+ self.__cc_session.add_remote_config_by_name('Zonemgr')
+
+ # The new set of secondary zones should be stored.
+ self.assertEqual({(TEST_ZONE_NAME, TEST_RRCLASS)},
+ self.ddns_server._secondary_zones)
+
+ # Similar to the above, but 'class' is unspecified. The default value
+ # should be used.
+ self.__cc_session._zonemgr_config = {'secondary_zones': [
+ {'name': TEST_ZONE_NAME_STR}]}
+ self.__cc_session.add_remote_config_by_name('Zonemgr')
+ self.assertEqual({(TEST_ZONE_NAME, TEST_RRCLASS)},
+ self.ddns_server._secondary_zones)
+
+ # The given list has a duplicate. The resulting set should unify them.
+ self.__cc_session._zonemgr_config = {'secondary_zones': [
+ {'name': TEST_ZONE_NAME_STR, 'class': TEST_RRCLASS_STR},
+ {'name': TEST_ZONE_NAME_STR, 'class': TEST_RRCLASS_STR}]}
+ self.__cc_session.add_remote_config_by_name('Zonemgr')
+ self.assertEqual({(TEST_ZONE_NAME, TEST_RRCLASS)},
+ self.ddns_server._secondary_zones)
+
+ # Check the 2ndary zones aren't changed if the new config doesn't
+ # update it.
+ seczones_orig = self.ddns_server._secondary_zones
+ self.ddns_server._secondary_zones = 42 # dummy value, should be kept.
+ self.__cc_session._zonemgr_config = {}
+ self.__cc_session.add_remote_config_by_name('Zonemgr')
+ self.assertEqual(42, self.ddns_server._secondary_zones)
+ self.ddns_server._secondary_zones = seczones_orig
+
+ # If the update config is broken, the existing set should be intact.
+ self.__cc_session._zonemgr_config = {'secondary_zones': [
+ {'name': 'good.example', 'class': TEST_RRCLASS_STR},
+ {'name': 'badd..example', 'class': TEST_RRCLASS_STR}]}
+ self.__cc_session.add_remote_config_by_name('Zonemgr')
+ self.assertEqual({(TEST_ZONE_NAME, TEST_RRCLASS)},
+ self.ddns_server._secondary_zones)
+
+ def __check_remote_config_fail(self, mod_name, num_ex, expected_ex):
+ '''Subroutine for remote_config_fail test.'''
+
+ # fake pause function for inspection and to avoid having timeouts
+ added_pause = []
+ ddns.add_pause = lambda sec: added_pause.append(sec)
+
+ # In our current implementation, there will be up to 3 tries of
+ # adding the module, each separated by a 1-sec pause. If all attempts
+ # fail the exception will be propagated.
+ exceptions = [expected_ex for i in range(0, num_ex)]
+ self.__cc_session._raise_mods = {mod_name: exceptions}
+ if num_ex >= 3:
+ self.assertRaises(expected_ex, ddns.DDNSServer, self.__cc_session)
+ else:
+ ddns.DDNSServer(self.__cc_session)
+ self.assertEqual([1 for i in range(0, num_ex)], added_pause)
+
+ def test_remote_config_fail(self):
+ # If getting config of Auth or Zonemgr fails on construction of
+ # DDNServer, it should result in an exception and a few times
+ # of retries. We test all possible cases, changing the number of
+ # raised exceptions and the type of exceptions that can happen,
+ # which should also cover the fatal error case.
+ for i in range(0, 4):
+ self.__check_remote_config_fail('Auth', i, ModuleCCSessionError)
+ self.__check_remote_config_fail('Auth', i, ModuleSpecError)
+ self.__check_remote_config_fail('Zonemgr', i, ModuleCCSessionError)
+ self.__check_remote_config_fail('Zonemgr', i, ModuleSpecError)
def test_shutdown_command(self):
'''Test whether the shutdown command works'''
def check_push_and_pop(self, family, type, protocol, local, remote,
data, new_connection):
- sock = self.create_socket(family, type, protocol, local, True)
- fwd_fd = sock.fileno()
- if protocol == IPPROTO_TCP:
- client_addr = ('::1', 0, 0, 0) if family == AF_INET6 \
- else ('127.0.0.1', 0)
- client_sock = self.create_socket(family, type, protocol,
- client_addr, False)
- client_sock.setblocking(False)
- try:
- client_sock.connect(local)
- except socket.error:
- pass
- server_sock, _ = sock.accept()
- fwd_fd = server_sock.fileno()
-
- # If a new connection is required, start the "server", have the
- # internal forwarder connect to it, and then internally accept it.
- if new_connection:
- self.start_listen()
- self.forwarder.connect_to_receiver()
- self.accept_sock = self.accept_forwarder()
-
- # Then push one socket session via the forwarder.
- self.forwarder.push(fwd_fd, family, type, protocol, local, remote,
- data)
-
- # Pop the socket session we just pushed from a local receiver, and
- # check the content.
- receiver = SocketSessionReceiver(self.accept_sock)
- signal.alarm(1)
- sock_session = receiver.pop()
- signal.alarm(0)
- passed_sock = sock_session[0]
- self.assertNotEqual(fwd_fd, passed_sock.fileno())
- self.assertEqual(family, passed_sock.family)
- self.assertEqual(type, passed_sock.type)
- self.assertEqual(protocol, passed_sock.proto)
- self.assertEqual(local, sock_session[1])
- self.assertEqual(remote, sock_session[2])
- self.assertEqual(data, sock_session[3])
-
- # Check if the passed FD is usable by sending some data from it.
- passed_sock.setblocking(True)
- if protocol == IPPROTO_UDP:
- self.assertEqual(len(TEST_DATA), passed_sock.sendto(TEST_DATA,
- local))
- sock.settimeout(10)
- self.assertEqual(TEST_DATA, sock.recvfrom(len(TEST_DATA))[0])
- else:
- server_sock.close()
- self.assertEqual(len(TEST_DATA), passed_sock.send(TEST_DATA))
- client_sock.setblocking(True)
- client_sock.settimeout(10)
- self.assertEqual(TEST_DATA, client_sock.recv(len(TEST_DATA)))
+ with self.create_socket(family, type, protocol, local, True) as sock:
+ fwd_fd = sock.fileno()
+ if protocol == IPPROTO_TCP:
+ client_addr = ('::1', 0, 0, 0) if family == AF_INET6 \
+ else ('127.0.0.1', 0)
+ client_sock = self.create_socket(family, type, protocol,
+ client_addr, False)
+ client_sock.setblocking(False)
+ try:
+ client_sock.connect(local)
+ except socket.error:
+ pass
+ server_sock, _ = sock.accept()
+ fwd_fd = server_sock.fileno()
+
+ # If a new connection is required, start the "server", have the
+ # internal forwarder connect to it, and then internally accept it.
+ if new_connection:
+ self.start_listen()
+ self.forwarder.connect_to_receiver()
+ self.accept_sock = self.accept_forwarder()
+
+ # Then push one socket session via the forwarder.
+ self.forwarder.push(fwd_fd, family, type, protocol, local, remote,
+ data)
+
+ # Pop the socket session we just pushed from a local receiver, and
+ # check the content.
+ receiver = SocketSessionReceiver(self.accept_sock)
+ signal.alarm(1)
+ sock_session = receiver.pop()
+ signal.alarm(0)
+ passed_sock = sock_session[0]
+ self.assertNotEqual(fwd_fd, passed_sock.fileno())
+ self.assertEqual(family, passed_sock.family)
+ self.assertEqual(type, passed_sock.type)
+ self.assertEqual(protocol, passed_sock.proto)
+ self.assertEqual(local, sock_session[1])
+ self.assertEqual(remote, sock_session[2])
+ self.assertEqual(data, sock_session[3])
+
+ # Check if the passed FD is usable by sending some data from it.
+ passed_sock.setblocking(True)
+ if protocol == IPPROTO_UDP:
+ self.assertEqual(len(TEST_DATA), passed_sock.sendto(TEST_DATA,
+ local))
+ sock.settimeout(10)
+ self.assertEqual(TEST_DATA, sock.recvfrom(len(TEST_DATA))[0])
+ else:
+ self.assertEqual(len(TEST_DATA), passed_sock.send(TEST_DATA))
+ client_sock.setblocking(True)
+ client_sock.settimeout(10)
+ self.assertEqual(TEST_DATA, client_sock.recv(len(TEST_DATA)))
+ server_sock.close()
+ client_sock.close()
+
+ passed_sock.close()
def test_push_and_pop(self):
- # This is a straightforward port of C++ pushAndPop test.
+ # This is a straightforward port of C++ pushAndPop test. See the
+ # C++ version why we use multiple ports for "local".
local6 = ('::1', TEST_PORT, 0, 0)
+ local6_alt = ('::1', TEST_PORT2, 0, 0)
+ local6_alt2 = ('::1', TEST_PORT3, 0, 0)
remote6 = ('2001:db8::1', 5300, 0, 0)
self.check_push_and_pop(AF_INET6, SOCK_DGRAM, IPPROTO_UDP,
local6, remote6, TEST_DATA, True)
projects / thirdparty / kea.git / commitdiff
