'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testAdvancedAllowDropped(self):
"""
"""
name = 'notallowed.advanced.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN')
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, None)
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
class TestAdvancedFixupCase(DNSDistTest):
response.answer.append(rrset)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(expectedResponse, receivedResponse)
-
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(expectedResponse, receivedResponse)
class TestAdvancedRemoveRD(DNSDistTest):
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.assertEquals(expectedQuery, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.assertEquals(expectedQuery, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testAdvancedKeepRD(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
class TestAdvancedAddCD(DNSDistTest):
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.assertEquals(expectedQuery, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.assertEquals(expectedQuery, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testAdvancedSetCDViaAction(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.assertEquals(expectedQuery, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.assertEquals(expectedQuery, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testAdvancedKeepNoCD(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
class TestAdvancedClearRD(DNSDistTest):
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.assertEquals(expectedQuery, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.assertEquals(expectedQuery, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testAdvancedClearRDViaAction(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.assertEquals(expectedQuery, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.assertEquals(expectedQuery, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testAdvancedKeepRD(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
class TestAdvancedACL(DNSDistTest):
name = 'tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN')
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, None)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, None)
class TestAdvancedDelay(DNSDistTest):
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, response)
def testAOverUDPReturnsNotImplemented(self):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.NOTIMP)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
class TestAdvancedLogAction(DNSDistTest):
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
- (_, receivedResponse) = self.sendUDPQuery(doquery, response)
- self.assertEquals(receivedResponse, None)
- (_, receivedResponse) = self.sendTCPQuery(doquery, response)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(doquery, response)
+ self.assertEquals(receivedResponse, None)
class TestAdvancedQClass(DNSDistTest):
name = 'qclasschaos.advanced.tests.powerdns.com.'
query = dns.message.make_query(name, 'TXT', 'CHAOS')
- (_, receivedResponse) = self.sendUDPQuery(query, response=None)
- self.assertEquals(receivedResponse, None)
- (_, receivedResponse) = self.sendTCPQuery(query, response=None)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None)
+ self.assertEquals(receivedResponse, None)
def testAdvancedQClassINAllow(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
class TestAdvancedOpcode(DNSDistTest):
query = dns.message.make_query(name, 'A', 'IN')
query.set_opcode(dns.opcode.NOTIFY)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None)
- self.assertEquals(receivedResponse, None)
- (_, receivedResponse) = self.sendTCPQuery(query, response=None)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None)
+ self.assertEquals(receivedResponse, None)
def testAdvancedOpcodeUpdateINAllow(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
class TestAdvancedNonTerminalRule(DNSDistTest):
'192.0.2.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.assertEquals(expectedQuery, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.assertEquals(expectedQuery, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
class TestAdvancedStringOnlyServer(DNSDistTest):
'192.0.2.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
class TestAdvancedRestoreFlagsOnSelfResponse(DNSDistTest):
'192.0.2.1')
response.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(response, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(response, receivedResponse)
class TestAdvancedQPS(DNSDistTest):
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
class TestAdvancedNMGRule(DNSDistTest):
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
class TestDSTPortRule(DNSDistTest):
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
class TestAdvancedLabelsCountRule(DNSDistTest):
'192.0.2.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
# more than 6 labels, the query should be refused
name = 'not.ok.labelscount.advanced.tests.powerdns.com.'
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
# less than 5 labels, the query should be refused
name = 'labelscountadvanced.tests.powerdns.com.'
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
class TestAdvancedWireLengthRule(DNSDistTest):
'192.0.2.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
# too short, the query should be refused
name = 'short.qnamewirelength.advanced.tests.powerdns.com.'
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
# too long, the query should be refused
name = 'toolongtobevalid.qnamewirelength.advanced.tests.powerdns.com.'
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
class TestAdvancedIncludeDir(DNSDistTest):
'192.0.2.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
# this one should be refused
name = 'notincludedir.advanced.tests.powerdns.com.'
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
class TestAdvancedLuaDO(DNSDistTest):
doResponse.set_rcode(dns.rcode.NXDOMAIN)
# without DO
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, response)
# with DO
- (_, receivedResponse) = self.sendUDPQuery(queryWithDO, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- doResponse.id = receivedResponse.id
- self.assertEquals(receivedResponse, doResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(queryWithDO, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- doResponse.id = receivedResponse.id
- self.assertEquals(receivedResponse, doResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(queryWithDO, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ doResponse.id = receivedResponse.id
+ self.assertEquals(receivedResponse, doResponse)
class TestAdvancedLuaRefused(DNSDistTest):
refusedResponse = dns.message.make_response(query)
refusedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- refusedResponse.id = receivedResponse.id
- self.assertEquals(receivedResponse, refusedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- refusedResponse.id = receivedResponse.id
- self.assertEquals(receivedResponse, refusedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ refusedResponse.id = receivedResponse.id
+ self.assertEquals(receivedResponse, refusedResponse)
class TestAdvancedLuaActionReturnSyntax(DNSDistTest):
refusedResponse = dns.message.make_response(query)
refusedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- refusedResponse.id = receivedResponse.id
- self.assertEquals(receivedResponse, refusedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- refusedResponse.id = receivedResponse.id
- self.assertEquals(receivedResponse, refusedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ refusedResponse.id = receivedResponse.id
+ self.assertEquals(receivedResponse, refusedResponse)
class TestAdvancedLuaTruncated(DNSDistTest):
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
def testAdvancedNoRDAllowed(self):
"""
query.flags &= ~dns.flags.RD
response = dns.message.make_response(query)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
class TestAdvancedGetLocalPort(DNSDistTest):
'port-was-{}.local-port.advanced.tests.powerdns.com.'.format(self._dnsDistPort))
response.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, response)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, response)
class TestAdvancedGetLocalPortOnAnyBind(DNSDistTest):
'port-was-{}.local-port-any.advanced.tests.powerdns.com.'.format(self._dnsDistPort))
response.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, response)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, response)
class TestAdvancedGetLocalAddressOnAnyBind(DNSDistTest):
'address-was-127-0-0-1.local-address-any.advanced.tests.powerdns.com.')
response.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, response)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, response)
class TestAdvancedLuaTempFailureTTL(DNSDistTest):
'::1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, response)
class TestAdvancedEDNSOptionRule(DNSDistTest):
ecso = clientsubnetoption.ClientSubnetOption('127.0.0.1', 24)
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, options=[ecso], payload=512)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None)
- self.assertEquals(receivedResponse, None)
- (_, receivedResponse) = self.sendTCPQuery(query, response=None)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None)
+ self.assertEquals(receivedResponse, None)
def testReplied(self):
"""
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, options=[], payload=512)
response = dns.message.make_response(query)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, response)
# and with no EDNS at all
query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
response = dns.message.make_response(query)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, response)
class TestAdvancedAllowHeaderOnly(DNSDistTest):
"""
name = 'drop.test.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN')
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, None)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, None)
def testAWithECS(self):
"""
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testSimpleA(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testAnyIsTruncated(self):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
def testQNameReturnsSpoofed(self):
"""
'1.2.3.4')
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
def testDomainAndQTypeReturnsNotImplemented(self):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.NOTIMP)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
def testDomainWithoutQTypeIsNotAffected(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testOtherDomainANDQTypeIsNotAffected(self):
"""
'nothing to see here')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testWrongResponse(self):
"""
'nothing to see here')
unrelatedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, unrelatedResponse)
- self.assertTrue(receivedQuery)
- self.assertEquals(receivedResponse, None)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, unrelatedResponse)
- self.assertTrue(receivedQuery)
- self.assertEquals(receivedResponse, None)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, unrelatedResponse)
+ self.assertTrue(receivedQuery)
+ self.assertEquals(receivedResponse, None)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
def testHeaderOnlyRefused(self):
"""
response.set_rcode(dns.rcode.REFUSED)
response.question = []
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, response)
def testHeaderOnlyNoErrorResponse(self):
"""
response = dns.message.make_response(query)
response.question = []
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, None)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, None)
def testHeaderOnlyNXDResponse(self):
"""
response.set_rcode(dns.rcode.NXDOMAIN)
response.question = []
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, None)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, None)
def testAddActionDNSName(self):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
def testAddActionDNSNames(self):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
-if __name__ == '__main__':
- unittest.main()
- exit(0)
response.answer.append(rrset)
for _ in range(numberOfQueries):
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, response)
for key in self._responsesCounter:
value = self._responsesCounter[key]
response.answer.append(rrset)
for _ in range(numberOfQueries):
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, response)
for key in self._responsesCounter:
value = self._responsesCounter[key]
response.answer.append(rrset)
# first query to fill the cache
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, response)
# next queries should hit the cache
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, response)
-
- # over TCP too
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, response)
# we mark the backend as down
self.sendConsoleCommand("getServer(0):setDown()")
# we should NOT get a cached entry since it has ECS and we haven't asked the pool
# to add ECS when no backend is up
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, None)
-
- # same over TCP
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, None)
class TestCachingECSWithPoolECS(DNSDistTest):
response.answer.append(rrset)
# first query to fill the cache
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, response)
# next queries should hit the cache
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, response)
-
- # over TCP too
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, response)
# we mark the backend as down
self.sendConsoleCommand("getServer(0):setDown()")
# we should STILL get a cached entry since it has ECS and we have asked the pool
# to add ECS when no backend is up
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, response)
-
- # same over TCP
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, response)
class TestCachingCollisionNoECSParsing(DNSDistTest):
'192.0.2.255')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
def testCookie(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
def testECS4(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
def testECS6(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
def testECS6Cookie(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
def testMultiCookiesECS6(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
class TestEDNSOptionsAddingECS(EDNSOptionsBase):
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseNoEDNS(response, receivedResponse)
def testCookie(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery, 1)
- self.checkResponseEDNSWithoutECS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery, 1)
- self.checkResponseEDNSWithoutECS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery, 1)
+ self.checkResponseEDNSWithoutECS(response, receivedResponse)
def testECS4(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryEDNSWithECS(query, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryEDNSWithECS(query, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.checkQueryEDNSWithECS(query, receivedQuery)
+ self.checkResponseEDNSWithECS(response, receivedResponse)
def testECS6(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryEDNSWithECS(query, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryEDNSWithECS(query, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.checkQueryEDNSWithECS(query, receivedQuery)
+ self.checkResponseEDNSWithECS(response, receivedResponse)
def testECS6Cookie(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryEDNSWithECS(query, receivedQuery, 1)
- self.checkResponseEDNSWithECS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryEDNSWithECS(query, receivedQuery, 1)
- self.checkResponseEDNSWithECS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.checkQueryEDNSWithECS(query, receivedQuery, 1)
+ self.checkResponseEDNSWithECS(response, receivedResponse)
def testMultiCookiesECS6(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(receivedQuery, query)
- self.assertEquals(receivedResponse, response)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageNoEDNS(expectedResponse, receivedResponse)
name = 'no-edns.tc.edns-self.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN')
expectedResponse = dns.message.make_response(query)
expectedResponse.flags |= dns.flags.TC
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageNoEDNS(expectedResponse, receivedResponse)
name = 'no-edns.lua.edns-self.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN')
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.NXDOMAIN)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageNoEDNS(expectedResponse, receivedResponse)
name = 'no-edns.spoof.edns-self.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN')
dns.rdatatype.A,
'192.0.2.1', '192.0.2.2'))
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageNoEDNS(expectedResponse, receivedResponse)
def testWithEDNSNoDO(self):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
+ self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
+ self.assertEquals(receivedResponse.payload, 1042)
name = 'edns-no-do.tc.edns-self.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, want_dnssec=False)
expectedResponse = dns.message.make_response(query)
expectedResponse.flags |= dns.flags.TC
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
+ self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
+ self.assertEquals(receivedResponse.payload, 1042)
name = 'edns-no-do.lua.edns-self.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, want_dnssec=False)
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.NXDOMAIN)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
+ self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
+ self.assertEquals(receivedResponse.payload, 1042)
name = 'edns-no-do.spoof.edns-self.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, want_dnssec=False)
dns.rdatatype.A,
'192.0.2.1', '192.0.2.2'))
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
+ self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
+ self.assertEquals(receivedResponse.payload, 1042)
def testWithEDNSWithDO(self):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
+ self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
+ self.assertEquals(receivedResponse.payload, 1042)
name = 'edns-do.tc.edns-self.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, want_dnssec=True)
expectedResponse = dns.message.make_response(query)
expectedResponse.flags |= dns.flags.TC
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
+ self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
+ self.assertEquals(receivedResponse.payload, 1042)
name = 'edns-do.lua.edns-self.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, want_dnssec=True)
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.NXDOMAIN)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
+ self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
+ self.assertEquals(receivedResponse.payload, 1042)
name = 'edns-do.spoof.edns-self.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, want_dnssec=True)
dns.rdatatype.A,
'192.0.2.1', '192.0.2.2'))
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
+ self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
+ self.assertEquals(receivedResponse.payload, 1042)
def testWithEDNSNoOptions(self):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
+ self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
+ self.assertEquals(receivedResponse.payload, 1042)
name = 'edns-options.tc.edns-self.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, options=[ecso], payload=512, want_dnssec=True)
expectedResponse = dns.message.make_response(query)
expectedResponse.flags |= dns.flags.TC
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
+ self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
+ self.assertEquals(receivedResponse.payload, 1042)
name = 'edns-options.lua.edns-self.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, options=[ecso], payload=512, want_dnssec=True)
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.NXDOMAIN)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
+ self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
+ self.assertEquals(receivedResponse.payload, 1042)
name = 'edns-options.spoof.edns-self.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, options=[ecso], payload=512, want_dnssec=True)
dns.rdatatype.A,
'192.0.2.1', '192.0.2.2'))
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
- self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
- self.assertEquals(receivedResponse.payload, 1042)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
+ self.assertTrue(receivedResponse.ednsflags & dns.flags.DO)
+ self.assertEquals(receivedResponse.payload, 1042)
class TestEDNSSelfGeneratedDisabled(DNSDistTest):
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageNoEDNS(expectedResponse, receivedResponse)
name = 'edns-no-do.tc.edns-self-disabled.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, want_dnssec=False)
expectedResponse = dns.message.make_response(query)
expectedResponse.flags |= dns.flags.TC
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageNoEDNS(expectedResponse, receivedResponse)
name = 'edns-no-do.lua.edns-self-disabled.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, want_dnssec=False)
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.NXDOMAIN)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageNoEDNS(expectedResponse, receivedResponse)
name = 'edns-no-do.spoof.edns-self-disabled.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, want_dnssec=False)
dns.rdatatype.A,
'192.0.2.1', '192.0.2.2'))
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.checkMessageNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.checkMessageNoEDNS(expectedResponse, receivedResponse)
response.answer.append(rrset)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseNoEDNS(expectedResponse, receivedResponse)
def testWithEDNSNoECS(self):
"""
response.answer.append(rrset)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse)
def testWithEDNSECS(self):
"""
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryEDNSWithECS(query, receivedQuery)
- self.checkResponseEDNSWithoutECS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryEDNSWithECS(query, receivedQuery)
- self.checkResponseEDNSWithoutECS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.checkQueryEDNSWithECS(query, receivedQuery)
+ self.checkResponseEDNSWithoutECS(response, receivedResponse)
def testWithoutEDNSResponseWithECS(self):
"""
response.answer.append(rrset)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseNoEDNS(expectedResponse, receivedResponse)
def testWithEDNSNoECSResponseWithECS(self):
"""
response.answer.append(rrset)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse)
def testWithEDNSNoECSResponseWithCookiesThenECS(self):
"""
expectedResponse.answer.append(rrset)
expectedResponse.use_edns(edns=True, payload=4096, options=[ecoResponse])
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse, withCookies=1)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse, withCookies=1)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse, withCookies=1)
def testWithEDNSNoECSResponseWithECSThenCookies(self):
"""
expectedResponse.answer.append(rrset)
response.use_edns(edns=True, payload=4096, options=[ecoResponse])
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse, withCookies=1)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse, withCookies=1)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse, withCookies=1)
def testWithEDNSNoECSResponseWithCookiesThenECSThenCookies(self):
"""
response.answer.append(rrset)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse, withCookies=2)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse, withCookies=2)
-
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse, withCookies=2)
class TestEdnsClientSubnetOverride(DNSDistTest):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseNoEDNS(expectedResponse, receivedResponse)
def testWithEDNSNoECS(self):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseEDNSWithoutECS(expectedResponse, receivedResponse)
def testWithEDNSShorterInitialECS(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseEDNSWithECS(response, receivedResponse)
def testWithEDNSLongerInitialECS(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseEDNSWithECS(response, receivedResponse)
def testWithEDNSSameSizeInitialECS(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseEDNSWithECS(response, receivedResponse)
class TestECSDisabledByRuleOrLua(DNSDistTest):
"""
response.answer.append(rrset)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseNoEDNS(expectedResponse, receivedResponse)
def testWithECSDisabledViaRule(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryNoEDNS(query, receivedQuery)
- self.checkResponseNoEDNS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryNoEDNS(query, receivedQuery)
- self.checkResponseNoEDNS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.checkQueryNoEDNS(query, receivedQuery)
+ self.checkResponseNoEDNS(response, receivedResponse)
def testWithECSDisabledViaLua(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryNoEDNS(query, receivedQuery)
- self.checkResponseNoEDNS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryNoEDNS(query, receivedQuery)
- self.checkResponseNoEDNS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.checkQueryNoEDNS(query, receivedQuery)
+ self.checkResponseNoEDNS(response, receivedResponse)
class TestECSOverrideSetByRuleOrLua(DNSDistTest):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryEDNSWithECS(query, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.checkQueryEDNSWithECS(query, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.checkQueryEDNSWithECS(query, receivedQuery)
+ self.checkResponseEDNSWithECS(response, receivedResponse)
def testWithECSOverrideSetViaRule(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseEDNSWithECS(response, receivedResponse)
def testWithECSOverrideSetViaLua(self):
"""
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseEDNSWithECS(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseEDNSWithECS(response, receivedResponse)
class TestECSPrefixLengthSetByRuleOrLua(DNSDistTest):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseNoEDNS(expectedResponse, receivedResponse)
def testWithECSPrefixLengthOverriddenViaRule(self):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseNoEDNS(expectedResponse, receivedResponse)
def testWithECSPrefixLengthOverriddenViaLua(self):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseNoEDNS(expectedResponse, receivedResponse)
class TestECSPrefixSetByRule(DNSDistTest):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseNoEDNS(expectedResponse, receivedResponse)
def testWithECSSetByRule(self):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = expectedQuery.id
- self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
- self.checkResponseNoEDNS(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.checkQueryEDNSWithECS(expectedQuery, receivedQuery)
+ self.checkResponseNoEDNS(expectedResponse, receivedResponse)
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
def testRecordsCountAllowOneAR(self):
"""
dns.rdatatype.A,
'127.0.0.1'))
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testRecordsCountRefuseTwoAR(self):
"""
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
class TestRecordsCountMoreThanOneLessThanFour(DNSDistTest):
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
def testRecordsCountAllowTwoAN(self):
"""
response = dns.message.make_response(query)
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testRecordsCountRefuseFourAN(self):
"""
expectedResponse.set_rcode(dns.rcode.REFUSED)
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
class TestRecordsCountNothingInNS(DNSDistTest):
expectedResponse.set_rcode(dns.rcode.REFUSED)
expectedResponse.authority.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
def testRecordsCountAllowEmptyNS(self):
dns.rdatatype.A,
'127.0.0.1'))
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
class TestRecordsCountNoOPTInAR(DNSDistTest):
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.REFUSED)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
def testRecordsCountAllowNoOPTInAR(self):
"""
dns.rdatatype.A,
'127.0.0.1'))
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testRecordsCountAllowTwoARButNoOPT(self):
"""
dns.rdatatype.A,
'127.0.0.1'))
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
query = dns.message.make_query(name, 'A', 'IN')
response = dns.message.make_response(query)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, None)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, None)
def testNotDropped(self):
"""
query = dns.message.make_query(name, 'A', 'IN')
response = dns.message.make_response(query)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
class TestResponseRuleQNameAllowed(DNSDistTest):
query = dns.message.make_query(name, 'A', 'IN')
response = dns.message.make_response(query)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testNotAllowed(self):
"""
query = dns.message.make_query(name, 'A', 'IN')
response = dns.message.make_response(query)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, None)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, None)
class TestResponseRuleEditTTL(DNSDistTest):
'192.0.2.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
- self.assertNotEquals(response.answer[0].ttl, receivedResponse.answer[0].ttl)
- self.assertEquals(receivedResponse.answer[0].ttl, self._ttl)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
- self.assertNotEquals(response.answer[0].ttl, receivedResponse.answer[0].ttl)
- self.assertEquals(receivedResponse.answer[0].ttl, self._ttl)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+ self.assertNotEquals(response.answer[0].ttl, receivedResponse.answer[0].ttl)
+ self.assertEquals(receivedResponse.answer[0].ttl, self._ttl)
class TestResponseLuaActionReturnSyntax(DNSDistTest):
query = dns.message.make_query(name, 'A', 'IN')
response = dns.message.make_response(query)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, None)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(receivedResponse, None)
'192.0.2.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testDefaultPool(self):
"""
name = 'notpool.routing.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN')
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, None)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, None)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, None)
class TestRoutingQPSPoolRouting(DNSDistTest):
_config_template = """
response.answer.append(rrset)
for _ in range(numberOfQueries):
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
total = 0
if 'UDP Responder' in self._responsesCounter:
expectedResponse = dns.message.make_response(query)
expectedResponse.set_rcode(dns.rcode.SERVFAIL)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertEquals(receivedResponse, expectedResponse)
-
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
class TestRoutingWRandom(DNSDistTest):
'192.0.2.1')
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(expectedResponse, receivedResponse)
def testSpoofActionAAAA(self):
"""
'2001:DB8::1')
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(expectedResponse, receivedResponse)
def testSpoofActionCNAME(self):
"""
'cnameaction.spoofing.tests.powerdns.com.')
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(expectedResponse, receivedResponse)
def testSpoofActionMultiA(self):
"""
'192.0.2.2', '192.0.2.1')
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(expectedResponse, receivedResponse)
def testSpoofActionMultiAAAA(self):
"""
'2001:DB8::1', '2001:DB8::2')
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(expectedResponse, receivedResponse)
def testSpoofActionMultiANY(self):
"""
'2001:DB8::1', '2001:DB8::2')
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(expectedResponse, receivedResponse)
class TestSpoofingLuaSpoof(DNSDistTest):
'192.0.2.1', '192.0.2.2')
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(expectedResponse, receivedResponse)
def testLuaSpoofAAAA(self):
"""
'2001:DB8::1')
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(expectedResponse, receivedResponse)
def testLuaSpoofAWithCNAME(self):
"""
'spoofedcname.spoofing.tests.powerdns.com.')
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(expectedResponse, receivedResponse)
def testLuaSpoofAAAAWithCNAME(self):
"""
'spoofedcname.spoofing.tests.powerdns.com.')
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(expectedResponse, receivedResponse)
class TestSpoofingLuaWithStatistics(DNSDistTest):
self.assertTrue(receivedResponse)
self.assertEquals(expectedResponse2, receivedResponse)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponseAfterwards, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponseAfterwards, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(expectedResponseAfterwards, receivedResponse)
'127.0.0.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testQuestionMatchTagAndValue(self):
"""
'1.2.3.50')
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(expectedResponse, receivedResponse)
def testQuestionMatchTagOnly(self):
"""
'1.2.3.100')
expectedResponse.answer.append(rrset)
- (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
- self.assertTrue(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (_, receivedResponse) = sender(query, response=None, useQueue=False)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(expectedResponse, receivedResponse)
def testResponseNoMatch(self):
"""
'192.0.2.1')
response.answer.append(rrset)
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
def testResponseMatchTagAndValue(self):
"""
# we will set TC if the tag matches
expectedResponse.flags |= dns.flags.TC
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- print(expectedResponse)
- print(receivedResponse)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(expectedResponse, receivedResponse)
def testResponseMatchResponseTagMatches(self):
"""
# we will set QR=0 if the tag matches
expectedResponse.flags &= ~dns.flags.QR
- (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(expectedResponse, receivedResponse)
-
- (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
- self.assertTrue(receivedQuery)
- self.assertTrue(receivedResponse)
- receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(expectedResponse, receivedResponse)
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(expectedResponse, receivedResponse)
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(receivedQuery, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
(receivedQuery, receivedResponse) = sender(query, response)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
(_, receivedResponse) = sender(query, response)
self.assertTrue(receivedResponse)
self.assertEquals(receivedResponse, expectedResponse)
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
(receivedQuery, receivedResponse) = sender(query, response)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
sender = getattr(self, method)
# Verify that queries with no trailing data make it through.
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
(receivedQuery, receivedResponse) = sender(query, response)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
self.assertEquals(response, receivedResponse)
# Verify that queries with trailing data don't make it through.
- # (_, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (_, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(_, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertEquals(receivedResponse, None)
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(receivedQuery, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(_, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertTrue(receivedResponse)
expectedResponse.flags = receivedResponse.flags
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(_, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertTrue(receivedResponse)
expectedResponse.flags = receivedResponse.flags
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(_, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertTrue(receivedResponse)
expectedResponse.flags = receivedResponse.flags
for method in ("sendUDPQuery", "sendTCPQuery"):
sender = getattr(self, method)
- # (receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
- # (receivedQuery, receivedResponse) = self.sendTCPQuery(raw, response, rawQuery=True)
(_, receivedResponse) = sender(raw, response, rawQuery=True)
self.assertTrue(receivedResponse)
expectedResponse.flags = receivedResponse.flags
projects / thirdparty / pdns.git / commitdiff
