[thirdparty/pdns.git] / regression-tests.dnsdist / test_AXFR.py

#!/usr/bin/env python
import threading
import dns
from dnsdisttests import DNSDistTest

class TestAXFR(DNSDistTest):

    # this test suite uses a different responder port
    # because, contrary to the other ones, its
    # TCP responder allows multiple responses and we don't want
    # to mix things up.
    _testServerPort = 5370
    _config_template = """
    newServer{address="127.0.0.1:%s"}
    """
    @classmethod
    def startResponders(cls):
        print("Launching responders..")

        cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
        cls._UDPResponder.setDaemon(True)
        cls._UDPResponder.start()
        cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue, False, True])
        cls._TCPResponder.setDaemon(True)
        cls._TCPResponder.start()

    def testOneMessageAXFR(self):
        """
        AXFR: One message
        """
        name = 'one.axfr.tests.powerdns.com.'
        query = dns.message.make_query(name, 'AXFR', 'IN')
        response = dns.message.make_response(query)
        soa = dns.rrset.from_text(name,
                                  60,
                                  dns.rdataclass.IN,
                                  dns.rdatatype.SOA,
                                  'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
        response.answer.append(soa)
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.A,
                                                   '192.0.2.1'))
        response.answer.append(soa)

        (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(response, receivedResponse)

    def testOneNoSOAAXFR(self):
        """
        AXFR: One message, no SOA
        """
        name = 'onenosoa.axfr.tests.powerdns.com.'
        query = dns.message.make_query(name, 'AXFR', 'IN')
        response = dns.message.make_response(query)
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.A,
                                                   '192.0.2.1'))

        (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(response, receivedResponse)

    def testFourMessagesAXFR(self):
        """
        AXFR: Four messages
        """
        name = 'four.axfr.tests.powerdns.com.'
        query = dns.message.make_query(name, 'AXFR', 'IN')
        responses = []
        soa = dns.rrset.from_text(name,
                                  60,
                                  dns.rdataclass.IN,
                                  dns.rdatatype.SOA,
                                  'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
        response = dns.message.make_response(query)
        response.answer.append(soa)
        responses.append(response)

        response = dns.message.make_response(query)
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.A,
                                                   '192.0.2.1'))
        responses.append(response)

        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.AAAA,
                                    '2001:DB8::1')
        response.answer.append(rrset)
        responses.append(response)

        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.TXT,
                                    'dummy')
        response.answer.append(rrset)
        responses.append(response)

        response = dns.message.make_response(query)
        response.answer.append(soa)
        responses.append(response)

        (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(len(receivedResponses), len(responses))

    def testFourNoFinalSOAAXFR(self):
        """
        AXFR: Four messages, no final SOA
        """
        name = 'fournosoa.axfr.tests.powerdns.com.'
        query = dns.message.make_query(name, 'AXFR', 'IN')
        responses = []
        soa = dns.rrset.from_text(name,
                                  60,
                                  dns.rdataclass.IN,
                                  dns.rdatatype.SOA,
                                  'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
        response = dns.message.make_response(query)
        response.answer.append(soa)
        responses.append(response)

        response = dns.message.make_response(query)
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.A,
                                                   '192.0.2.1'))
        responses.append(response)

        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.AAAA,
                                    '2001:DB8::1')
        response.answer.append(rrset)
        responses.append(response)

        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.TXT,
                                    'dummy')
        response.answer.append(rrset)
        responses.append(response)

        (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(len(receivedResponses), len(responses))

    # def testFourNoFirstSOAAXFR(self):
    #     """
    #     AXFR: Four messages, no SOA in the first one
    #     """
    #     name = 'fournosoainfirst.axfr.tests.powerdns.com.'
    #     query = dns.message.make_query(name, 'AXFR', 'IN')
    #     responses = []
    #     soa = dns.rrset.from_text(name,
    #                               60,
    #                               dns.rdataclass.IN,
    #                               dns.rdatatype.SOA,
    #                               'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
    #     response = dns.message.make_response(query)
    #     response.answer.append(dns.rrset.from_text(name,
    #                                                60,
    #                                                dns.rdataclass.IN,
    #                                                dns.rdatatype.A,
    #                                                '192.0.2.1'))
    #     responses.append(response)

    #     response = dns.message.make_response(query)
    #     rrset = dns.rrset.from_text(name,
    #                                 60,
    #                                 dns.rdataclass.IN,
    #                                 dns.rdatatype.AAAA,
    #                                 '2001:DB8::1')
    #     response.answer.append(soa)
    #     response.answer.append(rrset)
    #     responses.append(response)

    #     response = dns.message.make_response(query)
    #     rrset = dns.rrset.from_text('dummy.' + name,
    #                                 60,
    #                                 dns.rdataclass.IN,
    #                                 dns.rdatatype.AAAA,
    #                                 '2001:DB8::1')
    #     response.answer.append(rrset)
    #     responses.append(response)

    #     response = dns.message.make_response(query)
    #     rrset = dns.rrset.from_text(name,
    #                                 60,
    #                                 dns.rdataclass.IN,
    #                                 dns.rdatatype.TXT,
    #                                 'dummy')
    #     response.answer.append(rrset)
    #     response.answer.append(soa)
    #     responses.append(response)

    #     (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
    #     receivedQuery.id = query.id
    #     self.assertEqual(query, receivedQuery)
    #     self.assertEqual(len(receivedResponses), 1)

    # def testFourLastSOAInSecondAXFR(self):
    #     """
    #     AXFR: Four messages, SOA in the first one and the second one
    #     """
    #     name = 'foursecondsoainsecond.axfr.tests.powerdns.com.'
    #     query = dns.message.make_query(name, 'AXFR', 'IN')
    #     responses = []
    #     soa = dns.rrset.from_text(name,
    #                               60,
    #                               dns.rdataclass.IN,
    #                               dns.rdatatype.SOA,
    #                               'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')

    #     response = dns.message.make_response(query)
    #     response.answer.append(soa)
    #     response.answer.append(dns.rrset.from_text(name,
    #                                                60,
    #                                                dns.rdataclass.IN,
    #                                                dns.rdatatype.A,
    #                                                '192.0.2.1'))
    #     responses.append(response)

    #     response = dns.message.make_response(query)
    #     response.answer.append(soa)
    #     rrset = dns.rrset.from_text(name,
    #                                 60,
    #                                 dns.rdataclass.IN,
    #                                 dns.rdatatype.AAAA,
    #                                 '2001:DB8::1')
    #     response.answer.append(rrset)
    #     responses.append(response)

    #     response = dns.message.make_response(query)
    #     rrset = dns.rrset.from_text('dummy.' + name,
    #                                 60,
    #                                 dns.rdataclass.IN,
    #                                 dns.rdatatype.AAAA,
    #                                 '2001:DB8::1')
    #     response.answer.append(rrset)
    #     responses.append(response)

    #     response = dns.message.make_response(query)
    #     rrset = dns.rrset.from_text(name,
    #                                 60,
    #                                 dns.rdataclass.IN,
    #                                 dns.rdatatype.TXT,
    #                                 'dummy')
    #     response.answer.append(rrset)
    #     responses.append(response)

    #     (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
    #     receivedQuery.id = query.id
    #     self.assertEqual(query, receivedQuery)
    #     self.assertEqual(len(receivedResponses), 2)
Commit	Line	Data
548c8b66 RG	1	#!/usr/bin/env python
	2	import threading
	3	import dns
	4	from dnsdisttests import DNSDistTest
	5
	6	class TestAXFR(DNSDistTest):
	7
	8	# this test suite uses a different responder port
	9	# because, contrary to the other ones, its
	10	# TCP responder allows multiple responses and we don't want
	11	# to mix things up.
	12	_testServerPort = 5370
	13	_config_template = """
	14	newServer{address="127.0.0.1:%s"}
	15	"""
	16	@classmethod
	17	def startResponders(cls):
	18	print("Launching responders..")
	19
5df86a8a	20	cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
548c8b66 RG	21	cls._UDPResponder.setDaemon(True)
548c8b66 RG	22	cls._UDPResponder.start()
5df86a8a	23	cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue, False, True])
548c8b66 RG	24	cls._TCPResponder.setDaemon(True)
	25	cls._TCPResponder.start()
	26
