[thirdparty/pdns.git] / regression-tests.dnsdist / test_Routing.py

#!/usr/bin/env python
import base64
import threading
import time
import dns
from dnsdisttests import DNSDistTest

class TestRoutingPoolRouting(DNSDistTest):

    _config_template = """
    newServer{address="127.0.0.1:%s", pool="real"}
    addAction(makeRule("poolaction.routing.tests.powerdns.com"), PoolAction("real"))
    """

    def testPolicyPoolAction(self):
        """
        Routing: Set pool by qname via PoolAction

        Send an A query to "poolaction.routing.tests.powerdns.com.",
        check that dnsdist routes the query to the "real" pool.
        """
        name = 'poolaction.routing.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEquals(query, receivedQuery)
            self.assertEquals(response, receivedResponse)

    def testDefaultPool(self):
        """
        Routing: Set pool by qname canary

        Send an A query to "notpool.routing.tests.powerdns.com.",
        check that dnsdist sends no response (no servers
        in the default pool).
        """
        name = 'notpool.routing.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (_, receivedResponse) = sender(query, response=None, useQueue=False)
            self.assertEquals(receivedResponse, None)

class TestRoutingQPSPoolRouting(DNSDistTest):
    _config_template = """
    newServer{address="127.0.0.1:%s", pool="regular"}
    addAction(makeRule("qpspoolaction.routing.tests.powerdns.com"), QPSPoolAction(10, "regular"))
    """

    def testQPSPoolAction(self):
        """
        Routing: Set pool by QPS via action

        Send queries to "qpspoolaction.routing.tests.powerdns.com."
        check that dnsdist does not route the query to the "regular" pool
        when the max QPS has been reached.
        """
        maxQPS = 10
        name = 'qpspoolaction.routing.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)

        for _ in range(maxQPS):
            (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
            receivedQuery.id = query.id
            self.assertEquals(query, receivedQuery)
            self.assertEquals(response, receivedResponse)

        # we should now be sent to the "abuse" pool which is empty,
        # so the queries should be dropped
        (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
        self.assertEquals(receivedResponse, None)

        time.sleep(1)

        # again, over TCP this time
        for _ in range(maxQPS):
            (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
            receivedQuery.id = query.id
            self.assertEquals(query, receivedQuery)
            self.assertEquals(response, receivedResponse)


        (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
        self.assertEquals(receivedResponse, None)


class TestRoutingRoundRobinLB(DNSDistTest):

    _testServer2Port = 5351
    _config_params = ['_testServerPort', '_testServer2Port']
    _config_template = """
    setServerPolicy(roundrobin)
    s1 = newServer{address="127.0.0.1:%s"}
    s1:setUp()
    s2 = newServer{address="127.0.0.1:%s"}
    s2:setUp()
    """

    @classmethod
    def startResponders(cls):
        print("Launching responders..")
        cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
        cls._UDPResponder.setDaemon(True)
        cls._UDPResponder.start()
        cls._UDPResponder2 = threading.Thread(name='UDP Responder 2', target=cls.UDPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
        cls._UDPResponder2.setDaemon(True)
        cls._UDPResponder2.start()

        cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
        cls._TCPResponder.setDaemon(True)
        cls._TCPResponder.start()

        cls._TCPResponder2 = threading.Thread(name='TCP Responder 2', target=cls.TCPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
        cls._TCPResponder2.setDaemon(True)
        cls._TCPResponder2.start()

    def testRR(self):
        """
        Routing: Round Robin

        Send 10 A queries to "rr.routing.tests.powerdns.com.",
        check that dnsdist routes half of it to each backend.
        """
        numberOfQueries = 10
        name = 'rr.routing.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)

        # the round robin counter is shared for UDP and TCP,
        # so we need to do UDP then TCP to have a clean count
        for _ in range(numberOfQueries):
            (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
            receivedQuery.id = query.id
            self.assertEquals(query, receivedQuery)
            self.assertEquals(response, receivedResponse)

        for _ in range(numberOfQueries):
            (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
            receivedQuery.id = query.id
            self.assertEquals(query, receivedQuery)
            self.assertEquals(response, receivedResponse)

        for key in self._responsesCounter:
            value = self._responsesCounter[key]
            self.assertEquals(value, numberOfQueries / 2)

class TestRoutingRoundRobinLBOneDown(DNSDistTest):

    _testServer2Port = 5351
    _config_params = ['_testServerPort', '_testServer2Port']
    _config_template = """
    setServerPolicy(roundrobin)
    s1 = newServer{address="127.0.0.1:%s"}
    s1:setUp()
    s2 = newServer{address="127.0.0.1:%s"}
    s2:setDown()
    """

    def testRRWithOneDown(self):
        """
        Routing: Round Robin with one server down

        Send 100 A queries to "rr.routing.tests.powerdns.com.",
        check that dnsdist routes all of it to the only backend up.
        """
        numberOfQueries = 10
        name = 'rr.routing.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)

        # the round robin counter is shared for UDP and TCP,
        # so we need to do UDP then TCP to have a clean count
        for _ in range(numberOfQueries):
            (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
            receivedQuery.id = query.id
            self.assertEquals(query, receivedQuery)
            self.assertEquals(response, receivedResponse)

        for _ in range(numberOfQueries):
            (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
            receivedQuery.id = query.id
            self.assertEquals(query, receivedQuery)
            self.assertEquals(response, receivedResponse)

        total = 0
        for key in self._responsesCounter:
            value = self._responsesCounter[key]
            self.assertTrue(value == numberOfQueries or value == 0)
            total += value

        self.assertEquals(total, numberOfQueries * 2)

class TestRoutingOrder(DNSDistTest):

    _testServer2Port = 5351
    _config_params = ['_testServerPort', '_testServer2Port']
    _config_template = """
    setServerPolicy(firstAvailable)
    s1 = newServer{address="127.0.0.1:%s", order=2}
    s1:setUp()
    s2 = newServer{address="127.0.0.1:%s", order=1}
    s2:setUp()
    """

    @classmethod
    def startResponders(cls):
        print("Launching responders..")
        cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
        cls._UDPResponder.setDaemon(True)
        cls._UDPResponder.start()
        cls._UDPResponder2 = threading.Thread(name='UDP Responder 2', target=cls.UDPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
        cls._UDPResponder2.setDaemon(True)
        cls._UDPResponder2.start()

        cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
        cls._TCPResponder.setDaemon(True)
        cls._TCPResponder.start()

        cls._TCPResponder2 = threading.Thread(name='TCP Responder 2', target=cls.TCPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
        cls._TCPResponder2.setDaemon(True)
        cls._TCPResponder2.start()

    def testOrder(self):
        """
        Routing: firstAvailable policy based on 'order'

        Send 50 A queries to "order.routing.tests.powerdns.com.",
        check that dnsdist routes all of it to the second backend
        because it has the lower order value.
        """
        numberOfQueries = 50
        name = 'order.routing.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)

        for _ in range(numberOfQueries):
            for method in ("sendUDPQuery", "sendTCPQuery"):
                sender = getattr(self, method)
                (receivedQuery, receivedResponse) = sender(query, response)
                receivedQuery.id = query.id
                self.assertEquals(query, receivedQuery)
                self.assertEquals(response, receivedResponse)

        total = 0
        if 'UDP Responder' in self._responsesCounter:
            self.assertEquals(self._responsesCounter['UDP Responder'], 0)
        self.assertEquals(self._responsesCounter['UDP Responder 2'], numberOfQueries)
        if 'TCP Responder' in self._responsesCounter:
            self.assertEquals(self._responsesCounter['TCP Responder'], 0)
        self.assertEquals(self._responsesCounter['TCP Responder 2'], numberOfQueries)

class TestRoutingNoServer(DNSDistTest):

    _config_template = """
    newServer{address="127.0.0.1:%s", pool="real"}
    setServFailWhenNoServer(true)
    """

    def testPolicyPoolNoServer(self):
        """
        Routing: No server should return ServFail
        """
        name = 'noserver.routing.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        expectedResponse = dns.message.make_response(query)
        expectedResponse.set_rcode(dns.rcode.SERVFAIL)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (_, receivedResponse) = sender(query, response=None, useQueue=False)
            self.assertEquals(receivedResponse, expectedResponse)

class TestRoutingWRandom(DNSDistTest):

    _testServer2Port = 5351
    _config_params = ['_testServerPort', '_testServer2Port']
    _config_template = """
    setServerPolicy(wrandom)
    s1 = newServer{address="127.0.0.1:%s", weight=1}
    s1:setUp()
    s2 = newServer{address="127.0.0.1:%s", weight=2}
    s2:setUp()
    """

    @classmethod
    def startResponders(cls):
        print("Launching responders..")
        cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
        cls._UDPResponder.setDaemon(True)
        cls._UDPResponder.start()
        cls._UDPResponder2 = threading.Thread(name='UDP Responder 2', target=cls.UDPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
        cls._UDPResponder2.setDaemon(True)
        cls._UDPResponder2.start()

        cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
        cls._TCPResponder.setDaemon(True)
        cls._TCPResponder.start()

        cls._TCPResponder2 = threading.Thread(name='TCP Responder 2', target=cls.TCPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
        cls._TCPResponder2.setDaemon(True)
        cls._TCPResponder2.start()

    def testWRandom(self):
        """
        Routing: WRandom

        Send 100 A queries to "wrandom.routing.tests.powerdns.com.",
        check that dnsdist routes less than half to one, more to the other.
        """
        numberOfQueries = 100
        name = 'wrandom.routing.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)

        # the counter is shared for UDP and TCP,
        # so we need to do UDP then TCP to have a clean count
        for _ in range(numberOfQueries):
            (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
            receivedQuery.id = query.id
            self.assertEquals(query, receivedQuery)
            self.assertEquals(response, receivedResponse)

        for _ in range(numberOfQueries):
            (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
            receivedQuery.id = query.id
            self.assertEquals(query, receivedQuery)
            self.assertEquals(response, receivedResponse)

        # The lower weight downstream should receive less than half the queries
        self.assertLess(self._responsesCounter['UDP Responder'], numberOfQueries * 0.50)
        self.assertLess(self._responsesCounter['TCP Responder'], numberOfQueries * 0.50)

        # The higher weight downstream should receive more than half the queries
        self.assertGreater(self._responsesCounter['UDP Responder 2'], numberOfQueries * 0.50)
        self.assertGreater(self._responsesCounter['TCP Responder 2'], numberOfQueries * 0.50)


class TestRoutingHighValueWRandom(DNSDistTest):

    _testServer2Port = 5351
    _consoleKey = DNSDistTest.generateConsoleKey()
    _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
    _config_params = ['_consoleKeyB64', '_consolePort', '_testServerPort', '_testServer2Port']
    _config_template = """
    setKey("%s")
    controlSocket("127.0.0.1:%s")
    setServerPolicy(wrandom)
    s1 = newServer{address="127.0.0.1:%s", weight=2000000000}
    s1:setUp()
    s2 = newServer{address="127.0.0.1:%s", weight=2000000000}
    s2:setUp()
    """

    @classmethod
    def startResponders(cls):
        print("Launching responders..")
        cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
        cls._UDPResponder.setDaemon(True)
        cls._UDPResponder.start()
        cls._UDPResponder2 = threading.Thread(name='UDP Responder 2', target=cls.UDPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
        cls._UDPResponder2.setDaemon(True)
        cls._UDPResponder2.start()

        cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
        cls._TCPResponder.setDaemon(True)
        cls._TCPResponder.start()

        cls._TCPResponder2 = threading.Thread(name='TCP Responder 2', target=cls.TCPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
        cls._TCPResponder2.setDaemon(True)
        cls._TCPResponder2.start()

    def testHighValueWRandom(self):
        """
        Routing: WRandom (overflow)

        Send 100 A queries to "wrandom-overflow.routing.tests.powerdns.com.",
        check that dnsdist routes to each downstream, rather than failing with
        no-policy.
        """
        numberOfQueries = 100
        name = 'wrandom-overflow.routing.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)

        # the counter is shared for UDP and TCP,
        # so we need to do UDP then TCP to have a clean count
        for _ in range(numberOfQueries):
            (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
            receivedQuery.id = query.id
            self.assertEquals(query, receivedQuery)
            self.assertEquals(response, receivedResponse)

        for _ in range(numberOfQueries):
            (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
            receivedQuery.id = query.id
            self.assertEquals(query, receivedQuery)
            self.assertEquals(response, receivedResponse)

        stats = self.sendConsoleCommand("dumpStats()").split()
        stats_dict = {}

        # Map to a dict with every other element being the value to the previous one
        for i, x in enumerate(stats):
            if not i % 2:
                stats_dict[x] = stats[i+1]

        # There should be no queries getting "no-policy" responses
        self.assertEquals(stats_dict['no-policy'], '0')

        # Each downstream should receive some queries, but it will be unbalanced
        # because the sum of the weights is higher than INT_MAX.
        # The first downstream will receive more than half the queries
        self.assertGreater(self._responsesCounter['UDP Responder'], numberOfQueries / 2)
        self.assertGreater(self._responsesCounter['TCP Responder'], numberOfQueries / 2)

        # The second downstream will receive the remainder of the queries, but it might very well be 0
        if 'UDP Responder 2' in self._responsesCounter:
            self.assertEquals(self._responsesCounter['UDP Responder 2'], numberOfQueries - self._responsesCounter['UDP Responder'])
        if 'TCP Responder 2' in self._responsesCounter:
            self.assertEquals(self._responsesCounter['TCP Responder 2'], numberOfQueries - self._responsesCounter['TCP Responder'])

class TestRoutingBadWeightWRandom(DNSDistTest):

    _testServer2Port = 5351
    _consoleKey = DNSDistTest.generateConsoleKey()
    _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
    _config_params = ['_consoleKeyB64', '_consolePort', '_testServerPort', '_testServer2Port']
    _config_template = """
    setKey("%s")
    controlSocket("127.0.0.1:%s")
    setServerPolicy(wrandom)
    s1 = newServer{address="127.0.0.1:%s", weight=-1}
    s2 = newServer{address="127.0.0.1:%s", weight=2147483648}
    """

    def testBadWeightWRandom(self):
        """
        Routing: WRandom

        Test that downstreams cannot be added with invalid weights.
        """
        # There should be no downstreams
        self.assertTrue(self.sendConsoleCommand("getServer(0)").startswith("Error"))
Commit	Line	Data
903853f4	1	#!/usr/bin/env python
278403d3	2	import base64
903853f4 RG	3	import threading
	4	import time
	5	import dns
	6	from dnsdisttests import DNSDistTest
	7
	8	class TestRoutingPoolRouting(DNSDistTest):
	9
	10	_config_template = """
	11	newServer{address="127.0.0.1:%s", pool="real"}
903853f4	12	addAction(makeRule("poolaction.routing.tests.powerdns.com"), PoolAction("real"))
903853f4 RG	13	"""
903853f4 RG	14
903853f4 RG	15	def testPolicyPoolAction(self):
	16	"""
	17	Routing: Set pool by qname via PoolAction
	18
	19	Send an A query to "poolaction.routing.tests.powerdns.com.",
	20	check that dnsdist routes the query to the "real" pool.
	21	"""
6bb38cd6	22	name = 'poolaction.routing.tests.powerdns.com.'
903853f4 RG	23	query = dns.message.make_query(name, 'A', 'IN')
	24	response = dns.message.make_response(query)
	25	rrset = dns.rrset.from_text(name,
	26	60,
	27	dns.rdataclass.IN,
	28	dns.rdatatype.A,
	29	'192.0.2.1')
	30	response.answer.append(rrset)
	31
6ca2e796 RG	32	for method in ("sendUDPQuery", "sendTCPQuery"):
	33	sender = getattr(self, method)
	34	(receivedQuery, receivedResponse) = sender(query, response)
	35	receivedQuery.id = query.id
	36	self.assertEquals(query, receivedQuery)
	37	self.assertEquals(response, receivedResponse)
903853f4 RG	38
	39	def testDefaultPool(self):
	40	"""
	41	Routing: Set pool by qname canary
	42
	43	Send an A query to "notpool.routing.tests.powerdns.com.",
	44	check that dnsdist sends no response (no servers
	45	in the default pool).
	46	"""
	47	name = 'notpool.routing.tests.powerdns.com.'
	48	query = dns.message.make_query(name, 'A', 'IN')
	49
6ca2e796 RG	50	for method in ("sendUDPQuery", "sendTCPQuery"):
	51	sender = getattr(self, method)
	52	(_, receivedResponse) = sender(query, response=None, useQueue=False)
	53	self.assertEquals(receivedResponse, None)
903853f4 RG	54
	55	class TestRoutingQPSPoolRouting(DNSDistTest):
	56	_config_template = """
	57	newServer{address="127.0.0.1:%s", pool="regular"}
903853f4 RG	58	addAction(makeRule("qpspoolaction.routing.tests.powerdns.com"), QPSPoolAction(10, "regular"))
	59	"""
	60
903853f4 RG	61	def testQPSPoolAction(self):
	62	"""
	63	Routing: Set pool by QPS via action
	64
	65	Send queries to "qpspoolaction.routing.tests.powerdns.com."
	66	check that dnsdist does not route the query to the "regular" pool
	67	when the max QPS has been reached.
	68	"""
	69	maxQPS = 10
	70	name = 'qpspoolaction.routing.tests.powerdns.com.'
	71	query = dns.message.make_query(name, 'A', 'IN')
	72	response = dns.message.make_response(query)
	73	rrset = dns.rrset.from_text(name,
	74	60,
	75	dns.rdataclass.IN,
	76	dns.rdatatype.A,
	77	'192.0.2.1')
	78	response.answer.append(rrset)
	79
	80	for _ in range(maxQPS):
	81	(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
	82	receivedQuery.id = query.id
	83	self.assertEquals(query, receivedQuery)
	84	self.assertEquals(response, receivedResponse)
	85
	86	# we should now be sent to the "abuse" pool which is empty,
	87	# so the queries should be dropped
	88	(_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
	89	self.assertEquals(receivedResponse, None)
	90
	91	time.sleep(1)
	92
	93	# again, over TCP this time
	94	for _ in range(maxQPS):
	95	(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
	96	receivedQuery.id = query.id
	97	self.assertEquals(query, receivedQuery)
	98	self.assertEquals(response, receivedResponse)
	99
	100
	101	(_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False)
	102	self.assertEquals(receivedResponse, None)
	103
	104
	105	class TestRoutingRoundRobinLB(DNSDistTest):
	106
	107	_testServer2Port = 5351
	108	_config_params = ['_testServerPort', '_testServer2Port']
	109	_config_template = """
	110	setServerPolicy(roundrobin)
	111	s1 = newServer{address="127.0.0.1:%s"}
	112	s1:setUp()
	113	s2 = newServer{address="127.0.0.1:%s"}
	114	s2:setUp()
	115	"""
	116
	117	@classmethod
	118	def startResponders(cls):
	119	print("Launching responders..")
5df86a8a	120	cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
903853f4 RG	121	cls._UDPResponder.setDaemon(True)
903853f4 RG	122	cls._UDPResponder.start()
5df86a8a	123	cls._UDPResponder2 = threading.Thread(name='UDP Responder 2', target=cls.UDPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
903853f4 RG	124	cls._UDPResponder2.setDaemon(True)
	125	cls._UDPResponder2.start()
	126
5df86a8a	127	cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
903853f4 RG	128	cls._TCPResponder.setDaemon(True)
	129	cls._TCPResponder.start()
	130
5df86a8a	131	cls._TCPResponder2 = threading.Thread(name='TCP Responder 2', target=cls.TCPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
903853f4 RG	132	cls._TCPResponder2.setDaemon(True)
	133	cls._TCPResponder2.start()
	134
	135	def testRR(self):
	136	"""
	137	Routing: Round Robin
	138
278403d3	139	Send 10 A queries to "rr.routing.tests.powerdns.com.",
903853f4 RG	140	check that dnsdist routes half of it to each backend.
	141	"""
	142	numberOfQueries = 10
	143	name = 'rr.routing.tests.powerdns.com.'
	144	query = dns.message.make_query(name, 'A', 'IN')
	145	response = dns.message.make_response(query)
	146	rrset = dns.rrset.from_text(name,
	147	60,
	148	dns.rdataclass.IN,
	149	dns.rdatatype.A,
	150	'192.0.2.1')
	151	response.answer.append(rrset)
	152
	153	# the round robin counter is shared for UDP and TCP,
	154	# so we need to do UDP then TCP to have a clean count
	155	for _ in range(numberOfQueries):
	156	(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
	157	receivedQuery.id = query.id
	158	self.assertEquals(query, receivedQuery)
	159	self.assertEquals(response, receivedResponse)
	160
	161	for _ in range(numberOfQueries):
	162	(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
	163	receivedQuery.id = query.id
	164	self.assertEquals(query, receivedQuery)
	165	self.assertEquals(response, receivedResponse)
	166
02bbf9eb RG	167	for key in self._responsesCounter:
02bbf9eb RG	168	value = self._responsesCounter[key]
903853f4 RG	169	self.assertEquals(value, numberOfQueries / 2)
	170
	171	class TestRoutingRoundRobinLBOneDown(DNSDistTest):
	172
	173	_testServer2Port = 5351
	174	_config_params = ['_testServerPort', '_testServer2Port']
	175	_config_template = """
	176	setServerPolicy(roundrobin)
	177	s1 = newServer{address="127.0.0.1:%s"}
	178	s1:setUp()
	179	s2 = newServer{address="127.0.0.1:%s"}
	180	s2:setDown()
	181	"""
	182
	183	def testRRWithOneDown(self):
	184	"""
	185	Routing: Round Robin with one server down
	186
	187	Send 100 A queries to "rr.routing.tests.powerdns.com.",
	188	check that dnsdist routes all of it to the only backend up.
	189	"""
	190	numberOfQueries = 10
	191	name = 'rr.routing.tests.powerdns.com.'
	192	query = dns.message.make_query(name, 'A', 'IN')
	193	response = dns.message.make_response(query)
	194	rrset = dns.rrset.from_text(name,
	195	60,
	196	dns.rdataclass.IN,
	197	dns.rdatatype.A,
	198	'192.0.2.1')
	199	response.answer.append(rrset)
	200
	201	# the round robin counter is shared for UDP and TCP,
	202	# so we need to do UDP then TCP to have a clean count
	203	for _ in range(numberOfQueries):
	204	(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
	205	receivedQuery.id = query.id
	206	self.assertEquals(query, receivedQuery)
	207	self.assertEquals(response, receivedResponse)
	208
	209	for _ in range(numberOfQueries):
	210	(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
	211	receivedQuery.id = query.id
	212	self.assertEquals(query, receivedQuery)
	213	self.assertEquals(response, receivedResponse)
	214
	215	total = 0
02bbf9eb RG	216	for key in self._responsesCounter:
02bbf9eb RG	217	value = self._responsesCounter[key]
903853f4 RG	218	self.assertTrue(value == numberOfQueries or value == 0)
	219	total += value
	220
	221	self.assertEquals(total, numberOfQueries * 2)
d12cd8e9 RG	222
	223	class TestRoutingOrder(DNSDistTest):
	224
	225	_testServer2Port = 5351
	226	_config_params = ['_testServerPort', '_testServer2Port']
	227	_config_template = """
	228	setServerPolicy(firstAvailable)
	229	s1 = newServer{address="127.0.0.1:%s", order=2}
	230	s1:setUp()
	231	s2 = newServer{address="127.0.0.1:%s", order=1}
	232	s2:setUp()
	233	"""
	234
	235	@classmethod
	236	def startResponders(cls):
	237	print("Launching responders..")
5df86a8a	238	cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
d12cd8e9 RG	239	cls._UDPResponder.setDaemon(True)
d12cd8e9 RG	240	cls._UDPResponder.start()
5df86a8a	241	cls._UDPResponder2 = threading.Thread(name='UDP Responder 2', target=cls.UDPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
d12cd8e9 RG	242	cls._UDPResponder2.setDaemon(True)
	243	cls._UDPResponder2.start()
	244
5df86a8a	245	cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
d12cd8e9 RG	246	cls._TCPResponder.setDaemon(True)
	247	cls._TCPResponder.start()
	248
5df86a8a	249	cls._TCPResponder2 = threading.Thread(name='TCP Responder 2', target=cls.TCPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
d12cd8e9 RG	250	cls._TCPResponder2.setDaemon(True)
	251	cls._TCPResponder2.start()
	252
	253	def testOrder(self):
	254	"""
	255	Routing: firstAvailable policy based on 'order'
	256
	257	Send 50 A queries to "order.routing.tests.powerdns.com.",
	258	check that dnsdist routes all of it to the second backend
	259	because it has the lower order value.
	260	"""
	261	numberOfQueries = 50
	262	name = 'order.routing.tests.powerdns.com.'
	263	query = dns.message.make_query(name, 'A', 'IN')
	264	response = dns.message.make_response(query)
	265	rrset = dns.rrset.from_text(name,
	266	60,
	267	dns.rdataclass.IN,
	268	dns.rdatatype.A,
	269	'192.0.2.1')
	270	response.answer.append(rrset)
	271
	272	for _ in range(numberOfQueries):
6ca2e796 RG	273	for method in ("sendUDPQuery", "sendTCPQuery"):
	274	sender = getattr(self, method)
	275	(receivedQuery, receivedResponse) = sender(query, response)
	276	receivedQuery.id = query.id
	277	self.assertEquals(query, receivedQuery)
	278	self.assertEquals(response, receivedResponse)
d12cd8e9 RG	279
d12cd8e9 RG	280	total = 0
26a3cdb7 RG	281	if 'UDP Responder' in self._responsesCounter:
26a3cdb7 RG	282	self.assertEquals(self._responsesCounter['UDP Responder'], 0)
d12cd8e9	283	self.assertEquals(self._responsesCounter['UDP Responder 2'], numberOfQueries)
26a3cdb7 RG	284	if 'TCP Responder' in self._responsesCounter:
26a3cdb7 RG	285	self.assertEquals(self._responsesCounter['TCP Responder'], 0)
d12cd8e9	286	self.assertEquals(self._responsesCounter['TCP Responder 2'], numberOfQueries)
26a3cdb7 RG	287
	288	class TestRoutingNoServer(DNSDistTest):
	289
	290	_config_template = """
	291	newServer{address="127.0.0.1:%s", pool="real"}
	292	setServFailWhenNoServer(true)
	293	"""
	294
	295	def testPolicyPoolNoServer(self):
	296	"""
	297	Routing: No server should return ServFail
	298	"""
	299	name = 'noserver.routing.tests.powerdns.com.'
	300	query = dns.message.make_query(name, 'A', 'IN')
	301	expectedResponse = dns.message.make_response(query)
	302	expectedResponse.set_rcode(dns.rcode.SERVFAIL)
	303
6ca2e796 RG	304	for method in ("sendUDPQuery", "sendTCPQuery"):
	305	sender = getattr(self, method)
	306	(_, receivedResponse) = sender(query, response=None, useQueue=False)
	307	self.assertEquals(receivedResponse, expectedResponse)
278403d3 DM	308
	309	class TestRoutingWRandom(DNSDistTest):
	310
	311	_testServer2Port = 5351
	312	_config_params = ['_testServerPort', '_testServer2Port']
	313	_config_template = """
	314	setServerPolicy(wrandom)
	315	s1 = newServer{address="127.0.0.1:%s", weight=1}
	316	s1:setUp()
	317	s2 = newServer{address="127.0.0.1:%s", weight=2}
	318	s2:setUp()
	319	"""
	320
	321	@classmethod
	322	def startResponders(cls):
	323	print("Launching responders..")
	324	cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
	325	cls._UDPResponder.setDaemon(True)
	326	cls._UDPResponder.start()
	327	cls._UDPResponder2 = threading.Thread(name='UDP Responder 2', target=cls.UDPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
	328	cls._UDPResponder2.setDaemon(True)
	329	cls._UDPResponder2.start()
	330
	331	cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
	332	cls._TCPResponder.setDaemon(True)
	333	cls._TCPResponder.start()
	334
	335	cls._TCPResponder2 = threading.Thread(name='TCP Responder 2', target=cls.TCPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
	336	cls._TCPResponder2.setDaemon(True)
	337	cls._TCPResponder2.start()
	338
	339	def testWRandom(self):
	340	"""
	341	Routing: WRandom
	342
6f9839e9	343	Send 100 A queries to "wrandom.routing.tests.powerdns.com.",
278403d3 DM	344	check that dnsdist routes less than half to one, more to the other.
	345	"""
	346	numberOfQueries = 100
6f9839e9	347	name = 'wrandom.routing.tests.powerdns.com.'
278403d3 DM	348	query = dns.message.make_query(name, 'A', 'IN')
	349	response = dns.message.make_response(query)
	350	rrset = dns.rrset.from_text(name,
	351	60,
	352	dns.rdataclass.IN,
	353	dns.rdatatype.A,
	354	'192.0.2.1')
	355	response.answer.append(rrset)
	356
	357	# the counter is shared for UDP and TCP,
	358	# so we need to do UDP then TCP to have a clean count
	359	for _ in range(numberOfQueries):
	360	(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
	361	receivedQuery.id = query.id
	362	self.assertEquals(query, receivedQuery)
	363	self.assertEquals(response, receivedResponse)
	364
	365	for _ in range(numberOfQueries):
	366	(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
	367	receivedQuery.id = query.id
	368	self.assertEquals(query, receivedQuery)
	369	self.assertEquals(response, receivedResponse)
	370
	371	# The lower weight downstream should receive less than half the queries
6f9839e9 RG	372	self.assertLess(self._responsesCounter['UDP Responder'], numberOfQueries * 0.50)
6f9839e9 RG	373	self.assertLess(self._responsesCounter['TCP Responder'], numberOfQueries * 0.50)
278403d3 DM	374
278403d3 DM	375	# The higher weight downstream should receive more than half the queries
6f9839e9 RG	376	self.assertGreater(self._responsesCounter['UDP Responder 2'], numberOfQueries * 0.50)
6f9839e9 RG	377	self.assertGreater(self._responsesCounter['TCP Responder 2'], numberOfQueries * 0.50)
278403d3 DM	378
	379
	380	class TestRoutingHighValueWRandom(DNSDistTest):
	381
	382	_testServer2Port = 5351
	383	_consoleKey = DNSDistTest.generateConsoleKey()
	384	_consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
	385	_config_params = ['_consoleKeyB64', '_consolePort', '_testServerPort', '_testServer2Port']
	386	_config_template = """
	387	setKey("%s")
	388	controlSocket("127.0.0.1:%s")
	389	setServerPolicy(wrandom)
	390	s1 = newServer{address="127.0.0.1:%s", weight=2000000000}
	391	s1:setUp()
	392	s2 = newServer{address="127.0.0.1:%s", weight=2000000000}
	393	s2:setUp()
	394	"""
	395
	396	@classmethod
	397	def startResponders(cls):
	398	print("Launching responders..")
	399	cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
	400	cls._UDPResponder.setDaemon(True)
	401	cls._UDPResponder.start()
	402	cls._UDPResponder2 = threading.Thread(name='UDP Responder 2', target=cls.UDPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
	403	cls._UDPResponder2.setDaemon(True)
	404	cls._UDPResponder2.start()
	405
	406	cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
	407	cls._TCPResponder.setDaemon(True)
	408	cls._TCPResponder.start()
	409
	410	cls._TCPResponder2 = threading.Thread(name='TCP Responder 2', target=cls.TCPResponder, args=[cls._testServer2Port, cls._toResponderQueue, cls._fromResponderQueue])
	411	cls._TCPResponder2.setDaemon(True)
	412	cls._TCPResponder2.start()
	413
	414	def testHighValueWRandom(self):
	415	"""
6f9839e9	416	Routing: WRandom (overflow)
278403d3	417
6f9839e9	418	Send 100 A queries to "wrandom-overflow.routing.tests.powerdns.com.",
278403d3 DM	419	check that dnsdist routes to each downstream, rather than failing with
	420	no-policy.
	421	"""
	422	numberOfQueries = 100
6f9839e9	423	name = 'wrandom-overflow.routing.tests.powerdns.com.'
278403d3 DM	424	query = dns.message.make_query(name, 'A', 'IN')
	425	response = dns.message.make_response(query)
	426	rrset = dns.rrset.from_text(name,
	427	60,
	428	dns.rdataclass.IN,
	429	dns.rdatatype.A,
	430	'192.0.2.1')
	431	response.answer.append(rrset)
	432
	433	# the counter is shared for UDP and TCP,
	434	# so we need to do UDP then TCP to have a clean count
	435	for _ in range(numberOfQueries):
	436	(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
	437	receivedQuery.id = query.id
	438	self.assertEquals(query, receivedQuery)
	439	self.assertEquals(response, receivedResponse)
	440
	441	for _ in range(numberOfQueries):
	442	(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
	443	receivedQuery.id = query.id
	444	self.assertEquals(query, receivedQuery)
	445	self.assertEquals(response, receivedResponse)
	446
	447	stats = self.sendConsoleCommand("dumpStats()").split()
	448	stats_dict = {}
	449
	450	# Map to a dict with every other element being the value to the previous one
	451	for i, x in enumerate(stats):
	452	if not i % 2:
	453	stats_dict[x] = stats[i+1]
	454
	455	# There should be no queries getting "no-policy" responses
	456	self.assertEquals(stats_dict['no-policy'], '0')
	457
	458	# Each downstream should receive some queries, but it will be unbalanced
6f9839e9	459	# because the sum of the weights is higher than INT_MAX.
278403d3	460	# The first downstream will receive more than half the queries
6f9839e9 RG	461	self.assertGreater(self._responsesCounter['UDP Responder'], numberOfQueries / 2)
	462	self.assertGreater(self._responsesCounter['TCP Responder'], numberOfQueries / 2)
	463
	464	# The second downstream will receive the remainder of the queries, but it might very well be 0
	465	if 'UDP Responder 2' in self._responsesCounter:
	466	self.assertEquals(self._responsesCounter['UDP Responder 2'], numberOfQueries - self._responsesCounter['UDP Responder'])
	467	if 'TCP Responder 2' in self._responsesCounter:
	468	self.assertEquals(self._responsesCounter['TCP Responder 2'], numberOfQueries - self._responsesCounter['TCP Responder'])
278403d3 DM	469
	470	class TestRoutingBadWeightWRandom(DNSDistTest):
	471
	472	_testServer2Port = 5351
	473	_consoleKey = DNSDistTest.generateConsoleKey()
	474	_consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
	475	_config_params = ['_consoleKeyB64', '_consolePort', '_testServerPort', '_testServer2Port']
	476	_config_template = """
	477	setKey("%s")
	478	controlSocket("127.0.0.1:%s")
	479	setServerPolicy(wrandom)
	480	s1 = newServer{address="127.0.0.1:%s", weight=-1}
	481	s2 = newServer{address="127.0.0.1:%s", weight=2147483648}
	482	"""
	483
	484	def testBadWeightWRandom(self):
	485	"""
	486	Routing: WRandom
	487
	488	Test that downstreams cannot be added with invalid weights.
	489	"""
	490	# There should be no downstreams
	491	self.assertTrue(self.sendConsoleCommand("getServer(0)").startswith("Error"))