import struct
import sys
import time
-from dnsdisttests import DNSDistTest, Queue
+from dnsdisttests import DNSDistTest, Queue, pickAvailablePort
import dns
import dnstap_pb2
testinstance.assertTrue(dnstap.message.HasField('socket_protocol'))
testinstance.assertEqual(dnstap.message.socket_protocol, protocol)
testinstance.assertTrue(dnstap.message.HasField('socket_family'))
- testinstance.assertEquals(dnstap.message.socket_family, dnstap_pb2.INET)
+ testinstance.assertEqual(dnstap.message.socket_family, dnstap_pb2.INET)
testinstance.assertTrue(dnstap.message.HasField('query_address'))
- testinstance.assertEquals(socket.inet_ntop(socket.AF_INET, dnstap.message.query_address), initiator)
+ testinstance.assertEqual(socket.inet_ntop(socket.AF_INET, dnstap.message.query_address), initiator)
testinstance.assertTrue(dnstap.message.HasField('response_address'))
- testinstance.assertEquals(socket.inet_ntop(socket.AF_INET, dnstap.message.response_address), initiator)
+ testinstance.assertEqual(socket.inet_ntop(socket.AF_INET, dnstap.message.response_address), initiator)
testinstance.assertTrue(dnstap.message.HasField('response_port'))
- testinstance.assertEquals(dnstap.message.response_port, testinstance._dnsDistPort)
+ testinstance.assertEqual(dnstap.message.response_port, testinstance._dnsDistPort)
def checkDnstapQuery(testinstance, dnstap, protocol, query, initiator='127.0.0.1'):
- testinstance.assertEquals(dnstap.message.type, dnstap_pb2.Message.CLIENT_QUERY)
+ testinstance.assertEqual(dnstap.message.type, dnstap_pb2.Message.CLIENT_QUERY)
checkDnstapBase(testinstance, dnstap, protocol, initiator)
testinstance.assertTrue(dnstap.message.HasField('query_time_sec'))
def checkDnstapResponse(testinstance, dnstap, protocol, response, initiator='127.0.0.1'):
- testinstance.assertEquals(dnstap.message.type, dnstap_pb2.Message.CLIENT_RESPONSE)
+ testinstance.assertEqual(dnstap.message.type, dnstap_pb2.Message.CLIENT_RESPONSE)
checkDnstapBase(testinstance, dnstap, protocol, initiator)
testinstance.assertTrue(dnstap.message.HasField('query_time_sec'))
class TestDnstapOverRemoteLogger(DNSDistTest):
- _remoteLoggerServerPort = 4243
+ _remoteLoggerServerPort = pickAvailablePort()
_remoteLoggerQueue = Queue()
_remoteLoggerCounter = 0
_config_params = ['_testServerPort', '_remoteLoggerServerPort']
DNSDistTest.startResponders()
cls._remoteLoggerListener = threading.Thread(name='RemoteLogger Listener', target=cls.RemoteLoggerListener, args=[cls._remoteLoggerServerPort])
- cls._remoteLoggerListener.setDaemon(True)
+ cls._remoteLoggerListener.daemon = True
cls._remoteLoggerListener.start()
def getFirstDnstap(self):
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ self.assertEqual(query, receivedQuery)
+ self.assertEqual(response, receivedResponse)
# give the dnstap messages time to get here
time.sleep(1)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ self.assertEqual(query, receivedQuery)
+ self.assertEqual(response, receivedResponse)
# give the dnstap messages time to get here
time.sleep(1)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ self.assertEqual(query, receivedQuery)
+ self.assertEqual(response, receivedResponse)
# give the dnstap messages time to get here
time.sleep(1)
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ self.assertEqual(query, receivedQuery)
+ self.assertEqual(response, receivedResponse)
# give the dnstap messages time to get here
time.sleep(1)
DNSDistTest.startResponders()
cls._fstrmLoggerListener = threading.Thread(name='FrameStreamUnixListener', target=cls.FrameStreamUnixListener, args=[cls._fstrmLoggerAddress])
- cls._fstrmLoggerListener.setDaemon(True)
+ cls._fstrmLoggerListener.daemon = True
cls._fstrmLoggerListener.start()
def getFirstDnstap(self):
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ self.assertEqual(query, receivedQuery)
+ self.assertEqual(response, receivedResponse)
# check the dnstap message corresponding to the UDP query
dnstap = self.getFirstDnstap()
class TestDnstapOverFrameStreamTcpLogger(DNSDistTest):
- _fstrmLoggerPort = 4000
+ _fstrmLoggerPort = pickAvailablePort()
_fstrmLoggerQueue = Queue()
_fstrmLoggerCounter = 0
_config_params = ['_testServerPort', '_fstrmLoggerPort']
DNSDistTest.startResponders()
cls._fstrmLoggerListener = threading.Thread(name='FrameStreamUnixListener', target=cls.FrameStreamUnixListener, args=[cls._fstrmLoggerPort])
- cls._fstrmLoggerListener.setDaemon(True)
+ cls._fstrmLoggerListener.daemon = True
cls._fstrmLoggerListener.start()
def getFirstDnstap(self):
self.assertTrue(receivedQuery)
self.assertTrue(receivedResponse)
receivedQuery.id = query.id
- self.assertEquals(query, receivedQuery)
- self.assertEquals(response, receivedResponse)
+ self.assertEqual(query, receivedQuery)
+ self.assertEqual(response, receivedResponse)
# check the dnstap message corresponding to the UDP query
dnstap = self.getFirstDnstap()