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 = dns.message.make_response(expectedQuery)
ecsoResponse = clientsubnetoption.ClientSubnetOption('127.0.0.1', 24, scope=24)
response.use_edns(edns=True, payload=4096, options=[ecsoResponse])
- expectedResponse = dns.message.make_response(query)
+ expectedResponse = dns.message.make_response(query, our_payload=4096)
rrset = dns.rrset.from_text(name,
3600,
dns.rdataclass.IN,
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):
"""
ecoResponse = cookiesoption.CookiesOption(b'deadbeef', b'deadbeef')
ecsoResponse = clientsubnetoption.ClientSubnetOption('127.0.0.1', 24, scope=24)
response.use_edns(edns=True, payload=4096, options=[ecsoResponse, ecoResponse])
- expectedResponse = dns.message.make_response(query)
+ expectedResponse = dns.message.make_response(query, our_payload=4096)
rrset = dns.rrset.from_text(name,
3600,
dns.rdataclass.IN,
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):
"""
ecoResponse = cookiesoption.CookiesOption(b'deadbeef', b'deadbeef')
ecsoResponse = clientsubnetoption.ClientSubnetOption('127.0.0.1', 24, scope=24)
response.use_edns(edns=True, payload=4096, options=[ecoResponse, ecsoResponse, ecoResponse])
- expectedResponse = dns.message.make_response(query)
+ expectedResponse = dns.message.make_response(query, our_payload=4096)
rrset = dns.rrset.from_text(name,
3600,
dns.rdataclass.IN,
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):
"""
dns.rdatatype.A,
'127.0.0.1')
response.answer.append(rrset)
- expectedResponse = dns.message.make_response(query)
+ expectedResponse = dns.message.make_response(query, our_payload=4096)
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)