import isc.log
import unittest
from isc.dns import *
-from isc.datasrc import DataSourceClient
+from isc.datasrc import DataSourceClient, ZoneFinder
from isc.ddns.session import *
from isc.ddns.zone_config import *
rrset2)
def test_update_delete_name(self):
+ '''
+ Tests whether deletion of every RR for a name works
+ '''
self.__initialize_update_rrsets()
# First check it is there
isc.dns.Name("example.org"),
RRType.MX())
- # Check that we cannot delete the SOA record by direction deletion
+ # Check that we cannot delete the SOA record by direct deletion
# both by name+type and by full rrset
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_del_soa_apex,
RRType.NS(),
orig_ns_rrset)
- def DISABLED_test_update_apex_special_case_ns_rrset(self):
+ def test_update_apex_special_case_ns_rrset(self):
# If we delete the NS at the apex specifically, it should still
# keep one record
self.__initialize_update_rrsets()
RRType.NS(),
short_ns_rrset)
+ def test_update_apex_special_case_ns_rrset2(self):
+ # If we add new NS records, then delete all existing ones, it
+ # should not keep any
+ self.__initialize_update_rrsets()
+ new_ns = create_rrset("example.org", TEST_RRCLASS, RRType.NS(), 3600,
+ [ "newns1.example.org", "newns2.example.org" ])
+
+ self.check_full_handle_result(Rcode.NOERROR(),
+ [ new_ns,
+ self.rrset_update_del_rrset_ns ])
+ self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
+ isc.dns.Name("example.org"),
+ RRType.NS(),
+ new_ns)
+
def test_update_delete_normal_rrset_at_apex(self):
'''
Tests a number of 'normal rrset' deletes at the apex
isc.dns.Name("example.org"),
RRType.MX())
+ def test_update_add_then_delete_rrset(self):
+ # If we add data, then delete the whole rrset, added data should
+ # be gone as well
+ self.__initialize_update_rrsets()
+ self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
+ isc.dns.Name("www.example.org"),
+ RRType.A())
+ self.check_full_handle_result(Rcode.NOERROR(),
+ [ self.rrset_update_a,
+ self.rrset_update_del_rrset ])
+ self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
+ isc.dns.Name("www.example.org"),
+ RRType.A())
+
+ def test_update_add_then_delete_name(self):
+ # If we add data, then delete the entire name, added data should
+ # be gone as well
+ self.__initialize_update_rrsets()
+ self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
+ isc.dns.Name("www.example.org"),
+ RRType.A())
+ self.check_full_handle_result(Rcode.NOERROR(),
+ [ self.rrset_update_a,
+ self.rrset_update_del_name ])
+ self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
+ isc.dns.Name("www.example.org"),
+ RRType.A())
+
def test_update_cname_special_cases(self):
self.__initialize_update_rrsets()
"""
import isc.dns
+from isc.datasrc import ZoneFinder
import isc.log
from isc.datasrc import ZoneFinder
from isc.log_messages.libxfrin_messages import *
str(self.__updater.get_class()))
if self.__single_update_mode:
if operation == 'add':
- self.__append_with_soa_check(self.__additions, operation, rr)
+ if not self._remove_rr_from_deletions(rr):
+ self.__append_with_soa_check(self.__additions, operation,
+ rr)
elif operation == 'delete':
- self.__append_with_soa_check(self.__deletions, operation, rr)
+ if not self._remove_rr_from_additions(rr):
+ self.__append_with_soa_check(self.__deletions, operation,
+ rr)
else:
self.__buffer.append((operation, rr))
if len(self.__buffer) >= DIFF_APPLY_TRESHOLD:
"""
self.__check_committed()
return self.__updater.find_all(name, options)
+
+ def _remove_name_from_buffer(self, buf, name):
+ return [ op for op in buf if\
+ op[1].get_name() != name or\
+ op[1].get_type() == isc.dns.RRType.SOA() ]
+
+ def _remove_name_type_from_buffer(self, buf, name, rrtype):
+ if rrtype == isc.dns.RRType.SOA():
+ return buf
+ else:
+ return [ op for op in buf if\
+ op[1].get_name() != name or\
+ op[1].get_type() != rrtype ]
+
+ def __remove_rr_from_buffer(self, buf, rr):
+ '''Helper for common code in remove_rr_from_deletions() and
+ remove_rr_from_additions();
+ returns the result of the removal operation on the given buffer
+ '''
+ def same_rr(a, b):
+ # Consider two rr's the same if name, type, and rdata match
+ # Note that at this point it should have been checked that
+ # the rr in the buffer and the given rr have exactly one rdata
+ return a.get_name() == b.get_name() and\
+ a.get_type() == b.get_type() and\
+ a.get_rdata()[0] == b.get_rdata()[0]
+ if rr.get_type() == isc.dns.RRType.SOA():
+ return buf
+ else:
+ return [ op for op in buf if not same_rr(op[1], rr)]
+
+ def _remove_rr_from_deletions(self, rr):
+ '''
+ Removes the given rr from the currently buffered deletions;
+ returns True if anything is removed, False if the RR was not present.
+ This method is protected; it is not meant to be called from anywhere
+ but the add_data() method. It is not private for easier testing.
+ '''
+ orig_size = len(self.__deletions)
+ self.__deletions = self.__remove_rr_from_buffer(self.__deletions, rr)
+ return len(self.__deletions) != orig_size
+
+ def _remove_rr_from_additions(self, rr):
+ '''
+ Removes the given rr from the currently buffered additions;
+ returns True if anything is removed, False if the RR was not present.
+ This method is protected; it is not meant to be called from anywhere
+ but the delete_data() method. It is not private for easier testing.
+ '''
+ orig_size = len(self.__additions)
+ self.__additions = self.__remove_rr_from_buffer(self.__additions, rr)
+ return len(self.__additions) != orig_size
+
+ def _get_name_from_additions(self, name):
+ '''
+ Returns a list of all rrs in the additions queue that have the given
+ Name.
+ This method is protected; it is not meant to be called from anywhere
+ but the find_all_updated() method. It is not private for easier
+ testing.
+ '''
+ return [ rr for (_, rr) in self.__additions if rr.get_name() == name ]
+
+ def _get_name_from_deletions(self, name):
+ '''
+ Returns a list of all rrs in the deletions queue that have the given
+ Name
+ This method is protected; it is not meant to be called from anywhere
+ but the find_all_updated() method. It is not private for easier
+ testing.
+ '''
+ return [ rr for (_, rr) in self.__deletions if rr.get_name() == name ]
+
+ def _get_name_type_from_additions(self, name, rrtype):
+ '''
+ Returns a list of the rdatas of the rrs in the additions queue that
+ have the given name and type
+ This method is protected; it is not meant to be called from anywhere
+ but the find_updated() method. It is not private for easier testing.
+ '''
+ return [ rr.get_rdata()[0] for (_, rr) in self.__additions if rr.get_name() == name and rr.get_type() == rrtype ]
+
+ def _get_name_type_from_deletions(self, name, rrtype):
+ '''
+ Returns a list of the rdatas of the rrs in the deletions queue that
+ have the given name and type
+ This method is protected; it is not meant to be called from anywhere
+ but the find_updated() method. It is not private for easier testing.
+ '''
+ return [ rr.get_rdata()[0] for (_, rr) in self.__deletions if rr.get_name() == name and rr.get_type() == rrtype ]
+
+ def find_updated(self, name, rrtype):
+ '''
+ Returns the result of find(), but with current updates applied, i.e.
+ as if this diff has been committed. Only works for find results
+ SUCCESS, NXDOMAIN, and NXRRSET.
+ '''
+ result, rrset, flags = self.find(name, rrtype)
+
+ # Create a new rrset object;
+ # - add all rdatas from the found rrset, unless they are in __deletions
+ # - add all rdatas from additions
+ added_rrs = self._get_name_type_from_additions(name, rrtype)
+ deleted_rrs = self._get_name_type_from_deletions(name, rrtype)
+
+ if result == ZoneFinder.SUCCESS:
+ new_rrset = isc.dns.RRset(name, self.__updater.get_class(),
+ rrtype, rrset.get_ttl())
+ for rdata in rrset.get_rdata():
+ if rdata not in deleted_rrs:
+ new_rrset.add_rdata(rdata)
+ elif (result == ZoneFinder.NXDOMAIN or result == ZoneFinder.NXRRSET)\
+ and len(added_rrs) > 0:
+ new_rrset = isc.dns.RRset(name, self.__updater.get_class(),
+ rrtype, added_rrs[0].get_ttl())
+ # There was no data in the zone, but there is data now
+ result = SUCCESS
+ else:
+ # Can't reliably handle other cases, just return the original
+ # data
+ return result, rrset, flags
+
+ for rdata in added_rrs:
+ new_rrset.add_rdata(rdata)
+
+ return result, new_rrset, flags
+
+ def find_all_updated(self, name):
+ result, rrsets, flags = self.find_all(name)
+ new_rrsets = []
+ added_rrs = self._get_name_from_additions(name)
+ if result == ZoneFinder.SUCCESS and\
+ (flags & ZoneFinder.RESULT_WILDCARD == 0):
+ deleted_rrs = self._get_name_from_deletions(name)
+ for rr in rrsets:
+ if rr not in deleted_rrs:
+ new_rrsets.append(rr)
+ elif result == ZoneFinder.NXDOMAIN and\
+ len(added_rrs) > 0:
+ result = SUCCESS
+ for rr in added_rrs:
+ if rr.get_name() == name:
+ new_rrsets.append(rr)
+ return result, new_rrsets, flags
self.__rrset_multi.add_rdata(Rdata(self.__type, self.__rrclass,
'192.0.2.2'))
+ # Also create a few other (valid) rrsets
+ # A SOA record
+ self.__rrset_soa = RRset(Name('example.org.'), self.__rrclass,
+ RRType.SOA(), RRTTL(3600))
+ self.__rrset_soa.add_rdata(Rdata(RRType.SOA(), self.__rrclass,
+ "ns1.example.org. " +
+ "admin.example.org. " +
+ "1233 3600 1800 2419200 7200"))
+ # A few single-rr rrsets that together would for a multi-rr rrset
+ self.__rrset3 = RRset(Name('c.example.org.'), self.__rrclass,
+ RRType.TXT(), self.__ttl)
+ self.__rrset3.add_rdata(Rdata(RRType.TXT(), self.__rrclass, "one"))
+ self.__rrset4 = RRset(Name('c.example.org.'), self.__rrclass,
+ RRType.TXT(), self.__ttl)
+ self.__rrset4.add_rdata(Rdata(RRType.TXT(), self.__rrclass, "two"))
+ self.__rrset5 = RRset(Name('c.example.org.'), self.__rrclass,
+ RRType.TXT(), self.__ttl)
+ self.__rrset5.add_rdata(Rdata(RRType.TXT(), self.__rrclass, "three"))
+ self.__rrset6 = RRset(Name('d.example.org.'), self.__rrclass,
+ RRType.A(), self.__ttl)
+ self.__rrset6.add_rdata(Rdata(RRType.A(), self.__rrclass, "192.0.2.1"))
+ self.__rrset7 = RRset(Name('d.example.org.'), self.__rrclass,
+ RRType.A(), self.__ttl)
+ self.__rrset7.add_rdata(Rdata(RRType.A(), self.__rrclass, "192.0.2.2"))
+
+
def __mock_compact(self):
"""
This can be put into the diff to hook into its compact method and see
and it must be the first change.
'''
- # First create some RRsets to play with
- soa = RRset(Name('example.org.'), self.__rrclass, RRType.SOA(),
- RRTTL(3600))
- soa.add_rdata(Rdata(soa.get_type(), soa.get_class(),
- "ns.example.org. foo.example.org. 1234 28800 "+
- "7200 604800 3600"))
-
- a = RRset(Name('www.example.org.'), self.__rrclass, RRType.A(),
- RRTTL(3600))
- a.add_rdata(Rdata(a.get_type(), a.get_class(),
- "192.0.2.1"))
-
- a2 = RRset(Name('www.example.org.'), self.__rrclass, RRType.A(),
- RRTTL(3600))
- a2.add_rdata(Rdata(a2.get_type(), a2.get_class(),
- "192.0.2.2"))
-
# full rrset for A (to check compact())
- a_1_2 = RRset(Name('www.example.org.'), self.__rrclass, RRType.A(),
- RRTTL(3600))
- a_1_2.add_rdata(Rdata(a_1_2.get_type(), a_1_2.get_class(),
- "192.0.2.1"))
- a_1_2.add_rdata(Rdata(a_1_2.get_type(), a_1_2.get_class(),
- "192.0.2.2"))
+ txt = RRset(Name('c.example.org.'), self.__rrclass, RRType.TXT(),
+ RRTTL(3600))
+ txt.add_rdata(Rdata(txt.get_type(), txt.get_class(), "one"))
+ txt.add_rdata(Rdata(txt.get_type(), txt.get_class(), "two"))
+ txt.add_rdata(Rdata(txt.get_type(), txt.get_class(), "three"))
+ a = RRset(Name('d.example.org.'), self.__rrclass, RRType.A(),
+ RRTTL(3600))
+ a.add_rdata(Rdata(a.get_type(), a.get_class(), "192.0.2.1"))
+ a.add_rdata(Rdata(a.get_type(), a.get_class(), "192.0.2.2"))
diff = Diff(self, Name('example.org.'), single_update_mode=True)
# adding a first should fail
self.assertRaises(ValueError, diff.add_data, a)
# But soa should work
- diff.add_data(soa)
+ diff.add_data(self.__rrset_soa)
# And then A should as well
- diff.add_data(a)
- diff.add_data(a2)
+ diff.add_data(self.__rrset3)
+ diff.add_data(self.__rrset4)
+ diff.add_data(self.__rrset5)
# But another SOA should fail again
- self.assertRaises(ValueError, diff.add_data, soa)
+ self.assertRaises(ValueError, diff.add_data, self.__rrset_soa)
# Same for delete
- self.assertRaises(ValueError, diff.delete_data, a)
- diff.delete_data(soa)
- diff.delete_data(a)
- diff.delete_data(a2)
- self.assertRaises(ValueError, diff.delete_data, soa)
+ self.assertRaises(ValueError, diff.delete_data, self.__rrset6)
+ diff.delete_data(self.__rrset_soa)
+ diff.delete_data(self.__rrset6)
+ diff.delete_data(self.__rrset7)
+ self.assertRaises(ValueError, diff.delete_data, self.__rrset_soa)
# Not compacted yet, so the buffers should be as we
# filled them
(delbuf, addbuf) = diff.get_single_update_buffers()
- self.assertEqual([('delete', soa), ('delete', a), ('delete', a2)], delbuf)
- self.assertEqual([('add', soa), ('add', a), ('add', a2)], addbuf)
+ self.assertEqual([('delete', self.__rrset_soa),
+ ('delete', self.__rrset6),
+ ('delete', self.__rrset7)], delbuf)
+ self.assertEqual([('add', self.__rrset_soa),
+ ('add', self.__rrset3),
+ ('add', self.__rrset4),
+ ('add', self.__rrset5)], addbuf)
# Compact should compact the A records in both buffers
diff.compact()
self.assertEqual(2, len(delbuf))
self.assertEqual(2, len(delbuf[0]))
self.assertEqual('delete', delbuf[0][0])
- self.assertEqual(soa.to_text(), delbuf[0][1].to_text())
+ self.assertEqual(self.__rrset_soa.to_text(), delbuf[0][1].to_text())
self.assertEqual(2, len(delbuf[1]))
self.assertEqual('delete', delbuf[1][0])
- self.assertEqual(a_1_2.to_text(), delbuf[1][1].to_text())
+ self.assertEqual(a.to_text(), delbuf[1][1].to_text())
self.assertEqual(2, len(addbuf))
self.assertEqual(2, len(addbuf[0]))
self.assertEqual('add', addbuf[0][0])
- self.assertEqual(soa.to_text(), addbuf[0][1].to_text())
+ self.assertEqual(self.__rrset_soa.to_text(), addbuf[0][1].to_text())
self.assertEqual(2, len(addbuf[1]))
self.assertEqual('add', addbuf[1][0])
- self.assertEqual(a_1_2.to_text(), addbuf[1][1].to_text())
+ self.assertEqual(txt.to_text(), addbuf[1][1].to_text())
# Apply should reset the buffers
diff.apply()
self.assertRaises(ValueError, diff.add_data, a)
self.assertRaises(ValueError, diff.delete_data, a)
+ def test_add_delete_same(self):
+ '''
+ Test that if a record is added, then deleted, it is not added to
+ both buffers, but remove from the addition, and vice versa
+ '''
+ diff = Diff(self, Name('example.org.'), single_update_mode=True)
+ # Need SOA records first
+ diff.delete_data(self.__rrset_soa)
+ diff.add_data(self.__rrset_soa)
+
+ deletions, additions = diff.get_single_update_buffers()
+ self.assertEqual(1, len(deletions))
+ self.assertEqual(1, len(additions))
+
+
+ diff.add_data(self.__rrset1)
+ deletions, additions = diff.get_single_update_buffers()
+ self.assertEqual(1, len(deletions))
+ self.assertEqual(2, len(additions))
+
+ diff.delete_data(self.__rrset1)
+ deletions, additions = diff.get_single_update_buffers()
+ self.assertEqual(1, len(deletions))
+ self.assertEqual(1, len(additions))
+
+ diff.delete_data(self.__rrset2)
+ deletions, additions = diff.get_single_update_buffers()
+ self.assertEqual(2, len(deletions))
+ self.assertEqual(1, len(additions))
+
+ diff.add_data(self.__rrset2)
+ deletions, additions = diff.get_single_update_buffers()
+ self.assertEqual(1, len(deletions))
+ self.assertEqual(1, len(additions))
+
+
def test_find(self):
diff = Diff(self, Name('example.org.'))
name = Name('www.example.org.')
self.assertEqual(name, self.__find_all_name)
self.assertEqual(options, self.__find_all_options)
+ def __common_remove_rr_from_buffer(self, diff, add_method, remove_method,
+ op_str, buf_nr):
+ add_method(self.__rrset_soa)
+ add_method(self.__rrset2)
+ add_method(self.__rrset3)
+ add_method(self.__rrset4)
+
+ # sanity check
+ buf = diff.get_single_update_buffers()[buf_nr]
+ expected = [ (op_str, str(rr)) for rr in [ self.__rrset_soa,
+ self.__rrset2,
+ self.__rrset3,
+ self.__rrset4 ] ]
+ result = [ (op, str(rr)) for (op, rr) in buf ]
+ self.assertEqual(expected, result)
+
+ # remove one
+ self.assertTrue(remove_method(self.__rrset2))
+ buf = diff.get_single_update_buffers()[buf_nr]
+ expected = [ (op_str, str(rr)) for rr in [ self.__rrset_soa,
+ self.__rrset3,
+ self.__rrset4 ] ]
+ result = [ (op, str(rr)) for (op, rr) in buf ]
+ self.assertEqual(expected, result)
+
+ # SOA should not be removed
+ self.assertFalse(remove_method(self.__rrset_soa))
+ buf = diff.get_single_update_buffers()[buf_nr]
+ expected = [ (op_str, str(rr)) for rr in [ self.__rrset_soa,
+ self.__rrset3,
+ self.__rrset4 ] ]
+ result = [ (op, str(rr)) for (op, rr) in buf ]
+ self.assertEqual(expected, result)
+
+ # remove another
+ self.assertTrue(remove_method(self.__rrset4))
+ buf = diff.get_single_update_buffers()[buf_nr]
+ expected = [ (op_str, str(rr)) for rr in [ self.__rrset_soa,
+ self.__rrset3 ] ]
+ result = [ (op, str(rr)) for (op, rr) in buf ]
+ self.assertEqual(expected, result)
+
+ # remove nonexistent should return False
+ self.assertFalse(remove_method(self.__rrset4))
+ buf = diff.get_single_update_buffers()[buf_nr]
+ expected = [ (op_str, str(rr)) for rr in [ self.__rrset_soa,
+ self.__rrset3 ] ]
+ result = [ (op, str(rr)) for (op, rr) in buf ]
+ self.assertEqual(expected, result)
+
+ def test_remove_rr_from_additions(self):
+ diff = Diff(self, Name('example.org'), single_update_mode=True)
+ self.__common_remove_rr_from_buffer(diff, diff.add_data,
+ diff._remove_rr_from_additions,
+ 'add', 1)
+
+ def test_remove_rr_from_deletions(self):
+ diff = Diff(self, Name('example.org'), single_update_mode=True)
+ self.__common_remove_rr_from_buffer(diff, diff.delete_data,
+ diff._remove_rr_from_deletions,
+ 'delete', 0)
+
if __name__ == "__main__":
isc.log.init("bind10")
isc.log.resetUnitTestRootLogger()
projects / thirdparty / kea.git / commitdiff