548c8b66 RG	27	def testOneMessageAXFR(self):
	28	"""
	29	AXFR: One message
	30	"""
	31	name = 'one.axfr.tests.powerdns.com.'
	32	query = dns.message.make_query(name, 'AXFR', 'IN')
	33	response = dns.message.make_response(query)
	34	soa = dns.rrset.from_text(name,
	35	60,
	36	dns.rdataclass.IN,
	37	dns.rdatatype.SOA,
	38	'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
	39	response.answer.append(soa)
	40	response.answer.append(dns.rrset.from_text(name,
	41	60,
	42	dns.rdataclass.IN,
	43	dns.rdatatype.A,
	44	'192.0.2.1'))
	45	response.answer.append(soa)
	46
	47	(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
	48	receivedQuery.id = query.id
	49	self.assertEqual(query, receivedQuery)
	50	self.assertEqual(response, receivedResponse)
	51
	52	def testOneNoSOAAXFR(self):
	53	"""
	54	AXFR: One message, no SOA
	55	"""
	56	name = 'onenosoa.axfr.tests.powerdns.com.'
	57	query = dns.message.make_query(name, 'AXFR', 'IN')
	58	response = dns.message.make_response(query)
	59	response.answer.append(dns.rrset.from_text(name,
	60	60,
	61	dns.rdataclass.IN,
	62	dns.rdatatype.A,
	63	'192.0.2.1'))
	64
	65	(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
	66	receivedQuery.id = query.id
	67	self.assertEqual(query, receivedQuery)
	68	self.assertEqual(response, receivedResponse)
	69
	70	def testFourMessagesAXFR(self):
	71	"""
	72	AXFR: Four messages
	73	"""
	74	name = 'four.axfr.tests.powerdns.com.'
	75	query = dns.message.make_query(name, 'AXFR', 'IN')
	76	responses = []
	77	soa = dns.rrset.from_text(name,
	78	60,
	79	dns.rdataclass.IN,
	80	dns.rdatatype.SOA,
	81	'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
	82	response = dns.message.make_response(query)
	83	response.answer.append(soa)
	84	responses.append(response)
	85
	86	response = dns.message.make_response(query)
	87	response.answer.append(dns.rrset.from_text(name,
	88	60,
	89	dns.rdataclass.IN,
	90	dns.rdatatype.A,
91	'192.0.2.1'))
92	responses.append(response)
93
94	response = dns.message.make_response(query)
95	rrset = dns.rrset.from_text(name,
96	60,
97	dns.rdataclass.IN,
98	dns.rdatatype.AAAA,
99	'2001:DB8::1')
100	response.answer.append(rrset)
101	responses.append(response)
102
103	response = dns.message.make_response(query)
104	rrset = dns.rrset.from_text(name,
105	60,
106	dns.rdataclass.IN,
107	dns.rdatatype.TXT,
108	'dummy')
109	response.answer.append(rrset)
110	responses.append(response)
111
112	response = dns.message.make_response(query)
113	response.answer.append(soa)
114	responses.append(response)
115
116	(receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
117	receivedQuery.id = query.id
118	self.assertEqual(query, receivedQuery)
119	self.assertEqual(len(receivedResponses), len(responses))
120
121	def testFourNoFinalSOAAXFR(self):
122	"""
123	AXFR: Four messages, no final SOA
124	"""
125	name = 'fournosoa.axfr.tests.powerdns.com.'
126	query = dns.message.make_query(name, 'AXFR', 'IN')
127	responses = []
128	soa = dns.rrset.from_text(name,
129	60,
130	dns.rdataclass.IN,
131	dns.rdatatype.SOA,
132	'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
133	response = dns.message.make_response(query)
134	response.answer.append(soa)
135	responses.append(response)
136
137	response = dns.message.make_response(query)
138	response.answer.append(dns.rrset.from_text(name,
139	60,
140	dns.rdataclass.IN,
141	dns.rdatatype.A,
142	'192.0.2.1'))
143	responses.append(response)
144
145	response = dns.message.make_response(query)
146	rrset = dns.rrset.from_text(name,
147	60,
148	dns.rdataclass.IN,
149	dns.rdatatype.AAAA,
150	'2001:DB8::1')
151	response.answer.append(rrset)
152	responses.append(response)
153
154	response = dns.message.make_response(query)
155	rrset = dns.rrset.from_text(name,
156	60,
157	dns.rdataclass.IN,
158	dns.rdatatype.TXT,
159	'dummy')
160	response.answer.append(rrset)
161	responses.append(response)
162
163	(receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
164	receivedQuery.id = query.id
165	self.assertEqual(query, receivedQuery)
166	self.assertEqual(len(receivedResponses), len(responses))
167
31abd117 RG	168	# def testFourNoFirstSOAAXFR(self):
	169	# """
	170	# AXFR: Four messages, no SOA in the first one
	171	# """
	172	# name = 'fournosoainfirst.axfr.tests.powerdns.com.'
	173	# query = dns.message.make_query(name, 'AXFR', 'IN')
	174	# responses = []
	175	# soa = dns.rrset.from_text(name,
	176	# 60,
	177	# dns.rdataclass.IN,
	178	# dns.rdatatype.SOA,
	179	# 'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
	180	# response = dns.message.make_response(query)
	181	# response.answer.append(dns.rrset.from_text(name,
	182	# 60,
	183	# dns.rdataclass.IN,
	184	# dns.rdatatype.A,
	185	# '192.0.2.1'))
	186	# responses.append(response)
548c8b66	187
31abd117 RG	188	# response = dns.message.make_response(query)
	189	# rrset = dns.rrset.from_text(name,
	190	# 60,
	191	# dns.rdataclass.IN,
	192	# dns.rdatatype.AAAA,
	193	# '2001:DB8::1')
	194	# response.answer.append(soa)
	195	# response.answer.append(rrset)
	196	# responses.append(response)
548c8b66	197
31abd117 RG	198	# response = dns.message.make_response(query)
	199	# rrset = dns.rrset.from_text('dummy.' + name,
	200	# 60,
	201	# dns.rdataclass.IN,
	202	# dns.rdatatype.AAAA,
	203	# '2001:DB8::1')
	204	# response.answer.append(rrset)
	205	# responses.append(response)
548c8b66	206
31abd117 RG	207	# response = dns.message.make_response(query)
	208	# rrset = dns.rrset.from_text(name,
	209	# 60,
	210	# dns.rdataclass.IN,
	211	# dns.rdatatype.TXT,
	212	# 'dummy')
	213	# response.answer.append(rrset)
	214	# response.answer.append(soa)
	215	# responses.append(response)
548c8b66	216
31abd117 RG	217	# (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
	218	# receivedQuery.id = query.id
	219	# self.assertEqual(query, receivedQuery)
	220	# self.assertEqual(len(receivedResponses), 1)
548c8b66	221
31abd117 RG	222	# def testFourLastSOAInSecondAXFR(self):
	223	# """
	224	# AXFR: Four messages, SOA in the first one and the second one
	225	# """
	226	# name = 'foursecondsoainsecond.axfr.tests.powerdns.com.'
	227	# query = dns.message.make_query(name, 'AXFR', 'IN')
	228	# responses = []
	229	# soa = dns.rrset.from_text(name,
	230	# 60,
	231	# dns.rdataclass.IN,
	232	# dns.rdatatype.SOA,
	233	# 'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
548c8b66	234
31abd117 RG	235	# response = dns.message.make_response(query)
	236	# response.answer.append(soa)
	237	# response.answer.append(dns.rrset.from_text(name,
	238	# 60,
	239	# dns.rdataclass.IN,
	240	# dns.rdatatype.A,
	241	# '192.0.2.1'))
	242	# responses.append(response)
548c8b66	243
31abd117 RG	244	# response = dns.message.make_response(query)
	245	# response.answer.append(soa)
	246	# rrset = dns.rrset.from_text(name,
	247	# 60,
	248	# dns.rdataclass.IN,
	249	# dns.rdatatype.AAAA,
	250	# '2001:DB8::1')
	251	# response.answer.append(rrset)
	252	# responses.append(response)
548c8b66	253
31abd117 RG	254	# response = dns.message.make_response(query)
	255	# rrset = dns.rrset.from_text('dummy.' + name,
	256	# 60,
	257	# dns.rdataclass.IN,
	258	# dns.rdatatype.AAAA,
	259	# '2001:DB8::1')
	260	# response.answer.append(rrset)
	261	# responses.append(response)
548c8b66	262
31abd117 RG	263	# response = dns.message.make_response(query)
	264	# rrset = dns.rrset.from_text(name,
	265	# 60,
	266	# dns.rdataclass.IN,
	267	# dns.rdatatype.TXT,
	268	# 'dummy')
	269	# response.answer.append(rrset)
	270	# responses.append(response)
548c8b66	271
31abd117 RG	272	# (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
	273	# receivedQuery.id = query.id
	274	# self.assertEqual(query, receivedQuery)
	275	# self.assertEqual(len(receivedResponses), 2)