"Fossies" - the Fresh Open Source Software Archive

Member "pdns-auth-4.2.0/regression-tests.dnsdist/test_TCPLimits.py" (27 Aug 2019, 3729 Bytes) of package /linux/misc/dns/pdns-auth-4.2.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. For more information about "test_TCPLimits.py" see the Fossies "Dox" file reference documentation and the latest Fossies "Diffs" side-by-side code changes report: 4.1.13_vs_4.2.0.

    1 #!/usr/bin/env python
    2 import struct
    3 import time
    4 import dns
    5 from dnsdisttests import DNSDistTest
    6 
    7 try:
    8   range = xrange
    9 except NameError:
   10   pass
   11 
   12 class TestTCPLimits(DNSDistTest):
   13 
   14     # this test suite uses a different responder port
   15     # because it uses a different health check configuration
   16     _testServerPort = 5395
   17     _answerUnexpected = True
   18 
   19     _tcpIdleTimeout = 2
   20     _maxTCPQueriesPerConn = 5
   21     _maxTCPConnsPerClient = 3
   22     _maxTCPConnDuration = 5
   23     _config_template = """
   24     newServer{address="127.0.0.1:%s"}
   25     setTCPRecvTimeout(%s)
   26     setMaxTCPQueriesPerConnection(%s)
   27     setMaxTCPConnectionsPerClient(%s)
   28     setMaxTCPConnectionDuration(%s)
   29     """
   30     _config_params = ['_testServerPort', '_tcpIdleTimeout', '_maxTCPQueriesPerConn', '_maxTCPConnsPerClient', '_maxTCPConnDuration']
   31 
   32     def testTCPQueriesPerConn(self):
   33         """
   34         TCP Limits: Maximum number of queries
   35         """
   36         name = 'maxqueriesperconn.tcp.tests.powerdns.com.'
   37         query = dns.message.make_query(name, 'A', 'IN')
   38         conn = self.openTCPConnection()
   39 
   40         count = 0
   41         for idx in range(self._maxTCPQueriesPerConn):
   42             try:
   43                 self.sendTCPQueryOverConnection(conn, query)
   44                 response = self.recvTCPResponseOverConnection(conn)
   45                 self.assertTrue(response)
   46                 count = count + 1
   47             except:
   48                 pass
   49 
   50         # this one should fail
   51         failed = False
   52         try:
   53             self.sendTCPQueryOverConnection(conn, query)
   54             response = self.recvTCPResponseOverConnection(conn)
   55             self.assertFalse(response)
   56             if not response:
   57                 failed = True
   58             else:
   59                 count = count + 1
   60         except:
   61             failed = True
   62 
   63         conn.close()
   64         self.assertTrue(failed)
   65         self.assertEqual(count, self._maxTCPQueriesPerConn)
   66 
   67     def testTCPConnsPerClient(self):
   68         """
   69         TCP Limits: Maximum number of conns per client
   70         """
   71         name = 'maxconnsperclient.tcp.tests.powerdns.com.'
   72         query = dns.message.make_query(name, 'A', 'IN')
   73         conns = []
   74 
   75         for idx in range(self._maxTCPConnsPerClient + 1):
   76             conns.append(self.openTCPConnection())
   77 
   78         count = 0
   79         failed = 0
   80         for conn in conns:
   81             try:
   82                 self.sendTCPQueryOverConnection(conn, query)
   83                 response = self.recvTCPResponseOverConnection(conn)
   84                 if response:
   85                     count = count + 1
   86                 else:
   87                     failed = failed + 1
   88             except:
   89                 failed = failed + 1
   90 
   91         for conn in conns:
   92             conn.close()
   93 
   94         self.assertEqual(count, self._maxTCPConnsPerClient)
   95         self.assertEqual(failed, 1)
   96 
   97     def testTCPDuration(self):
   98         """
   99         TCP Limits: Maximum duration
  100         """
  101         name = 'duration.tcp.tests.powerdns.com.'
  102 
  103         start = time.time()
  104         conn = self.openTCPConnection()
  105         # immediately send the maximum size
  106         conn.send(struct.pack("!H", 65535))
  107 
  108         count = 0
  109         while count < (self._maxTCPConnDuration * 20):
  110             try:
  111                 # sleeping for only one second keeps us below the
  112                 # idle timeout (setTCPRecvTimeout())
  113                 time.sleep(0.1)
  114                 conn.send(b'A')
  115                 count = count + 1
  116             except Exception as e:
  117                 print("Exception: %s!" % (e))
  118                 break
  119 
  120         end = time.time()
  121 
  122         self.assertAlmostEquals(count / 10, self._maxTCPConnDuration, delta=2)
  123         self.assertAlmostEquals(end - start, self._maxTCPConnDuration, delta=2)
  124 
  125         conn.close()