new_rrset.add_rdata(isc.dns.Rdata(rrset.get_type(), rrclass, wire))
return new_rrset
-class DDNSSOA:
+class DDNS_SOA:
'''Class to handle the SOA in the DNS update '''
def __get_serial_internal(self, origin_soa):
- '''get serial number from soa'''
+ '''Get serial number from soa'''
return Serial(int(origin_soa.get_rdata()[0].to_text().split()[2]))
def __write_soa_internal(self, origin_soa, soa_num):
- '''write serial number to soa'''
+ '''Write back serial number to soa'''
new_soa = RRset(origin_soa.get_name(), origin_soa.get_class(),
RRType.SOA(), origin_soa.get_ttl())
- new_content = ""
- index = 0
- for i in origin_soa.get_rdata()[0].to_text().split():
- if(index == 2):
- # the second item is serial number
- new_content = "%s %d"%(new_content, soa_num.get_value())
- else:
- new_content = "%s %s"%(new_content, i)
- index += 1
+ soa_rdata_parts = origin_soa.get_rdata()[0].to_text().split()
+ soa_rdata_parts[2] = str(soa_num.get_value())
new_soa.add_rdata(Rdata(origin_soa.get_type(), origin_soa.get_class(),
- new_content))
+ " ".join(soa_rdata_parts)))
return new_soa
def soa_update_check(self, origin_soa, new_soa):
+ '''Check whether the new soa is valid. If the serial number is bigger
+ than the old one, it is valid, then return True, otherwise, return
+ False.
+ Parameters:
+ origin_soa, old SOA resource record.
+ new_soa, new SOA resource record.
+ Output:
+ if the serial number of new soa is bigger than the old one, return
+ True, otherwise return False.
+ '''
old_serial = self.__get_serial_internal(origin_soa)
new_serial = self.__get_serial_internal(new_soa)
- if(new_serial >= old_serial):
+ if(new_serial > old_serial):
return True
else:
return False
def update_soa(self, origin_soa, inc_number = 1):
+ ''' Update the soa number incrementally as RFC 2136.
+ Parameters:
+ origin_soa, the soa resource record which will be updated.
+ inc_number, the number which will be added into the serial number of
+ origin_soa, the default value is one.
+ Output:
+ the new origin soa whoes serial number has been updated.
+ '''
soa_num = self.__get_serial_internal(origin_soa)
soa_num = soa_num + inc_number
if soa_num.get_value() == 0:
result, old_soa, _ = self.__finder.find(self.__zname, RRType.SOA(),
ZoneFinder.NO_WILDCARD |
ZoneFinder.FIND_GLUE_OK)
- serial_operation = DDNSSOA()
+ serial_operation = DDNS_SOA()
if self.__added_soa is not None:
# serial check goes here
if serial_operation.soa_update_check(old_soa, self.__added_soa):
rdata))
class TestDDNSSOA(unittest.TestCase):
- '''unittest for the DDNSSOA'''
+ '''unittest for the DDNS_SOA'''
def test_update_soa(self):
'''unittest for update_soa function'''
- soa_update = DDNSSOA()
+ soa_update = DDNS_SOA()
soa_rr = create_rrset("example.org", TEST_RRCLASS,
RRType.SOA(), 3600, ["ns1.example.org. " +
"admin.example.org. " +
3600, ["ns1.example.org. " +
"admin.example.org. " +
"1234 3600 1800 2419200 7200"])
- soa_update = DDNSSOA()
+ soa_update = DDNS_SOA()
# The case of (i1 < i2 and i2 - i1 < 2^(SERIAL_BITS - 1)) in rfc 1982
self.assertTrue(soa_update.soa_update_check(small_soa_rr,
large_soa_rr))