if identifier == "zones/class":
return TEST_RRCLASS_STR
- self.diffs.append(('remove', rrset))
+ class MockDataSourceClient():
+ '''A simple mock data source client.
+
+ This class provides a minimal set of wrappers related the data source
+ API that would be used by Diff objects. For our testing purposes they
+ only keep truck of the history of the changes.
+
+ '''
+ def __init__(self):
+ self.committed_diffs = []
+ self.diffs = []
+
++ def get_class(self):
++ '''Mock version of get_class().
++
++ We simply return the commonly used constant RR class. If and when
++ we use this mock for a different RR class we need to adjust it
++ accordingly.
++
++ '''
++ return TEST_RRCLASS
++
+ def find_zone(self, zone_name):
+ '''Mock version of find_zone().
+
+ It returns itself (subsequently acting as a mock ZoneFinder) for
+ some test zone names. For some others it returns either NOTFOUND
+ or PARTIALMATCH.
+
+ '''
+ if zone_name == TEST_ZONE_NAME or \
+ zone_name == Name('no-soa.example') or \
+ zone_name == Name('dup-soa.example'):
+ return (isc.datasrc.DataSourceClient.SUCCESS, self)
+ elif zone_name == Name('no-such-zone.example'):
+ return (DataSourceClient.NOTFOUND, None)
+ elif zone_name == Name('partial-match-zone.example'):
+ return (DataSourceClient.PARTIALMATCH, self)
+ raise ValueError('Unexpected input to mock client: bug in test case?')
+
+ def find(self, name, rrtype, target, options):
+ '''Mock ZoneFinder.find().
+
+ It returns the predefined SOA RRset to queries for SOA of the common
+ test zone name. It also emulates some unusual cases for special
+ zone names.
+
+ '''
+ if name == TEST_ZONE_NAME and rrtype == RRType.SOA():
+ return (ZoneFinder.SUCCESS, begin_soa_rrset)
+ if name == Name('no-soa.example'):
+ return (ZoneFinder.NXDOMAIN, None)
+ if name == Name('dup-soa.example'):
+ dup_soa_rrset = RRset(name, TEST_RRCLASS, RRType.SOA(), RRTTL(0))
+ dup_soa_rrset.add_rdata(begin_soa_rdata)
+ dup_soa_rrset.add_rdata(soa_rdata)
+ return (ZoneFinder.SUCCESS, dup_soa_rrset)
+ raise ValueError('Unexpected input to mock finder: bug in test case?')
+
+ def get_updater(self, zone_name, replace):
+ return self
+
+ def add_rrset(self, rrset):
+ self.diffs.append(('add', rrset))
+
+ def delete_rrset(self, rrset):
++ self.diffs.append(('delete', rrset))
+
+ def commit(self):
+ self.committed_diffs.append(self.diffs)
+ self.diffs = []
+
class MockXfrin(Xfrin):
# This is a class attribute of a callable object that specifies a non
# default behavior triggered in _cc_check_command(). Specific test methods
return reply_data
- self.assertEqual([('remove', self.begin_soa)],
+ class TestXfrinState(unittest.TestCase):
+ def setUp(self):
+ self.sock_map = {}
+ self.conn = MockXfrinConnection(self.sock_map, TEST_ZONE_NAME,
+ TEST_RRCLASS, TEST_DB_FILE,
+ threading.Event(),
+ TEST_MASTER_IPV4_ADDRINFO)
+ self.begin_soa = RRset(TEST_ZONE_NAME, TEST_RRCLASS, RRType.SOA(),
+ RRTTL(3600))
+ self.begin_soa.add_rdata(Rdata(RRType.SOA(), TEST_RRCLASS,
+ 'm. r. 1230 0 0 0 0'))
+ self.ns_rrset = RRset(TEST_ZONE_NAME, TEST_RRCLASS, RRType.NS(),
+ RRTTL(3600))
+ self.ns_rrset.add_rdata(Rdata(RRType.NS(), TEST_RRCLASS,
+ 'ns.example.com'))
+ self.conn._datasrc_client = MockDataSourceClient()
+ self.conn._diff = Diff(MockDataSourceClient(), TEST_ZONE_NAME)
+
+ class TestXfrinStateBase(TestXfrinState):
+ def setUp(self):
+ super().setUp()
+
+ def test_handle_rr_on_base(self):
+ # The base version of handle_rr() isn't supposed to be called
+ # directly (the argument doesn't matter in this test)
+ self.assertRaises(XfrinException, XfrinState().handle_rr, None)
+
+ class TestXfrinInitialSOA(TestXfrinState):
+ def setUp(self):
+ super().setUp()
+ self.state = XfrinInitialSOA()
+
+ def test_handle_rr(self):
+ # normal case
+ self.assertTrue(self.state.handle_rr(self.conn, soa_rrset))
+ self.assertEqual(type(XfrinFirstData()),
+ type(self.conn.get_xfrstate()))
+ self.assertEqual(1234, self.conn._end_serial)
+
+ def test_handle_not_soa(self):
+ # The given RR is not of SOA
+ self.assertRaises(XfrinProtocolError, self.state.handle_rr, self.conn,
+ self.ns_rrset)
+
+ def test_finish_message(self):
+ self.assertTrue(self.state.finish_message(self.conn))
+
+ class TestXfrinFirstData(TestXfrinState):
+ def setUp(self):
+ super().setUp()
+ self.state = XfrinFirstData()
+ self.conn._request_type = RRType.IXFR()
+ self.conn._request_serial = 1230 # arbitrary chosen serial < 1234
+
+ def test_handle_ixfr_begin_soa(self):
+ self.conn._request_type = RRType.IXFR()
+ self.assertFalse(self.state.handle_rr(self.conn, self.begin_soa))
+ self.assertEqual(type(XfrinIXFRDeleteSOA()),
+ type(self.conn.get_xfrstate()))
+
+ def test_handle_axfr(self):
+ # If the original type is AXFR, other conditions aren't considered,
+ # and AXFR processing will continue
+ self.conn._request_type = RRType.AXFR()
+ self.assertFalse(self.state.handle_rr(self.conn, self.begin_soa))
+ self.assertEqual(type(XfrinAXFR()), type(self.conn.get_xfrstate()))
+
+ def test_handle_ixfr_to_axfr(self):
+ # Detecting AXFR-compatible IXFR response by seeing a non SOA RR after
+ # the initial SOA. Should switch to AXFR.
+ self.assertFalse(self.state.handle_rr(self.conn, self.ns_rrset))
+ self.assertEqual(type(XfrinAXFR()), type(self.conn.get_xfrstate()))
+
+ def test_handle_ixfr_to_axfr_by_different_soa(self):
+ # An unusual case: Response contains two consecutive SOA but the
+ # serial of the second does not match the requested one. See
+ # the documentation for XfrinFirstData.handle_rr().
+ self.assertFalse(self.state.handle_rr(self.conn, soa_rrset))
+ self.assertEqual(type(XfrinAXFR()), type(self.conn.get_xfrstate()))
+
+ def test_finish_message(self):
+ self.assertTrue(self.state.finish_message(self.conn))
+
+ class TestXfrinIXFRDeleteSOA(TestXfrinState):
+ def setUp(self):
+ super().setUp()
+ self.state = XfrinIXFRDeleteSOA()
+ # In this state a new Diff object is expected to be created. To
+ # confirm it, we nullify it beforehand.
+ self.conn._diff = None
+
+ def test_handle_rr(self):
+ self.assertTrue(self.state.handle_rr(self.conn, self.begin_soa))
+ self.assertEqual(type(XfrinIXFRDelete()),
+ type(self.conn.get_xfrstate()))
- # Non SOA RRs are simply (goting to be) removed in this state
++ self.assertEqual([('delete', self.begin_soa)],
+ self.conn._diff.get_buffer())
+
+ def test_handle_non_soa(self):
+ self.assertRaises(XfrinException, self.state.handle_rr, self.conn,
+ self.ns_rrset)
+
+ def test_finish_message(self):
+ self.assertTrue(self.state.finish_message(self.conn))
+
+ class TestXfrinIXFRDelete(TestXfrinState):
+ def setUp(self):
+ super().setUp()
+ # We need record the state in 'conn' to check the case where the
+ # state doesn't change.
+ XfrinIXFRDelete().set_xfrstate(self.conn, XfrinIXFRDelete())
+ self.state = self.conn.get_xfrstate()
+
+ def test_handle_delete_rr(self):
- self.assertEqual([('remove', self.ns_rrset)],
++ # Non SOA RRs are simply (goting to be) deleted in this state
+ self.assertTrue(self.state.handle_rr(self.conn, self.ns_rrset))
++ self.assertEqual([('delete', self.ns_rrset)],
+ self.conn._diff.get_buffer())
+ # The state shouldn't change
+ self.assertEqual(type(XfrinIXFRDelete()),
+ type(self.conn.get_xfrstate()))
+
+ def test_handle_soa(self):
+ # SOA in this state means the beginning of added RRs. This SOA
+ # should also be added in the next state, so handle_rr() should return
+ # false.
+ self.assertFalse(self.state.handle_rr(self.conn, soa_rrset))
+ self.assertEqual([], self.conn._diff.get_buffer())
+ self.assertEqual(1234, self.conn._current_serial)
+ self.assertEqual(type(XfrinIXFRAddSOA()),
+ type(self.conn.get_xfrstate()))
+
+ def test_finish_message(self):
+ self.assertTrue(self.state.finish_message(self.conn))
+
+ class TestXfrinIXFRAddSOA(TestXfrinState):
+ def setUp(self):
+ super().setUp()
+ self.state = XfrinIXFRAddSOA()
+
+ def test_handle_rr(self):
+ self.assertTrue(self.state.handle_rr(self.conn, soa_rrset))
+ self.assertEqual(type(XfrinIXFRAdd()), type(self.conn.get_xfrstate()))
+ self.assertEqual([('add', soa_rrset)],
+ self.conn._diff.get_buffer())
+
+ def test_handle_non_soa(self):
+ self.assertRaises(XfrinException, self.state.handle_rr, self.conn,
+ self.ns_rrset)
+
+ def test_finish_message(self):
+ self.assertTrue(self.state.finish_message(self.conn))
+
+ class TestXfrinIXFRAdd(TestXfrinState):
+ def setUp(self):
+ super().setUp()
+ # We need record the state in 'conn' to check the case where the
+ # state doesn't change.
+ XfrinIXFRAdd().set_xfrstate(self.conn, XfrinIXFRAdd())
+ self.conn._current_serial = 1230
+ self.state = self.conn.get_xfrstate()
+
+ def test_handle_add_rr(self):
+ # Non SOA RRs are simply (goting to be) added in this state
+ self.assertTrue(self.state.handle_rr(self.conn, self.ns_rrset))
+ self.assertEqual([('add', self.ns_rrset)],
+ self.conn._diff.get_buffer())
+ # The state shouldn't change
+ self.assertEqual(type(XfrinIXFRAdd()), type(self.conn.get_xfrstate()))
+
+ def test_handle_end_soa(self):
+ self.conn._end_serial = 1234
+ self.conn._diff.add_data(self.ns_rrset) # put some dummy change
+ self.assertTrue(self.state.handle_rr(self.conn, soa_rrset))
+ self.assertEqual(type(XfrinIXFREnd()), type(self.conn.get_xfrstate()))
+ # handle_rr should have caused commit, and the buffer should now be
+ # empty.
+ self.assertEqual([], self.conn._diff.get_buffer())
+
+ def test_handle_new_delete(self):
+ self.conn._end_serial = 1234
+ # SOA RR whose serial is the current one means we are going to a new
+ # difference, starting with removing that SOA.
+ self.conn._diff.add_data(self.ns_rrset) # put some dummy change
+ self.assertFalse(self.state.handle_rr(self.conn, self.begin_soa))
+ self.assertEqual([], self.conn._diff.get_buffer())
+ self.assertEqual(type(XfrinIXFRDeleteSOA()),
+ type(self.conn.get_xfrstate()))
+
+ def test_handle_out_of_sync(self):
+ # getting SOA with an inconsistent serial. This is an error.
+ self.conn._end_serial = 1235
+ self.assertRaises(XfrinProtocolError, self.state.handle_rr,
+ self.conn, soa_rrset)
+
+ def test_finish_message(self):
+ self.assertTrue(self.state.finish_message(self.conn))
+
+ class TestXfrinIXFREnd(TestXfrinState):
+ def setUp(self):
+ super().setUp()
+ self.state = XfrinIXFREnd()
+
+ def test_handle_rr(self):
+ self.assertRaises(XfrinProtocolError, self.state.handle_rr, self.conn,
+ self.ns_rrset)
+
+ def test_finish_message(self):
+ self.assertFalse(self.state.finish_message(self.conn))
+
+ class TestXfrinAXFR(TestXfrinState):
+ def setUp(self):
+ super().setUp()
+ self.state = XfrinAXFR()
+
+ def test_handle_rr(self):
+ self.assertRaises(XfrinException, self.state.handle_rr, self.conn,
+ soa_rrset)
+
+ def test_finish_message(self):
+ self.assertTrue(self.state.finish_message(self.conn))
+
class TestXfrinConnection(unittest.TestCase):
+ '''Convenient parent class for XFR-protocol tests.
+
+ This class provides common setups and helper methods for protocol related
+ tests on AXFR and IXFR.
+
+ '''
+
def setUp(self):
if os.path.exists(TEST_DB_FILE):
os.remove(TEST_DB_FILE)
self.conn.response_generator = self._create_soa_response_data
self.assertEqual(self.conn.do_xfrin(True), XFRIN_FAIL)
- def _handle_xfrin_response(self):
- # This helper methods iterates over all RRs (excluding the ending SOA)
- # transferred, and simply returns the number of RRs. The return value
- # may be used an assertion value for test cases.
- rrs = 0
- for rr in self.conn._handle_xfrin_response():
- rrs += 1
- return rrs
-
- def _create_normal_response_data(self):
- # This helper method creates a simple sequence of DNS messages that
- # forms a valid XFR transaction. It consists of two messages, each
- # containing just a single SOA RR.
- tsig_1st = self.axfr_response_params['tsig_1st']
- tsig_2nd = self.axfr_response_params['tsig_2nd']
- self.conn.reply_data = self.conn.create_response_data(tsig_ctx=tsig_1st)
- self.conn.reply_data += \
- self.conn.create_response_data(tsig_ctx=tsig_2nd)
+ class TestIXFRResponse(TestXfrinConnection):
+ def setUp(self):
+ super().setUp()
+ self.conn._query_id = self.conn.qid = 1035
+ self.conn._request_serial = 1230
+ self.conn._request_type = RRType.IXFR()
+ self._zone_name = TEST_ZONE_NAME
+ self.conn._datasrc_client = MockDataSourceClient()
+ XfrinInitialSOA().set_xfrstate(self.conn, XfrinInitialSOA())
- def _create_soa_response_data(self):
- # This helper method creates a DNS message that is supposed to be
- # used a valid response to SOA queries prior to XFR.
- # If tsig is True, it tries to verify the query with a locally
- # created TSIG context (which may or may not succeed) so that the
- # response will include a TSIG.
- # If axfr_after_soa is True, it resets the response_generator so that
- # a valid XFR messages will follow.
+ def test_ixfr_response(self):
+ '''A simplest form of IXFR response.
- verify_ctx = None
- if self.soa_response_params['tsig']:
- # xfrin (curreently) always uses TCP. strip off the length field.
- query_data = self.conn.query_data[2:]
- query_message = Message(Message.PARSE)
- query_message.from_wire(query_data)
- verify_ctx = TSIGContext(TSIG_KEY)
- verify_ctx.verify(query_message.get_tsig_record(), query_data)
+ It simply updates the zone's SOA one time.
+ '''
self.conn.reply_data = self.conn.create_response_data(
- bad_qid=self.soa_response_params['bad_qid'],
- response=self.soa_response_params['response'],
- rcode=self.soa_response_params['rcode'],
- questions=self.soa_response_params['questions'],
- tsig_ctx=verify_ctx)
- if self.soa_response_params['axfr_after_soa'] != None:
- self.conn.response_generator = \
- self.soa_response_params['axfr_after_soa']
+ questions=[Question(TEST_ZONE_NAME, TEST_RRCLASS, RRType.IXFR())],
+ answers=[soa_rrset, begin_soa_rrset, soa_rrset, soa_rrset])
+ self.conn._handle_xfrin_responses()
+ self.assertEqual(type(XfrinIXFREnd()), type(self.conn.get_xfrstate()))
+ self.assertEqual([], self.conn._datasrc_client.diffs)
- self.check_diffs([[('remove', begin_soa_rrset), ('add', soa_rrset)]],
++ self.check_diffs([[('delete', begin_soa_rrset), ('add', soa_rrset)]],
+ self.conn._datasrc_client.committed_diffs)
+
+ def test_ixfr_response_multi_sequences(self):
+ '''Similar to the previous case, but with multiple diff seqs.
+
+ '''
+ self.conn.reply_data = self.conn.create_response_data(
+ questions=[Question(TEST_ZONE_NAME, TEST_RRCLASS, RRType.IXFR())],
+ answers=[soa_rrset,
+ # removing one A in serial 1230
+ begin_soa_rrset, self._create_a('192.0.2.1'),
+ # adding one A in serial 1231
+ self._create_soa('1231'), self._create_a('192.0.2.2'),
+ # removing one A in serial 1231
+ self._create_soa('1231'), self._create_a('192.0.2.3'),
+ # adding one A in serial 1232
+ self._create_soa('1232'), self._create_a('192.0.2.4'),
+ # removing one A in serial 1232
+ self._create_soa('1232'), self._create_a('192.0.2.5'),
+ # adding one A in serial 1234
+ soa_rrset, self._create_a('192.0.2.6'),
+ soa_rrset])
+ self.conn._handle_xfrin_responses()
+ self.assertEqual(type(XfrinIXFREnd()), type(self.conn.get_xfrstate()))
+ self.assertEqual([], self.conn._datasrc_client.diffs)
- self.check_diffs([[('remove', begin_soa_rrset),
- ('remove', self._create_a('192.0.2.1')),
++ self.check_diffs([[('delete', begin_soa_rrset),
++ ('delete', self._create_a('192.0.2.1')),
+ ('add', self._create_soa('1231')),
+ ('add', self._create_a('192.0.2.2'))],
- [('remove', self._create_soa('1231')),
- ('remove', self._create_a('192.0.2.3')),
++ [('delete', self._create_soa('1231')),
++ ('delete', self._create_a('192.0.2.3')),
+ ('add', self._create_soa('1232')),
+ ('add', self._create_a('192.0.2.4'))],
- [('remove', self._create_soa('1232')),
- ('remove', self._create_a('192.0.2.5')),
++ [('delete', self._create_soa('1232')),
++ ('delete', self._create_a('192.0.2.5')),
+ ('add', soa_rrset),
+ ('add', self._create_a('192.0.2.6'))]],
+ self.conn._datasrc_client.committed_diffs)
+
+ def test_ixfr_response_multi_messages(self):
+ '''Similar to the first case, but RRs span over multiple messages.
+
+ '''
+ self.conn.reply_data = self.conn.create_response_data(
+ questions=[Question(TEST_ZONE_NAME, TEST_RRCLASS, RRType.IXFR())],
+ answers=[soa_rrset, begin_soa_rrset, soa_rrset])
+ self.conn.reply_data += self.conn.create_response_data(
+ questions=[Question(TEST_ZONE_NAME, TEST_RRCLASS, RRType.IXFR())],
+ answers=[soa_rrset])
+ self.conn._handle_xfrin_responses()
+ self.assertEqual(type(XfrinIXFREnd()), type(self.conn.get_xfrstate()))
- self.check_diffs([[('remove', begin_soa_rrset), ('add', soa_rrset)]],
++ self.check_diffs([[('delete', begin_soa_rrset), ('add', soa_rrset)]],
+ self.conn._datasrc_client.committed_diffs)
+
+ def test_ixfr_response_broken(self):
+ '''Test with a broken response.
+
+ '''
+ # SOA sequence is out-of-sync
+ self.conn.reply_data = self.conn.create_response_data(
+ questions=[Question(TEST_ZONE_NAME, TEST_RRCLASS, RRType.IXFR())],
+ answers=[soa_rrset, begin_soa_rrset, soa_rrset,
+ self._create_soa('1235')])
+ self.assertRaises(XfrinProtocolError,
+ self.conn._handle_xfrin_responses)
+ # no diffs should have been committed
+ self.check_diffs([], self.conn._datasrc_client.committed_diffs)
+
+ def test_ixfr_response_extra(self):
+ '''Test with an extra RR after the end of IXFR diff sequences.
+
+ IXFR should be rejected, but complete diff sequences should be
+ committed; it's not clear whether it's compliant to the protocol
+ specification, but it is how BIND 9 works and we do the same.
+ '''
+ self.conn.reply_data = self.conn.create_response_data(
+ questions=[Question(TEST_ZONE_NAME, TEST_RRCLASS, RRType.IXFR())],
+ answers=[soa_rrset, begin_soa_rrset, soa_rrset, soa_rrset,
+ self._create_a('192.0.2.1')])
+ self.assertRaises(XfrinProtocolError,
+ self.conn._handle_xfrin_responses)
- self.check_diffs([[('remove', begin_soa_rrset), ('add', soa_rrset)]],
++ self.check_diffs([[('delete', begin_soa_rrset), ('add', soa_rrset)]],
+ self.conn._datasrc_client.committed_diffs)
+
+ class TestIXFRSession(TestXfrinConnection):
+ '''Tests for a full IXFR session (query and response).
+
+ Detailed corner cases should have been covered in test_create_query()
+ and TestIXFRResponse, so we'll only check some typical cases to confirm
+ the general logic flow.
+ '''
+ def setUp(self):
+ super().setUp()
- def _create_broken_response_data(self):
- # This helper method creates a bogus "DNS message" that only contains
- # 4 octets of data. The DNS message parser will raise an exception.
- bogus_data = b'xxxx'
- self.conn.reply_data = struct.pack('H', socket.htons(len(bogus_data)))
- self.conn.reply_data += bogus_data
+ def test_do_xfrin(self):
+ def create_ixfr_response():
+ self.conn.reply_data = self.conn.create_response_data(
+ questions=[Question(TEST_ZONE_NAME, TEST_RRCLASS,
+ RRType.IXFR())],
+ answers=[soa_rrset, begin_soa_rrset, soa_rrset, soa_rrset])
+ self.conn.response_generator = create_ixfr_response
+ self.assertEqual(XFRIN_OK, self.conn.do_xfrin(False, RRType.IXFR()))
+
+ # Check some details of the IXFR protocol processing
+ self.assertEqual(type(XfrinIXFREnd()), type(self.conn.get_xfrstate()))
- self.check_diffs([[('remove', begin_soa_rrset), ('add', soa_rrset)]],
++ self.check_diffs([[('delete', begin_soa_rrset), ('add', soa_rrset)]],
+ self.conn._datasrc_client.committed_diffs)
+
+ # Check if the query was IXFR.
+ qdata = self.conn.query_data[2:]
+ qmsg = Message(Message.PARSE)
+ qmsg.from_wire(qdata, len(qdata))
+ self.assertEqual(1, qmsg.get_rr_count(Message.SECTION_QUESTION))
+ self.assertEqual(TEST_ZONE_NAME, qmsg.get_question()[0].get_name())
+ self.assertEqual(RRType.IXFR(), qmsg.get_question()[0].get_type())
+
+ def test_do_xfrin_fail(self):
+ '''IXFR fails due to a protocol error.
+
+ '''
+ def create_ixfr_response():
+ self.conn.reply_data = self.conn.create_response_data(
+ questions=[Question(TEST_ZONE_NAME, TEST_RRCLASS,
+ RRType.IXFR())],
+ answers=[soa_rrset, begin_soa_rrset, soa_rrset,
+ self._create_soa('1235')])
+ self.conn.response_generator = create_ixfr_response
+ self.assertEqual(XFRIN_FAIL, self.conn.do_xfrin(False, RRType.IXFR()))
+
+ def test_do_xfrin_fail(self):
+ '''IXFR fails due to a bogus DNS message.
+
+ '''
+ self._create_broken_response_data()
+ self.assertEqual(XFRIN_FAIL, self.conn.do_xfrin(False, RRType.IXFR()))
+
+ class TestIXFRSessionWithSQLite3(TestXfrinConnection):
+ '''Tests for IXFR sessions using an SQLite3 DB.
+
+ These are provided mainly to confirm the implementation actually works
+ in an environment closer to actual operational environments. So we
+ only check a few common cases; other details are tested using mock
+ data sources.
+
+ '''
+ def setUp(self):
+ self.sqlite3db_src = TESTDATA_SRCDIR + '/example.com.sqlite3'
+ self.sqlite3db_obj = TESTDATA_OBJDIR + '/example.com.sqlite3.copy'
+ super().setUp()
+ if os.path.exists(self.sqlite3db_obj):
+ os.unlink(self.sqlite3db_obj)
+ shutil.copyfile(self.sqlite3db_src, self.sqlite3db_obj)
+ self.conn._datasrc_client = DataSourceClient(self.sqlite3db_obj)
+
+ def tearDown(self):
+ if os.path.exists(self.sqlite3db_obj):
+ os.unlink(self.sqlite3db_obj)
+
+ def get_zone_serial(self):
+ result, finder = self.conn._datasrc_client.find_zone(TEST_ZONE_NAME)
+ self.assertEqual(DataSourceClient.SUCCESS, result)
+ result, soa = finder.find(TEST_ZONE_NAME, RRType.SOA(),
+ None, ZoneFinder.FIND_DEFAULT)
+ self.assertEqual(ZoneFinder.SUCCESS, result)
+ self.assertEqual(1, soa.get_rdata_count())
+ return get_soa_serial(soa.get_rdata()[0])
+
+ def test_do_xfrin_sqlite3(self):
+ def create_ixfr_response():
+ self.conn.reply_data = self.conn.create_response_data(
+ questions=[Question(TEST_ZONE_NAME, TEST_RRCLASS,
+ RRType.IXFR())],
+ answers=[soa_rrset, begin_soa_rrset, soa_rrset, soa_rrset])
+ self.conn.response_generator = create_ixfr_response
+
+ # Confirm xfrin succeeds and SOA is updated
+ self.assertEqual(1230, self.get_zone_serial())
+ self.assertEqual(XFRIN_OK, self.conn.do_xfrin(False, RRType.IXFR()))
+ self.assertEqual(1234, self.get_zone_serial())
+
+ def test_do_xfrin_sqlite3_fail(self):
+ '''Similar to the previous test, but xfrin fails due to error.
+
+ Check the DB is not changed.
+
+ '''
+ def create_ixfr_response():
+ self.conn.reply_data = self.conn.create_response_data(
+ questions=[Question(TEST_ZONE_NAME, TEST_RRCLASS,
+ RRType.IXFR())],
+ answers=[soa_rrset, begin_soa_rrset, soa_rrset,
+ self._create_soa('1235')])
+ self.conn.response_generator = create_ixfr_response
+
+ self.assertEqual(1230, self.get_zone_serial())
+ self.assertEqual(XFRIN_FAIL, self.conn.do_xfrin(False, RRType.IXFR()))
+ self.assertEqual(1230, self.get_zone_serial())
class TestXfrinRecorder(unittest.TestCase):
def setUp(self):
This one is part of pretending to be a zone updater. It writes down
removal of an rrset was requested.
"""
-- self.__data_operations.append(('remove', rrset))
++ self.__data_operations.append(('delete', rrset))
+
+ def get_class(self):
+ """
+ This one is part of pretending to be a zone updater. It returns
+ the IN class.
+ """
+ return self.__rrclass
def get_updater(self, zone_name, replace):
"""
self.assertRaises(NoSuchZone, Diff, self, Name('none.example.org.'))
self.assertTrue(self.__updater_requested)
- def __data_common(self, diff, method, name):
+ def __data_common(self, diff, method, operation):
"""
-- Common part of test for test_add and test_remove.
++ Common part of test for test_add and test_delte.
"""
# Try putting there the bad data first
self.assertRaises(ValueError, method, self.__rrset_empty)
diff = Diff(self, Name('example.org.'))
self.__data_common(diff, diff.add_data, 'add')
-- def test_remove(self):
++ def test_delete(self):
"""
Try scheduling removal of few items into the diff and see they are
stored in there.
Also try passing an rrset that has different amount of RRs than 1.
"""
diff = Diff(self, Name('example.org.'))
- self.__data_common(diff, diff.remove_data, 'remove')
- self.__data_common(diff, diff.delete_data, 'remove')
++ self.__data_common(diff, diff.delete_data, 'delete')
def test_apply(self):
"""
# Prepare the diff
diff = Diff(self, Name('example.org.'))
diff.add_data(self.__rrset1)
- diff.remove_data(self.__rrset2)
- dlist = [('add', self.__rrset1), ('remove', self.__rrset2)]
+ diff.delete_data(self.__rrset2)
- dlist = [('add', self.__rrset1), ('remove', self.__rrset2)]
++ dlist = [('add', self.__rrset1), ('delete', self.__rrset2)]
self.assertEqual(dlist, diff.get_buffer())
# Do the apply, hook the compact method
diff.compact = self.__mock_compact
# Reset the buffer by calling the original apply.
orig_apply()
self.assertEqual([], diff.get_buffer())
-- # Similar with remove
++ # Similar with delete
self.__apply_called = False
for i in range(0, 99):
- diff.remove_data(self.__rrset2)
- expected = [('remove', self.__rrset2)] * 99
+ diff.delete_data(self.__rrset2)
- expected = [('remove', self.__rrset2)] * 99
++ expected = [('delete', self.__rrset2)] * 99
self.assertEqual(expected, diff.get_buffer())
self.assertFalse(self.__apply_called)
- diff.remove_data(self.__rrset2)
+ diff.delete_data(self.__rrset2)
self.assertTrue(self.__apply_called)
def test_compact(self):
# Different type.
('add', 'a', 'AAAA', ['2001:db8::1', '2001:db8::2']),
# Different operation
-- ('remove', 'a', 'AAAA', ['2001:db8::3']),
++ ('delete', 'a', 'AAAA', ['2001:db8::3']),
# Different domain
-- ('remove', 'b', 'AAAA', ['2001:db8::4']),
++ ('delete', 'b', 'AAAA', ['2001:db8::4']),
# This does not get merged with the first, even if logically
# possible. We just don't do this.
('add', 'a', 'A', ['192.0.2.3'])
diff.compact()
check()
- self.assertRaises(ValueError, diff.remove_data, rrset)
+ def test_wrong_class(self):
+ """
+ Test a wrong class of rrset is rejected.
+ """
+ diff = Diff(self, Name('example.org.'))
+ rrset = RRset(Name('a.example.org.'), RRClass.CH(), RRType.NS(),
+ self.__ttl)
+ rrset.add_rdata(Rdata(RRType.NS(), RRClass.CH(), 'ns.example.org.'))
+ self.assertRaises(ValueError, diff.add_data, rrset)
- diff.remove_data(self.__rrset2)
++ self.assertRaises(ValueError, diff.delete_data, rrset)
+
+ def __do_raise_test(self):
+ """
+ Do a raise test. Expects that one of the operations is exchanged for
+ broken version.
+ """
+ diff = Diff(self, Name('example.org.'))
+ diff.add_data(self.__rrset1)
- self.assertRaises(ValueError, diff.remove_data, self.__rrset2)
++ diff.delete_data(self.__rrset2)
+ self.assertRaises(TestError, diff.commit)
+ self.assertTrue(self.__broken_called)
+ self.assertRaises(ValueError, diff.add_data, self.__rrset1)
- def test_raise_remove(self):
++ self.assertRaises(ValueError, diff.delete_data, self.__rrset2)
+ self.assertRaises(ValueError, diff.commit)
+ self.assertRaises(ValueError, diff.apply)
+
+ def test_raise_add(self):
+ """
+ Test the exception from add_rrset is propagated and the diff can't be
+ used afterwards.
+ """
+ self.add_rrset = self.__broken_operation
+ self.__do_raise_test()
+
- Test the exception from remove_rrset is propagated and the diff can't be
++ def test_raise_delete(self):
+ """
- self.remove_rrset = self.__broken_operation
++ Test the exception from delete_rrset is propagated and the diff can't be
+ used afterwards.
+ """
++ self.delete_rrset = self.__broken_operation
+ self.__do_raise_test()
+
+ def test_raise_commit(self):
+ """
+ Test the exception from updater's commit gets propagated and it can't be
+ used afterwards.
+ """
+ self.commit = self.__broken_operation
+ self.__do_raise_test()
+
+ def test_ttl(self):
+ """
+ Test the TTL handling. A warn function should have been called if they
+ differ, but that's all, it should not crash or raise.
+ """
+ orig_logger = isc.xfrin.diff.logger
+ try:
+ isc.xfrin.diff.logger = self
+ diff = Diff(self, Name('example.org.'))
+ diff.add_data(self.__rrset1)
+ rrset2 = RRset(Name('a.example.org.'), self.__rrclass,
+ self.__type, RRTTL(120))
+ rrset2.add_rdata(Rdata(self.__type, self.__rrclass, '192.10.2.2'))
+ diff.add_data(rrset2)
+ rrset2 = RRset(Name('a.example.org.'), self.__rrclass,
+ self.__type, RRTTL(6000))
+ rrset2.add_rdata(Rdata(self.__type, self.__rrclass, '192.10.2.3'))
+ diff.add_data(rrset2)
+ # They should get compacted together and complain.
+ diff.compact()
+ self.assertEqual(1, len(diff.get_buffer()))
+ # The TTL stays on the first value, no matter if smaller or bigger
+ # ones come later.
+ self.assertEqual(self.__ttl, diff.get_buffer()[0][1].get_ttl())
+ self.assertTrue(self.__warn_called)
+ finally:
+ isc.xfrin.diff.logger = orig_logger
+
if __name__ == "__main__":
isc.log.init("bind10")
unittest.main()
projects / thirdparty / kea.git / commitdiff
