.history
dnsdist.log
/*_pb2.py
+/__pycache__/
import dns.message
import libnacl
import libnacl.utils
+import binascii
+
class DNSCryptResolverCertificate(object):
DNSCRYPT_CERT_MAGIC = '\x44\x4e\x53\x43'
def __init__(self, providerName, providerFingerprint, resolverAddress, resolverPort=443, timeout=2):
self._providerName = providerName
- self._providerFingerprint = providerFingerprint.lower().replace(':', '').decode('hex')
+ self._providerFingerprint = binascii.unhexlify(providerFingerprint.lower().replace(':', ''))
self._resolverAddress = resolverAddress
self._resolverPort = resolverPort
self._resolverCertificates = []
#!/usr/bin/env python2
import copy
-import Queue
import os
import socket
import ssl
import libnacl
import libnacl.utils
+# Python2/3 compatibility hacks
+if sys.version_info[0] == 2:
+ from Queue import Queue
+ range = xrange
+else:
+ from queue import Queue
+ range = range # allow re-export of the builtin name
+
+
class DNSDistTest(unittest.TestCase):
"""
Set up a dnsdist instance and responder threads.
_dnsDistPort = 5340
_dnsDistListeningAddr = "127.0.0.1"
_testServerPort = 5350
- _toResponderQueue = Queue.Queue()
- _fromResponderQueue = Queue.Queue()
+ _toResponderQueue = Queue()
+ _fromResponderQueue = Queue()
_queueTimeout = 1
_dnsdistStartupDelay = 2.0
_dnsdist = None
@classmethod
def _encryptConsole(cls, command, nonce):
+ command = command.encode('UTF-8')
if cls._consoleKey is None:
return command
return libnacl.crypto_secretbox(command, nonce, cls._consoleKey)
@classmethod
def _decryptConsole(cls, command, nonce):
if cls._consoleKey is None:
- return command
- return libnacl.crypto_secretbox_open(command, nonce, cls._consoleKey)
+ result = command
+ else:
+ result = libnacl.crypto_secretbox_open(command, nonce, cls._consoleKey)
+ return result.decode('UTF-8')
@classmethod
def sendConsoleCommand(cls, command, timeout=1.0):
print("Received a nonce of size %d, expecting %d, console command will not be sent!" % (len(theirNonce), len(ourNonce)))
return None
- halfNonceSize = len(ourNonce) / 2
+ halfNonceSize = int(len(ourNonce) / 2)
readingNonce = ourNonce[0:halfNonceSize] + theirNonce[halfNonceSize:]
writingNonce = theirNonce[0:halfNonceSize] + ourNonce[halfNonceSize:]
msg = cls._encryptConsole(command, writingNonce)
def compareOptions(self, a, b):
self.assertEquals(len(a), len(b))
- for idx in xrange(len(a)):
+ for idx in range(len(a)):
self.assertEquals(a[idx], b[idx])
def checkMessageNoEDNS(self, expected, received):
#!/usr/bin/env bash
set -e
-PYTHON=${PYTHON:-python2}
-
if [ ! -d .venv ]; then
- virtualenv -p ${PYTHON} .venv
+ if [ -z "$PYTHON" ]; then
+ if [ ! -z "$(python3 --version | egrep '^Python 3.[6789]' 2>/dev/null)" ]; then
+ # found python3.6 or better
+ PYTHON=python3
+ else
+ # until we have better Linux distribution detection.
+ PYTHON=python2
+ fi
+ fi
+
+ virtualenv -p ${PYTHON} .venv
fi
. .venv/bin/activate
python -V
protoc -I=../pdns/ --python_out=. ../pdns/dnstap.proto
if [ -z "${DNSDISTBIN}" ]; then
- DNSDISTBIN=$(ls ../pdns/dnsdistdist/dnsdist-*/dnsdist)
+ DNSDISTBIN=$(ls ../pdns/dnsdistdist/dnsdist-*/dnsdist)
fi
export DNSDISTBIN
configFile = self._APIWriteDir + '/' + 'acl.conf'
self.assertTrue(os.path.isfile(configFile))
fileContent = None
- with file(configFile) as f:
+ with open(configFile, 'rt') as f:
fileContent = f.read()
self.assertEquals(fileContent, """-- Generated by the REST API, DO NOT EDIT
class TestStatNodeRespRingSince(DNSDistTest):
_consoleKey = DNSDistTest.generateConsoleKey()
- _consoleKeyB64 = base64.b64encode(_consoleKey)
+ _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
_config_params = ['_consoleKeyB64', '_consolePort', '_testServerPort']
_config_template = """
setKey("%s")
self.sendConsoleCommand("nodesSeen = {}")
self.sendConsoleCommand("statNodeRespRing(visitor)")
nodes = self.sendConsoleCommand("str = '' for key,value in pairs(nodesSeen) do str = str..value..\"\\n\" end return str")
- nodes = string.strip(nodes, "\n")
+ nodes = nodes.strip("\n")
self.assertEquals(nodes, """statnodesince.advanced.tests.powerdns.com.
advanced.tests.powerdns.com.
tests.powerdns.com.
self.sendConsoleCommand("nodesSeen = {}")
self.sendConsoleCommand("statNodeRespRing(visitor, 0)")
nodes = self.sendConsoleCommand("str = '' for key,value in pairs(nodesSeen) do str = str..value..\"\\n\" end return str")
- nodes = string.strip(nodes, "\n")
+ nodes = nodes.strip("\n")
self.assertEquals(nodes, """statnodesince.advanced.tests.powerdns.com.
advanced.tests.powerdns.com.
tests.powerdns.com.
self.sendConsoleCommand("nodesSeen = {}")
self.sendConsoleCommand("statNodeRespRing(visitor)")
nodes = self.sendConsoleCommand("str = '' for key,value in pairs(nodesSeen) do str = str..value..\"\\n\" end return str")
- nodes = string.strip(nodes, "\n")
+ nodes = nodes.strip("\n")
self.assertEquals(nodes, """statnodesince.advanced.tests.powerdns.com.
advanced.tests.powerdns.com.
tests.powerdns.com.
self.sendConsoleCommand("nodesSeen = {}")
self.sendConsoleCommand("statNodeRespRing(visitor, 5)")
nodes = self.sendConsoleCommand("str = '' for key,value in pairs(nodesSeen) do str = str..value..\"\\n\" end return str")
- nodes = string.strip(nodes, "\n")
+ nodes = nodes.strip("\n")
self.assertEquals(nodes, """""")
self.sendConsoleCommand("nodesSeen = {}")
self.sendConsoleCommand("statNodeRespRing(visitor, 10)")
nodes = self.sendConsoleCommand("str = '' for key,value in pairs(nodesSeen) do str = str..value..\"\\n\" end return str")
- nodes = string.strip(nodes, "\n")
+ nodes = nodes.strip("\n")
self.assertEquals(nodes, """statnodesince.advanced.tests.powerdns.com.
advanced.tests.powerdns.com.
tests.powerdns.com.
class TestCachingNoStale(DNSDistTest):
_consoleKey = DNSDistTest.generateConsoleKey()
- _consoleKeyB64 = base64.b64encode(_consoleKey)
+ _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
_config_params = ['_consoleKeyB64', '_consolePort', '_testServerPort']
_config_template = """
pc = newPacketCache(100, 86400, 1)
class TestCachingStale(DNSDistTest):
_consoleKey = DNSDistTest.generateConsoleKey()
- _consoleKeyB64 = base64.b64encode(_consoleKey)
+ _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
_staleCacheTTL = 60
_config_params = ['_staleCacheTTL', '_consoleKeyB64', '_consolePort', '_testServerPort']
_config_template = """
class TestCacheManagement(DNSDistTest):
_consoleKey = DNSDistTest.generateConsoleKey()
- _consoleKeyB64 = base64.b64encode(_consoleKey)
+ _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
_config_params = ['_consoleKeyB64', '_consolePort', '_testServerPort']
_config_template = """
pc = newPacketCache(100, 86400, 1)
#!/usr/bin/env python
-import Queue
import threading
import socket
import sys
import time
-from dnsdisttests import DNSDistTest
+from dnsdisttests import DNSDistTest, Queue
class TestCarbon(DNSDistTest):
_carbonServer1Name = "carbonname1"
_carbonServer2Port = 8001
_carbonServer2Name = "carbonname2"
- _carbonQueue1 = Queue.Queue()
- _carbonQueue2 = Queue.Queue()
+ _carbonQueue1 = Queue()
+ _carbonQueue2 = Queue()
_carbonInterval = 2
_carbonCounters = {}
_config_params = ['_carbonServer1Port', '_carbonServer1Name', '_carbonInterval', '_carbonServer2Port', '_carbonServer2Name', '_carbonInterval']
while True:
(conn, _) = sock.accept()
conn.settimeout(2.0)
- lines = ""
+ lines = b''
while True:
data = conn.recv(4096)
if not data:
self.assertTrue(data1)
self.assertTrue(len(data1.splitlines()) > 1)
- expectedStart = "dnsdist." + self._carbonServer1Name + ".main."
+ expectedStart = b"dnsdist.%s.main." % self._carbonServer1Name.encode('UTF-8')
for line in data1.splitlines():
self.assertTrue(line.startswith(expectedStart))
- parts = line.split(' ')
+ parts = line.split(b' ')
self.assertEquals(len(parts), 3)
self.assertTrue(parts[1].isdigit())
self.assertTrue(parts[2].isdigit())
self.assertTrue(data2)
self.assertTrue(len(data2.splitlines()) > 1)
- expectedStart = "dnsdist." + self._carbonServer2Name + ".main."
+ expectedStart = b"dnsdist.%s.main." % self._carbonServer2Name.encode('UTF-8')
for line in data2.splitlines():
self.assertTrue(line.startswith(expectedStart))
- parts = line.split(' ')
+ parts = line.split(b' ')
self.assertEquals(len(parts), 3)
self.assertTrue(parts[1].isdigit())
self.assertTrue(parts[2].isdigit())
_dnsDistPortDNSCrypt = 8443
_consoleKey = DNSDistTest.generateConsoleKey()
- _consoleKeyB64 = base64.b64encode(_consoleKey)
+ _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
_providerFingerprint = 'E1D7:2108:9A59:BF8D:F101:16FA:ED5E:EA6A:9F6C:C78F:7F91:AF6B:027E:62F4:69C3:B1AA'
_providerName = "2.provider.name"
#!/usr/bin/env python
-import Queue
import threading
import os
import socket
import struct
import sys
import time
-from dnsdisttests import DNSDistTest
+from dnsdisttests import DNSDistTest, Queue
import dns
import dnstap_pb2
def checkDnstapBase(testinstance, dnstap, protocol, initiator):
testinstance.assertTrue(dnstap)
testinstance.assertTrue(dnstap.HasField('identity'))
- testinstance.assertEqual(dnstap.identity, 'a.server')
+ testinstance.assertEqual(dnstap.identity, b'a.server')
testinstance.assertTrue(dnstap.HasField('version'))
- testinstance.assertIn('dnsdist ', dnstap.version)
+ testinstance.assertIn(b'dnsdist ', dnstap.version)
testinstance.assertTrue(dnstap.HasField('type'))
testinstance.assertEqual(dnstap.type, dnstap.MESSAGE)
testinstance.assertTrue(dnstap.HasField('message'))
class TestDnstapOverRemoteLogger(DNSDistTest):
_remoteLoggerServerPort = 4243
- _remoteLoggerQueue = Queue.Queue()
+ _remoteLoggerQueue = Queue()
_remoteLoggerCounter = 0
_config_params = ['_testServerPort', '_remoteLoggerServerPort']
_config_template = """
# check the dnstap message corresponding to the UDP query
dnstap = self.getFirstDnstap()
checkDnstapQuery(self, dnstap, dnstap_pb2.UDP, query)
- checkDnstapExtra(self, dnstap, "Type,Query")
+ checkDnstapExtra(self, dnstap, b"Type,Query")
# check the dnstap message corresponding to the UDP response
dnstap = self.getFirstDnstap()
checkDnstapResponse(self, dnstap, dnstap_pb2.UDP, response)
- checkDnstapExtra(self, dnstap, "Type,Response")
+ checkDnstapExtra(self, dnstap, b"Type,Response")
(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
self.assertTrue(receivedQuery)
# check the dnstap message corresponding to the TCP query
dnstap = self.getFirstDnstap()
checkDnstapQuery(self, dnstap, dnstap_pb2.TCP, query)
- checkDnstapExtra(self, dnstap, "Type,Query")
+ checkDnstapExtra(self, dnstap, b"Type,Query")
# check the dnstap message corresponding to the TCP response
dnstap = self.getFirstDnstap()
checkDnstapResponse(self, dnstap, dnstap_pb2.TCP, response)
- checkDnstapExtra(self, dnstap, "Type,Response")
+ checkDnstapExtra(self, dnstap, b"Type,Response")
def fstrm_get_control_frame_type(data):
def fstrm_make_control_frame_reply(cft, data):
if cft == FSTRM_CONTROL_READY:
# Reply with ACCEPT frame and content-type
- contenttype = 'protobuf:dnstap.Dnstap'
+ contenttype = b'protobuf:dnstap.Dnstap'
frame = struct.pack('!LLL', FSTRM_CONTROL_ACCEPT, 1,
len(contenttype)) + contenttype
buf = struct.pack("!LL", 0, len(frame)) + frame
class TestDnstapOverFrameStreamUnixLogger(DNSDistTest):
_fstrmLoggerAddress = '/tmp/fslutest.sock'
- _fstrmLoggerQueue = Queue.Queue()
+ _fstrmLoggerQueue = Queue()
_fstrmLoggerCounter = 0
_config_params = ['_testServerPort', '_fstrmLoggerAddress']
_config_template = """
class TestDnstapOverFrameStreamTcpLogger(DNSDistTest):
_fstrmLoggerPort = 4000
- _fstrmLoggerQueue = Queue.Queue()
+ _fstrmLoggerQueue = Queue()
_fstrmLoggerCounter = 0
_config_params = ['_testServerPort', '_fstrmLoggerPort']
_config_template = """
import base64
import time
import dns
-from dnsdisttests import DNSDistTest
+from dnsdisttests import DNSDistTest, range
class TestDynBlockQPS(DNSDistTest):
allowed = 0
sent = 0
- for _ in xrange((self._dynBlockQPS * self._dynBlockPeriod) + 1):
+ for _ in range((self._dynBlockQPS * self._dynBlockPeriod) + 1):
(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
sent = sent + 1
if receivedQuery:
# again, over TCP this time
allowed = 0
sent = 0
- for _ in xrange((self._dynBlockQPS * self._dynBlockPeriod) + 1):
+ for _ in range((self._dynBlockQPS * self._dynBlockPeriod) + 1):
(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
sent = sent + 1
if receivedQuery:
allowed = 0
sent = 0
- for _ in xrange((self._dynBlockQPS * self._dynBlockPeriod) + 1):
+ for _ in range((self._dynBlockQPS * self._dynBlockPeriod) + 1):
(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
sent = sent + 1
if receivedQuery:
allowed = 0
sent = 0
# again, over TCP this time
- for _ in xrange((self._dynBlockQPS * self._dynBlockPeriod) + 1):
+ for _ in range((self._dynBlockQPS * self._dynBlockPeriod) + 1):
(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
sent = sent + 1
if receivedQuery:
allowed = 0
sent = 0
- for _ in xrange((self._dynBlockQPS * self._dynBlockPeriod) + 1):
+ for _ in range((self._dynBlockQPS * self._dynBlockPeriod) + 1):
(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
sent = sent + 1
if receivedQuery:
allowed = 0
sent = 0
# again, over TCP this time
- for _ in xrange((self._dynBlockQPS * self._dynBlockPeriod) + 1):
+ for _ in range((self._dynBlockQPS * self._dynBlockPeriod) + 1):
(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
sent = sent + 1
if receivedQuery:
allowed = 0
sent = 0
- for _ in xrange((self._dynBlockQPS * self._dynBlockPeriod) + 1):
+ for _ in range((self._dynBlockQPS * self._dynBlockPeriod) + 1):
(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
sent = sent + 1
if receivedQuery:
allowed = 0
sent = 0
# again, over TCP this time, we should never get truncated!
- for _ in xrange((self._dynBlockQPS * self._dynBlockPeriod) + 1):
+ for _ in range((self._dynBlockQPS * self._dynBlockPeriod) + 1):
(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
sent = sent + 1
self.assertEquals(query, receivedQuery)
servfailResponse.set_rcode(dns.rcode.SERVFAIL)
# start with normal responses
- for _ in xrange((self._dynBlockQPS * self._dynBlockPeriod) + 1):
+ for _ in range((self._dynBlockQPS * self._dynBlockPeriod) + 1):
(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
receivedQuery.id = query.id
self.assertEquals(query, receivedQuery)
# now with ServFail!
sent = 0
allowed = 0
- for _ in xrange((self._dynBlockQPS * self._dynBlockPeriod) + 1):
+ for _ in range((self._dynBlockQPS * self._dynBlockPeriod) + 1):
(receivedQuery, receivedResponse) = self.sendUDPQuery(query, servfailResponse)
sent = sent + 1
if receivedQuery:
# again, over TCP this time
# start with normal responses
- for _ in xrange((self._dynBlockQPS * self._dynBlockPeriod) + 1):
+ for _ in range((self._dynBlockQPS * self._dynBlockPeriod) + 1):
(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
receivedQuery.id = query.id
self.assertEquals(query, receivedQuery)
# now with ServFail!
sent = 0
allowed = 0
- for _ in xrange((self._dynBlockQPS * self._dynBlockPeriod) + 1):
+ for _ in range((self._dynBlockQPS * self._dynBlockPeriod) + 1):
(receivedQuery, receivedResponse) = self.sendTCPQuery(query, servfailResponse)
sent = sent + 1
if receivedQuery:
_dynBlockPeriod = 2
_dynBlockDuration = 5
_consoleKey = DNSDistTest.generateConsoleKey()
- _consoleKeyB64 = base64.b64encode(_consoleKey)
+ _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
_config_params = ['_consoleKeyB64', '_consolePort', '_dynBlockBytesPerSecond', '_dynBlockPeriod', '_dynBlockDuration', '_testServerPort']
_config_template = """
setKey("%s")
print(time.time())
- for _ in xrange(self._dynBlockBytesPerSecond * 5 / len(response.to_wire())):
+ for _ in range(int(self._dynBlockBytesPerSecond * 5 / len(response.to_wire()))):
(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
sent = sent + len(response.to_wire())
if receivedQuery:
# again, over TCP this time
allowed = 0
sent = 0
- for _ in xrange(self._dynBlockBytesPerSecond * 5 / len(response.to_wire())):
+ for _ in range(int(self._dynBlockBytesPerSecond * 5 / len(response.to_wire()))):
(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
sent = sent + len(response.to_wire())
if receivedQuery:
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096)
expectedQuery = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, options=[ecso])
response = dns.message.make_response(expectedQuery)
- ecoResponse = cookiesoption.CookiesOption('deadbeef', 'deadbeef')
+ 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])
expectedResponse = dns.message.make_response(query)
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096)
expectedQuery = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, options=[ecso])
response = dns.message.make_response(expectedQuery)
- ecoResponse = cookiesoption.CookiesOption('deadbeef', 'deadbeef')
+ 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)
query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096)
expectedQuery = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, options=[ecso])
response = dns.message.make_response(expectedQuery)
- ecoResponse = cookiesoption.CookiesOption('deadbeef', 'deadbeef')
+ 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)
#!/usr/bin/env python
-import Queue
import threading
import socket
import struct
import sys
import time
-from dnsdisttests import DNSDistTest
+from dnsdisttests import DNSDistTest, Queue
import dns
import dnsmessage_pb2
class TestProtobuf(DNSDistTest):
_protobufServerPort = 4242
- _protobufQueue = Queue.Queue()
+ _protobufQueue = Queue()
_protobufCounter = 0
_config_params = ['_testServerPort', '_protobufServerPort']
_config_template = """
"""
def _checkStatsValues(self, results, queriesCountersValue):
- for i in range(1, 5) + range(6, 20) + range(24, 35) + [ 35 ] :
+ for i in list(range(1, 5)) + list(range(6, 20)) + list(range(24, 35)) + [ 35 ] :
oid = self._snmpOID + '.1.' + str(i) + '.0'
self.assertTrue(oid in results)
self.assertTrue(isinstance(results[oid], Counter64))
self.assertEquals(results[oid], queriesCountersValue)
# the others counters (except for latency ones) should still be at 0
- for i in range(3, 5) + range(6, 14) + [26, 27, 29, 30, 31, 35, 36]:
+ for i in [3, 4, 6, 7, 8, 9, 10, 11, 12, 13, 26, 27, 29, 30, 31, 35, 36]:
oid = self._snmpOID + '.1.' + str(i) + '.0'
self.assertEquals(results[oid], 0)
import struct
import time
import dns
-from dnsdisttests import DNSDistTest
+from dnsdisttests import DNSDistTest, range
class TestTCPKeepAlive(DNSDistTest):
"""
conn = self.openTCPConnection()
count = 0
- for idx in xrange(5):
+ for idx in range(5):
try:
self.sendTCPQueryOverConnection(conn, query)
response = self.recvTCPResponseOverConnection(conn)
conn = self.openTCPConnection()
count = 0
- for idx in xrange(5):
+ for idx in range(5):
try:
self.sendTCPQueryOverConnection(conn, query)
response = self.recvTCPResponseOverConnection(conn)
conn = self.openTCPConnection()
count = 0
- for idx in xrange(5):
+ for idx in range(5):
try:
self.sendTCPQueryOverConnection(conn, query)
response = self.recvTCPResponseOverConnection(conn)
conn = self.openTCPConnection()
count = 0
- for idx in xrange(5):
+ for idx in range(5):
try:
self.sendTCPQueryOverConnection(conn, query)
response = self.recvTCPResponseOverConnection(conn)
conn = self.openTCPConnection()
count = 0
- for idx in xrange(5):
+ for idx in range(5):
try:
self.sendTCPQueryOverConnection(conn, query)
response = self.recvTCPResponseOverConnection(conn)
conn = self.openTCPConnection()
count = 0
- for idx in xrange(5):
+ for idx in range(5):
try:
self.sendTCPQueryOverConnection(conn, query)
response = self.recvTCPResponseOverConnection(conn)
conn = self.openTCPConnection()
count = 0
- for idx in xrange(5):
+ for idx in range(5):
try:
self.sendTCPQueryOverConnection(conn, query)
response = self.recvTCPResponseOverConnection(conn)
import struct
import time
import dns
-from dnsdisttests import DNSDistTest
+from dnsdisttests import DNSDistTest, range
class TestTCPLimits(DNSDistTest):
conn = self.openTCPConnection()
count = 0
- for idx in xrange(self._maxTCPQueriesPerConn):
+ for idx in range(self._maxTCPQueriesPerConn):
try:
self.sendTCPQueryOverConnection(conn, query)
response = self.recvTCPResponseOverConnection(conn)
query = dns.message.make_query(name, 'A', 'IN')
conns = []
- for idx in xrange(self._maxTCPConnsPerClient + 1):
+ for idx in range(self._maxTCPConnsPerClient + 1):
conns.append(self.openTCPConnection())
count = 0
#!/usr/bin/env python
import base64
-import Queue
import threading
import clientsubnetoption
import dns
-from dnsdisttests import DNSDistTest
+from dnsdisttests import DNSDistTest, Queue
class TestTeeAction(DNSDistTest):
_consoleKey = DNSDistTest.generateConsoleKey()
- _consoleKeyB64 = base64.b64encode(_consoleKey)
+ _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
_teeServerPort = 5390
- _toTeeQueue = Queue.Queue()
- _fromTeeQueue = Queue.Queue()
+ _toTeeQueue = Queue()
+ _fromTeeQueue = Queue()
_config_template = """
setKey("%s")
controlSocket("127.0.0.1:%s")
response.answer.append(rrset)
raw = query.to_wire()
- raw = raw + 'A'* 20
+ raw = raw + b'A'* 20
(receivedQuery, receivedResponse) = self.sendUDPQuery(raw, response, rawQuery=True)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
query = dns.message.make_query(name, 'AAAA', 'IN')
raw = query.to_wire()
- raw = raw + 'A'* 20
+ raw = raw + b'A'* 20
(_, receivedResponse) = self.sendUDPQuery(raw, response=None, rawQuery=True)
self.assertEquals(receivedResponse, None)
projects / thirdparty / pdns.git / commitdiff
