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

#!/usr/bin/env python

import requests
from dnsdisttests import DNSDistTest

class TestBasics(DNSDistTest):

    _webTimeout = 2.0
    _webServerPort = 8083
    _webServerBasicAuthPassword = 'secret'
    _webServerAPIKey = 'apisecret'
    # paths accessible using the API key
    _apiPaths = ['/api/v1/servers/localhost', '/api/v1/servers/localhost/config', '/api/v1/servers/localhost/statistics', '/jsonstat?command=stats', '/jsonstat?command=dynblocklist']
    # paths accessible using basic auth only (list not exhaustive)
    _basicOnlyPaths = ['/', '/index.html']
    _config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey']
    _config_template = """
    newServer{address="127.0.0.1:%s"}
    webserver("127.0.0.1:%s", "%s", "%s")
    """

    def testBasicAuth(self):
        """
        API: Basic Authentication
        """
        for path in self._basicOnlyPaths + self._apiPaths:
            url = 'http://127.0.0.1:' + str(self._webServerPort) + path
            r = requests.get(url, auth=('whatever', self._webServerBasicAuthPassword), timeout=self._webTimeout)
            self.assertTrue(r)
            self.assertEquals(r.status_code, 200)

    def testXAPIKey(self):
        """
        API: X-Api-Key
        """
        headers = {'x-api-key': self._webServerAPIKey}
        for path in self._apiPaths:
            url = 'http://127.0.0.1:' + str(self._webServerPort) + path
            r = requests.get(url, headers=headers, timeout=self._webTimeout)
            self.assertTrue(r)
            self.assertEquals(r.status_code, 200)

    def testBasicAuthOnly(self):
        """
        API: Basic Authentication Only
        """
        headers = {'x-api-key': self._webServerAPIKey}
        for path in self._basicOnlyPaths:
            url = 'http://127.0.0.1:' + str(self._webServerPort) + path
            r = requests.get(url, headers=headers, timeout=self._webTimeout)
            self.assertEquals(r.status_code, 401)

    def testServersLocalhost(self):
        """
        API: /api/v1/servers/localhost
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()

        self.assertEquals(content['daemon_type'], 'dnsdist')

        for key in ['version', 'acl', 'local', 'rules', 'response-rules', 'servers', 'frontends']:
            self.assertIn(key, content)

        for rule in content['rules']:
            for key in ['id', 'matches', 'rule', 'action']:
                self.assertIn(key, rule)
            for key in ['id', 'matches']:
                self.assertTrue(rule[key] >= 0)

        for rule in content['response-rules']:
            for key in ['id', 'matches', 'rule', 'action']:
                self.assertIn(key, rule)
            for key in ['id', 'matches']:
                self.assertTrue(rule[key] >= 0)

        for server in content['servers']:
            for key in ['id', 'latency', 'name', 'weight', 'outstanding', 'qpsLimit',
                        'reuseds', 'state', 'address', 'pools', 'qps', 'queries', 'order']:
                self.assertIn(key, server)

            for key in ['id', 'latency', 'weight', 'outstanding', 'qpsLimit', 'reuseds',
                        'qps', 'queries', 'order']:
                self.assertTrue(server[key] >= 0)

            self.assertTrue(server['state'] in ['up', 'down', 'UP', 'DOWN'])

        for frontend in content['frontends']:
            for key in ['id', 'address', 'udp', 'tcp', 'queries']:
                self.assertIn(key, frontend)

            for key in ['id', 'queries']:
                self.assertTrue(frontend[key] >= 0)

    def testServersLocalhostConfig(self):
        """
        API: /api/v1/servers/localhost/config
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()
        values = {}
        for entry in content:
            for key in ['type', 'name', 'value']:
                self.assertIn(key, entry)

            self.assertEquals(entry['type'], 'ConfigSetting')
            values[entry['name']] = entry['value']

        for key in ['acl', 'control-socket', 'ecs-override', 'ecs-source-prefix-v4',
                    'ecs-source-prefix-v6', 'fixup-case', 'max-outstanding', 'server-policy',
                    'stale-cache-entries-ttl', 'tcp-recv-timeout', 'tcp-send-timeout',
                    'truncate-tc', 'verbose', 'verbose-health-checks']:
            self.assertIn(key, values)

        for key in ['max-outstanding', 'stale-cache-entries-ttl', 'tcp-recv-timeout',
                    'tcp-send-timeout']:
            self.assertTrue(values[key] >= 0)

        self.assertTrue(values['ecs-source-prefix-v4'] >= 0 and values['ecs-source-prefix-v4'] <= 32)
        self.assertTrue(values['ecs-source-prefix-v6'] >= 0 and values['ecs-source-prefix-v6'] <= 128)

    def testServersLocalhostStatistics(self):
        """
        API: /api/v1/servers/localhost/statistics
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/statistics'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()
        values = {}
        for entry in content:
            self.assertIn('type', entry)
            self.assertIn('name', entry)
            self.assertIn('value', entry)
            self.assertEquals(entry['type'], 'StatisticItem')
            values[entry['name']] = entry['value']

        expected = ['responses', 'servfail-responses', 'queries', 'acl-drops', 'block-filter',
                    'rule-drop', 'rule-nxdomain', 'self-answered', 'downstream-timeouts',
                    'downstream-send-errors', 'trunc-failures', 'no-policy', 'latency0-1',
                    'latency1-10', 'latency10-50', 'latency50-100', 'latency100-1000',
                    'latency-slow', 'latency-avg100', 'latency-avg1000', 'latency-avg10000',
                    'latency-avg1000000', 'uptime', 'real-memory-usage', 'noncompliant-queries',
                    'noncompliant-responses', 'rdqueries', 'empty-queries', 'cache-hits',
                    'cache-misses', 'cpu-user-msec', 'cpu-sys-msec', 'fd-usage', 'dyn-blocked',
                    'dyn-block-nmg-size']

        for key in expected:
            self.assertIn(key, values)
            self.assertTrue(values[key] >= 0)

    def testJsonstatStats(self):
        """
        API: /jsonstat?command=stats
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/jsonstat?command=stats'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()

        for key in ['packetcache-hits', 'packetcache-misses', 'over-capacity-drops', 'too-old-drops']:
            self.assertIn(key, content)
            self.assertTrue(content[key] >= 0)

        expected = ['responses', 'servfail-responses', 'queries', 'acl-drops', 'block-filter',
                    'rule-drop', 'rule-nxdomain', 'self-answered', 'downstream-timeouts',
                    'downstream-send-errors', 'trunc-failures', 'no-policy', 'latency0-1',
                    'latency1-10', 'latency10-50', 'latency50-100', 'latency100-1000',
                    'latency-slow', 'latency-avg100', 'latency-avg1000', 'latency-avg10000',
                    'latency-avg1000000', 'uptime', 'real-memory-usage', 'noncompliant-queries',
                    'noncompliant-responses', 'rdqueries', 'empty-queries', 'cache-hits',
                    'cache-misses', 'cpu-user-msec', 'cpu-sys-msec', 'fd-usage', 'dyn-blocked',
                    'dyn-block-nmg-size']

        for key in expected:
            self.assertIn(key, content)
            self.assertTrue(content[key] >= 0)

    def testJsonstatDynblocklist(self):
        """
        API: /jsonstat?command=dynblocklist
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/jsonstat?command=dynblocklist'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)

        content = r.json()

        if content:
            for key in ['reason', 'seconds', 'blocks']:
                self.assertIn(key, content)

            for key in ['blocks']:
                self.assertTrue(content[key] >= 0)
Commit	Line	Data
02bbf9eb RG	1	#!/usr/bin/env python
	2
	3	import requests
	4	from dnsdisttests import DNSDistTest
	5
	6	class TestBasics(DNSDistTest):
	7
	8	_webTimeout = 2.0
	9	_webServerPort = 8083
	10	_webServerBasicAuthPassword = 'secret'
	11	_webServerAPIKey = 'apisecret'
	12	# paths accessible using the API key
	13	_apiPaths = ['/api/v1/servers/localhost', '/api/v1/servers/localhost/config', '/api/v1/servers/localhost/statistics', '/jsonstat?command=stats', '/jsonstat?command=dynblocklist']
	14	# paths accessible using basic auth only (list not exhaustive)
	15	_basicOnlyPaths = ['/', '/index.html']
	16	_config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey']
	17	_config_template = """
	18	newServer{address="127.0.0.1:%s"}
	19	webserver("127.0.0.1:%s", "%s", "%s")
	20	"""
	21
	22	def testBasicAuth(self):
	23	"""
	24	API: Basic Authentication
	25	"""
	26	for path in self._basicOnlyPaths + self._apiPaths:
	27	url = 'http://127.0.0.1:' + str(self._webServerPort) + path
	28	r = requests.get(url, auth=('whatever', self._webServerBasicAuthPassword), timeout=self._webTimeout)
	29	self.assertTrue(r)
	30	self.assertEquals(r.status_code, 200)
	31
	32	def testXAPIKey(self):
	33	"""
	34	API: X-Api-Key
	35	"""
	36	headers = {'x-api-key': self._webServerAPIKey}
	37	for path in self._apiPaths:
	38	url = 'http://127.0.0.1:' + str(self._webServerPort) + path
	39	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	40	self.assertTrue(r)
	41	self.assertEquals(r.status_code, 200)
	42
	43	def testBasicAuthOnly(self):
	44	"""
	45	API: Basic Authentication Only
	46	"""
	47	headers = {'x-api-key': self._webServerAPIKey}
	48	for path in self._basicOnlyPaths:
	49	url = 'http://127.0.0.1:' + str(self._webServerPort) + path
	50	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	51	self.assertEquals(r.status_code, 401)
	52
	53	def testServersLocalhost(self):
	54	"""
	55	API: /api/v1/servers/localhost
	56	"""
	57	headers = {'x-api-key': self._webServerAPIKey}
	58	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost'
	59	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	60	self.assertTrue(r)
	61	self.assertEquals(r.status_code, 200)
	62	self.assertTrue(r.json())
	63	content = r.json()
	64
65	self.assertEquals(content['daemon_type'], 'dnsdist')
66
46e8b49e	67	for key in ['version', 'acl', 'local', 'rules', 'response-rules', 'servers', 'frontends']:
02bbf9eb RG	68	self.assertIn(key, content)
	69
	70	for rule in content['rules']:
	71	for key in ['id', 'matches', 'rule', 'action']:
	72	self.assertIn(key, rule)
	73	for key in ['id', 'matches']:
	74	self.assertTrue(rule[key] >= 0)
46e8b49e RG	75
	76	for rule in content['response-rules']:
	77	for key in ['id', 'matches', 'rule', 'action']:
	78	self.assertIn(key, rule)
	79	for key in ['id', 'matches']:
	80	self.assertTrue(rule[key] >= 0)
02bbf9eb RG	81
	82	for server in content['servers']:
	83	for key in ['id', 'latency', 'name', 'weight', 'outstanding', 'qpsLimit',
	84	'reuseds', 'state', 'address', 'pools', 'qps', 'queries', 'order']:
	85	self.assertIn(key, server)
	86
	87	for key in ['id', 'latency', 'weight', 'outstanding', 'qpsLimit', 'reuseds',
	88	'qps', 'queries', 'order']:
	89	self.assertTrue(server[key] >= 0)
	90
	91	self.assertTrue(server['state'] in ['up', 'down', 'UP', 'DOWN'])
	92
	93	for frontend in content['frontends']:
	94	for key in ['id', 'address', 'udp', 'tcp', 'queries']:
	95	self.assertIn(key, frontend)
	96
	97	for key in ['id', 'queries']:
	98	self.assertTrue(frontend[key] >= 0)
	99
	100	def testServersLocalhostConfig(self):
	101	"""
	102	API: /api/v1/servers/localhost/config
	103	"""
	104	headers = {'x-api-key': self._webServerAPIKey}
	105	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config'
	106	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	107	self.assertTrue(r)
	108	self.assertEquals(r.status_code, 200)
	109	self.assertTrue(r.json())
	110	content = r.json()
	111	values = {}
	112	for entry in content:
	113	for key in ['type', 'name', 'value']:
	114	self.assertIn(key, entry)
	115
	116	self.assertEquals(entry['type'], 'ConfigSetting')
	117	values[entry['name']] = entry['value']
	118
	119	for key in ['acl', 'control-socket', 'ecs-override', 'ecs-source-prefix-v4',
	120	'ecs-source-prefix-v6', 'fixup-case', 'max-outstanding', 'server-policy',
	121	'stale-cache-entries-ttl', 'tcp-recv-timeout', 'tcp-send-timeout',
	122	'truncate-tc', 'verbose', 'verbose-health-checks']:
	123	self.assertIn(key, values)
	124
	125	for key in ['max-outstanding', 'stale-cache-entries-ttl', 'tcp-recv-timeout',
	126	'tcp-send-timeout']:
	127	self.assertTrue(values[key] >= 0)
	128
	129	self.assertTrue(values['ecs-source-prefix-v4'] >= 0 and values['ecs-source-prefix-v4'] <= 32)
	130	self.assertTrue(values['ecs-source-prefix-v6'] >= 0 and values['ecs-source-prefix-v6'] <= 128)
	131
	132	def testServersLocalhostStatistics(self):
	133	"""
	134	API: /api/v1/servers/localhost/statistics
	135	"""
	136	headers = {'x-api-key': self._webServerAPIKey}
	137	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/statistics'
	138	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	139	self.assertTrue(r)
	140	self.assertEquals(r.status_code, 200)
	141	self.assertTrue(r.json())
	142	content = r.json()
	143	values = {}
	144	for entry in content:
145	self.assertIn('type', entry)
146	self.assertIn('name', entry)
147	self.assertIn('value', entry)
148	self.assertEquals(entry['type'], 'StatisticItem')
149	values[entry['name']] = entry['value']
150
151	expected = ['responses', 'servfail-responses', 'queries', 'acl-drops', 'block-filter',
152	'rule-drop', 'rule-nxdomain', 'self-answered', 'downstream-timeouts',
153	'downstream-send-errors', 'trunc-failures', 'no-policy', 'latency0-1',
154	'latency1-10', 'latency10-50', 'latency50-100', 'latency100-1000',
155	'latency-slow', 'latency-avg100', 'latency-avg1000', 'latency-avg10000',
156	'latency-avg1000000', 'uptime', 'real-memory-usage', 'noncompliant-queries',
157	'noncompliant-responses', 'rdqueries', 'empty-queries', 'cache-hits',
158	'cache-misses', 'cpu-user-msec', 'cpu-sys-msec', 'fd-usage', 'dyn-blocked',
159	'dyn-block-nmg-size']
160
161	for key in expected:
162	self.assertIn(key, values)
163	self.assertTrue(values[key] >= 0)
164
165	def testJsonstatStats(self):
166	"""
167	API: /jsonstat?command=stats
168	"""
169	headers = {'x-api-key': self._webServerAPIKey}
170	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/jsonstat?command=stats'
171	r = requests.get(url, headers=headers, timeout=self._webTimeout)
172	self.assertTrue(r)
173	self.assertEquals(r.status_code, 200)
174	self.assertTrue(r.json())
175	content = r.json()
176
177	for key in ['packetcache-hits', 'packetcache-misses', 'over-capacity-drops', 'too-old-drops']:
178	self.assertIn(key, content)
179	self.assertTrue(content[key] >= 0)
180
181	expected = ['responses', 'servfail-responses', 'queries', 'acl-drops', 'block-filter',
182	'rule-drop', 'rule-nxdomain', 'self-answered', 'downstream-timeouts',
183	'downstream-send-errors', 'trunc-failures', 'no-policy', 'latency0-1',
184	'latency1-10', 'latency10-50', 'latency50-100', 'latency100-1000',
185	'latency-slow', 'latency-avg100', 'latency-avg1000', 'latency-avg10000',
186	'latency-avg1000000', 'uptime', 'real-memory-usage', 'noncompliant-queries',
187	'noncompliant-responses', 'rdqueries', 'empty-queries', 'cache-hits',
188	'cache-misses', 'cpu-user-msec', 'cpu-sys-msec', 'fd-usage', 'dyn-blocked',
189	'dyn-block-nmg-size']
190
191	for key in expected:
192	self.assertIn(key, content)
193	self.assertTrue(content[key] >= 0)
194
195	def testJsonstatDynblocklist(self):
196	"""
197	API: /jsonstat?command=dynblocklist
198	"""
199	headers = {'x-api-key': self._webServerAPIKey}
200	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/jsonstat?command=dynblocklist'
201	r = requests.get(url, headers=headers, timeout=self._webTimeout)
202	self.assertTrue(r)
203	self.assertEquals(r.status_code, 200)
204
205	content = r.json()
206
207	if content:
208	for key in ['reason', 'seconds', 'blocks']:
209	self.assertIn(key, content)
210
211	for key in ['blocks']:
212	self.assertTrue(content[key] >= 0)