zone_type, datasrc_client = self.__zone_config.find_zone(zname, zclass)
if zone_type == isc.ddns.zone_config.ZONE_PRIMARY:
return datasrc_client, zname, zclass
+ elif zone_type == isc.ddns.zone_config.ZONE_SECONDARY:
+ # unconditionally refused forwarding (we don't support it yet)
+ raise ZoneError('Update forwarding not supported',
+ isc.dns.Rcode.REFUSED())
def __make_response(self, rcode):
'''Transform the internal message to the update response.
self.__update_msgdata, self.__update_msg = create_update_msg()
self.__session = UpdateSession(self.__update_msg,
self.__update_msgdata,
- self.__client_addr, ZoneConfig())
+ self.__client_addr,
+ ZoneConfig([(Name("example.org"),
+ TEST_RRCLASS)]))
def check_response(self, msg, expected_rcode):
'''Perform common checks on update resposne message.'''
self.assertEqual(UPDATE_DONE, session.handle())
self.check_response(session.get_message(), Rcode.FORMERR())
+ def test_update_secondary(self):
+ # specified zone is configured as a secondary. Since this
+ # implementation doesn't support update forwarding, the result
+ # should be REFUSED.
+ sec_zone = Name("example.org")
+ msg_data, msg = create_update_msg(zones=[Question(sec_zone,
+ TEST_RRCLASS,
+ RRType.SOA())])
+ session = UpdateSession(msg, msg_data, None,
+ ZoneConfig([(sec_zone, TEST_RRCLASS)]))
+ self.assertEqual(UPDATE_DONE, session.handle())
+ self.check_response(session.get_message(), Rcode.REFUSED())
+
def test_handle(self):
self.assertEqual(UPDATE_DONE, self.__session.handle())
self.assertNotEqual(UPDATE_DROP, self.__session.handle())
until the details are fixed.
'''
+ def __init__(self, secondaries):
+ self.__secondaries = {}
+ for (zname, zclass) in secondaries:
+ self.__secondaries[(zname, zclass)] = True
+
def find_zone(self, zone_name, zone_class):
'''Return the type and accessor client object for given zone.'''
# Right now, the client is not used, so we simply return None.
+ if (zone_name, zone_class) in self.__secondaries:
+ return ZONE_SECONDARY, None
return ZONE_PRIMARY, None