_backgroundThreads = {}
_UDPResponder = None
_TCPResponder = None
+ _extraStartupSleep = 0
@classmethod
def waitForTCPSocket(cls, ipaddress, port):
print(fdLog.read())
print(f"*** End startDNSDist log for {logFile} ***")
raise AssertionError('%s failed (%d)' % (dnsdistcmd, cls._dnsdist.returncode))
+ time.sleep(cls._extraStartupSleep)
@classmethod
def setUpSockets(cls):
class TestTempFailureCacheTTLAction(DNSDistTest):
+ _extraStartupSleep = 1
_config_template = """
pc = newPacketCache(100, {maxTTL=86400, minTTL=1})
getPool(""):setCache(pc)
class TestCarbon(DNSDistTest):
+ _extraStartupSleep = 2
_carbonServer1Port = 8000
_carbonServer1Name = "carbonname1"
_carbonServer2Port = 8001
cls._CarbonResponder2.setDaemon(True)
cls._CarbonResponder2.start()
- @classmethod
- def setUpClass(cls):
-
- cls.startResponders()
- cls.startDNSDist()
- cls.setUpSockets()
- time.sleep(1)
-
- print("Launching tests..")
-
-
def testCarbon(self):
"""
Carbon: send data to 2 carbon servers
self.assertTrue(len(data1.splitlines()) > 1)
expectedStart = b"dnsdist.%s.main." % self._carbonServer1Name.encode('UTF-8')
for line in data1.splitlines():
+ print(line, file=sys.stderr)
self.assertTrue(line.startswith(expectedStart))
parts = line.split(b' ')
self.assertEqual(len(parts), 3)
self.assertTrue(len(data2.splitlines()) > 1)
expectedStart = b"dnsdist.%s.main." % self._carbonServer2Name.encode('UTF-8')
for line in data2.splitlines():
+ print(line, file=sys.stderr)
self.assertTrue(line.startswith(expectedStart))
parts = line.split(b' ')
self.assertEqual(len(parts), 3)
_dohHealthCheckQueries = 0
class TestLazyHealthChecks(HealthCheckTest):
+ _extraStartupSleep = 1
_do53Port = 10700
_dotPort = 10701
_dohPort = 10702
class TestResponseRuleERCode(DNSDistTest):
+ _extraStartupSleep = 1
_config_template = """
newServer{address="127.0.0.1:%s"}
addResponseAction(ERCodeRule(DNSRCode.BADVERS), DelayResponseAction(1000))
from dnsdisttests import DNSDistTest
class TestSNMP(DNSDistTest):
-
+ # wait 1s so that the uptime is > 0
+ _extraStartupSleep = 1
_snmpTimeout = 2.0
_snmpServer = '127.0.0.1'
_snmpPort = 161
return results
def _checkStats(self, auth, name):
- # wait 1s so that the uptime is > 0
- time.sleep(1)
results = self._getSNMPStats(auth)
self._checkStatsValues(results, self.__class__._queriesSent)
class TLSTests(object):
- @classmethod
- def setUpClass(cls):
-
- cls.startResponders()
- cls.startDNSDist()
- cls.setUpSockets()
- time.sleep(1)
-
- print("Launching tests..")
-
def getServerCertificate(self):
conn = self.openTLSConnection(self._tlsServerPort, self._serverName, self._caCert)
cert = conn.getpeercert()
class TestOpenSSL(DNSDistTest, TLSTests):
+ _extraStartupSleep = 1
_consoleKey = DNSDistTest.generateConsoleKey()
_consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
_serverKey = 'server.key'