Commit 0e945f09 authored by JINMEI Tatuya's avatar JINMEI Tatuya
Browse files

[1389] test cleanup: extracted common setup/checks to separate methods

to avoid duplicates.
parent c9be5877
......@@ -893,97 +893,104 @@ class TestXfroutSession(TestXfroutSessionBase):
self.assertTrue(rrsets_equal(self.soa_rrset,
r.get_section(Message.SECTION_ANSWER)[0]))
def test_reply_xfrout_query_axfr_maxlen_with_soa(self):
# Similar to the 'maxlen' test, but the first message should be
# able to contain both SOA and the large RR.
# For this test we use '.' for all owner names and names in RDATA
# to avoid having unexpected results due to compression.
# there should be no more message
self.assertEqual(0, len(self.sock.sendqueue))
def maxlen_test_common_setup(self, tsig=False):
'''Common initialization for some of the tests below
For those tests we use '.' for all owner names and names in RDATA
to avoid having unexpected results due to compression. It returns
the created SOA for convenience.
If tsig is True, also setup TSIG (mock) context. In our test cases
the size of the TSIG RR is 81 bytes (key name = example.com,
algorithm = hmac-md5)
'''
soa = RRset(Name('.'), RRClass.IN(), RRType.SOA(), RRTTL(3600))
soa.add_rdata(Rdata(RRType.SOA(), RRClass.IN(), '. . 0 0 0 0 0'))
self.mdata = self.create_request_data(zone_name=Name('.'))
self.xfrsess._soa = soa
if tsig:
self.xfrsess._tsig_ctx = \
self.create_mock_tsig_ctx(TSIGError.NOERROR)
self.xfrsess._tsig_len = 81
return soa
def maxlen_test_common_checks(self, soa_rr, test_rr, expected_n_rr):
'''A set of common assertion checks for some tests below.
In all cases two AXFR response messages should have been created.
expected_n_rr is a list of two elements, each specifies the expected
number of answer RRs for each message: expected_n_rr[0] is the expected
number of the first answer RRs; expected_n_rr[1] is the expected number
of the second answer RRs. The message that contains two RRs should
have the maximum possible wire length (65535 bytes). And, in all
cases, the resulting RRs should be in the order of SOA, another RR,
SOA.
'''
# Check the first message
r, rlen = self.sock.read_msg(need_len=True)
if expected_n_rr[0] == 2:
self.assertEqual(65535, rlen)
self.assertEqual(expected_n_rr[0],
r.get_rr_count(Message.SECTION_ANSWER))
actual_rrs = r.get_section(Message.SECTION_ANSWER)[:]
# Check the second message
r, rlen = self.sock.read_msg(need_len=True)
if expected_n_rr[1] == 2:
self.assertEqual(65535, rlen)
self.assertEqual(expected_n_rr[1],
r.get_rr_count(Message.SECTION_ANSWER))
actual_rrs.extend(r.get_section(Message.SECTION_ANSWER))
for (expected_rr, actual_rr) in zip([soa_rr, test_rr, soa_rr],
actual_rrs):
self.assertTrue(rrsets_equal(expected_rr, actual_rr))
# there should be no more message
self.assertEqual(0, len(self.sock.sendqueue))
def test_reply_xfrout_query_axfr_maxlen_with_soa(self):
# Similar to the 'maxlen' test, but the first message should be
# able to contain both SOA and the large RR.
soa = self.maxlen_test_common_setup()
# The first message will contain the question (5 bytes), so the
# test RDATA should allow a room for that.
test_rr = create_generic(Name('.'), 65512 - 5 - get_rrset_len(soa))
self.xfrsess._iterator = [soa, test_rr]
self.xfrsess._reply_xfrout_query(self.getmsg(), self.sock)
r, rlen = self.sock.read_msg(need_len=True)
self.assertEqual(65535, rlen)
self.assertEqual(2, r.get_rr_count(Message.SECTION_ANSWER))
self.assertTrue(rrsets_equal(soa,
r.get_section(Message.SECTION_ANSWER)[0]))
self.assertTrue(rrsets_equal(test_rr,
r.get_section(Message.SECTION_ANSWER)[1]))
self.maxlen_test_common_checks(soa, test_rr, [2, 1])
def test_reply_xfrout_query_axfr_maxlen_with_soa_with_tsig(self):
# Similar to the previous case, but with TSIG. In our test cases
# the size of the TSIG RR is 81 bytes (key name = example.com,
# algorithm = hmac-md5)
soa = RRset(Name('.'), RRClass.IN(), RRType.SOA(), RRTTL(3600))
soa.add_rdata(Rdata(RRType.SOA(), RRClass.IN(), '. . 0 0 0 0 0'))
self.mdata = self.create_request_data(zone_name=Name('.'))
self.xfrsess._soa = soa
self.xfrsess._tsig_ctx = self.create_mock_tsig_ctx(TSIGError.NOERROR)
self.xfrsess._tsig_len = 81
# Similar to the previous case, but with TSIG (whose size is 81 bytes).
soa = self.maxlen_test_common_setup(True)
test_rr = create_generic(Name('.'), 65512 - 5 - 81 -
get_rrset_len(soa))
self.xfrsess._iterator = [soa, test_rr]
self.xfrsess._reply_xfrout_query(self.getmsg(), self.sock)
r, rlen = self.sock.read_msg(need_len=True)
self.assertEqual(65535, rlen)
self.assertEqual(2, r.get_rr_count(Message.SECTION_ANSWER))
self.assertTrue(rrsets_equal(soa,
r.get_section(Message.SECTION_ANSWER)[0]))
self.assertTrue(rrsets_equal(test_rr,
r.get_section(Message.SECTION_ANSWER)[1]))
self.maxlen_test_common_checks(soa, test_rr, [2, 1])
def test_reply_xfrout_query_axfr_maxlen_with_endsoa(self):
# Similar to the previous test, but the first message cannot contain
# Similar to the max w/ soa test, but the first message cannot contain
# both SOA and the long RR due to the question section. The second
# message should be able to contain both.
soa = RRset(Name('.'), RRClass.IN(), RRType.SOA(), RRTTL(3600))
soa.add_rdata(Rdata(RRType.SOA(), RRClass.IN(), '. . 0 0 0 0 0'))
self.mdata = self.create_request_data(zone_name=Name('.'))
self.xfrsess._soa = soa
soa = self.maxlen_test_common_setup()
test_rr = create_generic(Name('.'), 65512 - get_rrset_len(soa))
self.xfrsess._iterator = [soa, test_rr]
self.xfrsess._reply_xfrout_query(self.getmsg(), self.sock)
r = self.sock.read_msg()
self.assertEqual(1, r.get_rr_count(Message.SECTION_ANSWER))
self.assertTrue(rrsets_equal(soa,
r.get_section(Message.SECTION_ANSWER)[0]))
r, rlen = self.sock.read_msg(need_len=True)
self.assertEqual(65535, rlen)
self.assertEqual(2, r.get_rr_count(Message.SECTION_ANSWER))
self.assertTrue(rrsets_equal(test_rr,
r.get_section(Message.SECTION_ANSWER)[0]))
self.assertTrue(rrsets_equal(soa,
r.get_section(Message.SECTION_ANSWER)[1]))
self.maxlen_test_common_checks(soa, test_rr, [1, 2])
def test_reply_xfrout_query_axfr_maxlen_with_endsoa_with_tsig(self):
# Similar to the previous case, but with TSIG. In our test cases
# the size of the TSIG RR is 81 bytes (key name = example.com,
# algorithm = hmac-md5)
soa = RRset(Name('.'), RRClass.IN(), RRType.SOA(), RRTTL(3600))
soa.add_rdata(Rdata(RRType.SOA(), RRClass.IN(), '. . 0 0 0 0 0'))
self.mdata = self.create_request_data(zone_name=Name('.'))
self.xfrsess._soa = soa
self.xfrsess._tsig_ctx = self.create_mock_tsig_ctx(TSIGError.NOERROR)
self.xfrsess._tsig_len = 81
# Similar to the previous case, but with TSIG.
soa = self.maxlen_test_common_setup(True)
test_rr = create_generic(Name('.'), 65512 - 81 - get_rrset_len(soa))
self.xfrsess._iterator = [soa, test_rr]
self.xfrsess._reply_xfrout_query(self.getmsg(), self.sock)
r = self.sock.read_msg()
self.assertEqual(1, r.get_rr_count(Message.SECTION_ANSWER))
self.assertTrue(rrsets_equal(soa,
r.get_section(Message.SECTION_ANSWER)[0]))
r, rlen = self.sock.read_msg(need_len=True)
self.assertEqual(65535, rlen)
self.assertEqual(2, r.get_rr_count(Message.SECTION_ANSWER))
self.assertTrue(rrsets_equal(test_rr,
r.get_section(Message.SECTION_ANSWER)[0]))
self.assertTrue(rrsets_equal(soa,
r.get_section(Message.SECTION_ANSWER)[1]))
self.maxlen_test_common_checks(soa, test_rr, [1, 2])
def test_reply_xfrout_query_axfr_toobigdata(self):
# Similar to the 'maxlen' test, but the RR doesn't even fit in a
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment