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

#!/usr/bin/env python
import os.path

import json
import requests
from dnsdisttests import DNSDistTest

class TestAPIBasics(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/config/allow-from', '/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 = """
    setACL({"127.0.0.1/32", "::1/128"})
    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 testServersLocalhostConfigAllowFrom(self):
        """
        API: /api/v1/servers/localhost/config/allow-from
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config/allow-from'
        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 ['type', 'name', 'value']:
            self.assertIn(key, content)

        self.assertEquals(content['name'], 'allow-from')
        self.assertEquals(content['type'], 'ConfigSetting')
        self.assertEquals(content['value'], ["127.0.0.1/32", "::1/128"])

    def testServersLocalhostConfigAllowFromPut(self):
        """
        API: PUT /api/v1/servers/localhost/config/allow-from (should be refused)

        The API is read-only by default, so this should be refused
        """
        newACL = ["192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"]
        payload = json.dumps({"name": "allow-from",
                              "type": "ConfigSetting",
                              "value": newACL})
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config/allow-from'
        r = requests.put(url, headers=headers, timeout=self._webTimeout, data=payload)
        self.assertFalse(r)
        self.assertEquals(r.status_code, 405)

    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', 'rule-refused', '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)

        for key in values:
            self.assertIn(key, expected)

    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()

        expected = ['responses', 'servfail-responses', 'queries', 'acl-drops', 'block-filter',
                    'rule-drop', 'rule-nxdomain', 'rule-refused', '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', 'packetcache-hits', 'packetcache-misses', 'over-capacity-drops',
                    'too-old-drops']

        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)

class TestAPIServerDown(DNSDistTest):

    _webTimeout = 2.0
    _webServerPort = 8083
    _webServerBasicAuthPassword = 'secret'
    _webServerAPIKey = 'apisecret'
    # paths accessible using the API key
    _config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey']
    _config_template = """
    setACL({"127.0.0.1/32", "::1/128"})
    newServer{address="127.0.0.1:%s"}
    getServer(0):setDown()
    webserver("127.0.0.1:%s", "%s", "%s")
    """

    def testServerDownNoLatencyLocalhost(self):
        """
        API: /api/v1/servers/localhost, no latency for a down server
        """
        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['servers'][0]['latency'], None)

class TestAPIWritable(DNSDistTest):

    _webTimeout = 2.0
    _webServerPort = 8083
    _webServerBasicAuthPassword = 'secret'
    _webServerAPIKey = 'apisecret'
    _APIWriteDir = '/tmp'
    _config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey', '_APIWriteDir']
    _config_template = """
    setACL({"127.0.0.1/32", "::1/128"})
    newServer{address="127.0.0.1:%s"}
    webserver("127.0.0.1:%s", "%s", "%s")
    setAPIWritable(true, "%s")
    """

    def testSetACL(self):
        """
        API: Set ACL
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config/allow-from'
        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['value'], ["127.0.0.1/32", "::1/128"])

        newACL = ["192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"]
        payload = json.dumps({"name": "allow-from",
                              "type": "ConfigSetting",
                              "value": newACL})
        r = requests.put(url, headers=headers, timeout=self._webTimeout, data=payload)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()
        self.assertEquals(content['value'], newACL)

        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['value'], newACL)

        configFile = self._APIWriteDir + '/' + 'acl.conf'
        self.assertTrue(os.path.isfile(configFile))
        fileContent = None
        with file(configFile) as f:
            fileContent = f.read()

        self.assertEquals(fileContent, """-- Generated by the REST API, DO NOT EDIT
setACL({"192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"})
""")
Commit	Line	Data
02bbf9eb	1	#!/usr/bin/env python
56d68fad	2	import os.path
02bbf9eb	3
56d68fad	4	import json
02bbf9eb RG	5	import requests
	6	from dnsdisttests import DNSDistTest
	7
56d68fad	8	class TestAPIBasics(DNSDistTest):
02bbf9eb RG	9
	10	_webTimeout = 2.0
	11	_webServerPort = 8083
	12	_webServerBasicAuthPassword = 'secret'
	13	_webServerAPIKey = 'apisecret'
	14	# paths accessible using the API key
56d68fad	15	_apiPaths = ['/api/v1/servers/localhost', '/api/v1/servers/localhost/config', '/api/v1/servers/localhost/config/allow-from', '/api/v1/servers/localhost/statistics', '/jsonstat?command=stats', '/jsonstat?command=dynblocklist']
02bbf9eb RG	16	# paths accessible using basic auth only (list not exhaustive)
	17	_basicOnlyPaths = ['/', '/index.html']
	18	_config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey']
	19	_config_template = """
56d68fad	20	setACL({"127.0.0.1/32", "::1/128"})
02bbf9eb RG	21	newServer{address="127.0.0.1:%s"}
	22	webserver("127.0.0.1:%s", "%s", "%s")
	23	"""
	24
	25	def testBasicAuth(self):
	26	"""
	27	API: Basic Authentication
	28	"""
	29	for path in self._basicOnlyPaths + self._apiPaths:
	30	url = 'http://127.0.0.1:' + str(self._webServerPort) + path
	31	r = requests.get(url, auth=('whatever', self._webServerBasicAuthPassword), timeout=self._webTimeout)
	32	self.assertTrue(r)
	33	self.assertEquals(r.status_code, 200)
	34
	35	def testXAPIKey(self):
	36	"""
	37	API: X-Api-Key
	38	"""
	39	headers = {'x-api-key': self._webServerAPIKey}
	40	for path in self._apiPaths:
	41	url = 'http://127.0.0.1:' + str(self._webServerPort) + path
	42	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	43	self.assertTrue(r)
	44	self.assertEquals(r.status_code, 200)
	45
	46	def testBasicAuthOnly(self):
	47	"""
	48	API: Basic Authentication Only
	49	"""
	50	headers = {'x-api-key': self._webServerAPIKey}
	51	for path in self._basicOnlyPaths:
	52	url = 'http://127.0.0.1:' + str(self._webServerPort) + path
	53	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	54	self.assertEquals(r.status_code, 401)
	55
	56	def testServersLocalhost(self):
	57	"""
	58	API: /api/v1/servers/localhost
	59	"""
	60	headers = {'x-api-key': self._webServerAPIKey}
	61	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost'
	62	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	63	self.assertTrue(r)
	64	self.assertEquals(r.status_code, 200)
	65	self.assertTrue(r.json())
	66	content = r.json()
	67
	68	self.assertEquals(content['daemon_type'], 'dnsdist')
	69
46e8b49e	70	for key in ['version', 'acl', 'local', 'rules', 'response-rules', 'servers', 'frontends']:
02bbf9eb RG	71	self.assertIn(key, content)
	72
	73	for rule in content['rules']:
	74	for key in ['id', 'matches', 'rule', 'action']:
	75	self.assertIn(key, rule)
	76	for key in ['id', 'matches']:
	77	self.assertTrue(rule[key] >= 0)
46e8b49e RG	78
	79	for rule in content['response-rules']:
	80	for key in ['id', 'matches', 'rule', 'action']:
	81	self.assertIn(key, rule)
	82	for key in ['id', 'matches']:
	83	self.assertTrue(rule[key] >= 0)
02bbf9eb RG	84
	85	for server in content['servers']:
	86	for key in ['id', 'latency', 'name', 'weight', 'outstanding', 'qpsLimit',
	87	'reuseds', 'state', 'address', 'pools', 'qps', 'queries', 'order']:
	88	self.assertIn(key, server)
	89
	90	for key in ['id', 'latency', 'weight', 'outstanding', 'qpsLimit', 'reuseds',
	91	'qps', 'queries', 'order']:
	92	self.assertTrue(server[key] >= 0)
	93
	94	self.assertTrue(server['state'] in ['up', 'down', 'UP', 'DOWN'])
	95
	96	for frontend in content['frontends']:
	97	for key in ['id', 'address', 'udp', 'tcp', 'queries']:
	98	self.assertIn(key, frontend)
	99
	100	for key in ['id', 'queries']:
	101	self.assertTrue(frontend[key] >= 0)
	102
	103	def testServersLocalhostConfig(self):
	104	"""
	105	API: /api/v1/servers/localhost/config
	106	"""
	107	headers = {'x-api-key': self._webServerAPIKey}
	108	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config'
	109	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	110	self.assertTrue(r)
	111	self.assertEquals(r.status_code, 200)
	112	self.assertTrue(r.json())
	113	content = r.json()
	114	values = {}
	115	for entry in content:
	116	for key in ['type', 'name', 'value']:
	117	self.assertIn(key, entry)
	118
	119	self.assertEquals(entry['type'], 'ConfigSetting')
	120	values[entry['name']] = entry['value']
	121
	122	for key in ['acl', 'control-socket', 'ecs-override', 'ecs-source-prefix-v4',
	123	'ecs-source-prefix-v6', 'fixup-case', 'max-outstanding', 'server-policy',
	124	'stale-cache-entries-ttl', 'tcp-recv-timeout', 'tcp-send-timeout',
	125	'truncate-tc', 'verbose', 'verbose-health-checks']:
	126	self.assertIn(key, values)
	127
	128	for key in ['max-outstanding', 'stale-cache-entries-ttl', 'tcp-recv-timeout',
	129	'tcp-send-timeout']:
	130	self.assertTrue(values[key] >= 0)
	131
	132	self.assertTrue(values['ecs-source-prefix-v4'] >= 0 and values['ecs-source-prefix-v4'] <= 32)
	133	self.assertTrue(values['ecs-source-prefix-v6'] >= 0 and values['ecs-source-prefix-v6'] <= 128)
	134
56d68fad RG	135	def testServersLocalhostConfigAllowFrom(self):
	136	"""
	137	API: /api/v1/servers/localhost/config/allow-from
	138	"""
	139	headers = {'x-api-key': self._webServerAPIKey}
	140	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config/allow-from'
	141	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	142	self.assertTrue(r)
	143	self.assertEquals(r.status_code, 200)
	144	self.assertTrue(r.json())
	145	content = r.json()
	146	for key in ['type', 'name', 'value']:
	147	self.assertIn(key, content)
	148
	149	self.assertEquals(content['name'], 'allow-from')
	150	self.assertEquals(content['type'], 'ConfigSetting')
	151	self.assertEquals(content['value'], ["127.0.0.1/32", "::1/128"])
	152
	153	def testServersLocalhostConfigAllowFromPut(self):
	154	"""
	155	API: PUT /api/v1/servers/localhost/config/allow-from (should be refused)
	156
	157	The API is read-only by default, so this should be refused
	158	"""
	159	newACL = ["192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"]
	160	payload = json.dumps({"name": "allow-from",
	161	"type": "ConfigSetting",
	162	"value": newACL})
	163	headers = {'x-api-key': self._webServerAPIKey}
	164	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config/allow-from'
	165	r = requests.put(url, headers=headers, timeout=self._webTimeout, data=payload)
	166	self.assertFalse(r)
	167	self.assertEquals(r.status_code, 405)
	168
02bbf9eb RG	169	def testServersLocalhostStatistics(self):
	170	"""
	171	API: /api/v1/servers/localhost/statistics
	172	"""
	173	headers = {'x-api-key': self._webServerAPIKey}
	174	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/statistics'
	175	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	176	self.assertTrue(r)
	177	self.assertEquals(r.status_code, 200)
	178	self.assertTrue(r.json())
	179	content = r.json()
	180	values = {}
	181	for entry in content:
	182	self.assertIn('type', entry)
	183	self.assertIn('name', entry)
	184	self.assertIn('value', entry)
	185	self.assertEquals(entry['type'], 'StatisticItem')
	186	values[entry['name']] = entry['value']
	187
	188	expected = ['responses', 'servfail-responses', 'queries', 'acl-drops', 'block-filter',
dd46e5e3	189	'rule-drop', 'rule-nxdomain', 'rule-refused', 'self-answered', 'downstream-timeouts',
02bbf9eb RG	190	'downstream-send-errors', 'trunc-failures', 'no-policy', 'latency0-1',
	191	'latency1-10', 'latency10-50', 'latency50-100', 'latency100-1000',
	192	'latency-slow', 'latency-avg100', 'latency-avg1000', 'latency-avg10000',
	193	'latency-avg1000000', 'uptime', 'real-memory-usage', 'noncompliant-queries',
	194	'noncompliant-responses', 'rdqueries', 'empty-queries', 'cache-hits',
	195	'cache-misses', 'cpu-user-msec', 'cpu-sys-msec', 'fd-usage', 'dyn-blocked',
	196	'dyn-block-nmg-size']
	197
	198	for key in expected:
	199	self.assertIn(key, values)
	200	self.assertTrue(values[key] >= 0)
	201
dd46e5e3 RG	202	for key in values:
	203	self.assertIn(key, expected)
	204
02bbf9eb RG	205	def testJsonstatStats(self):
	206	"""
	207	API: /jsonstat?command=stats
	208	"""
	209	headers = {'x-api-key': self._webServerAPIKey}
	210	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/jsonstat?command=stats'
	211	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	212	self.assertTrue(r)
	213	self.assertEquals(r.status_code, 200)
	214	self.assertTrue(r.json())
	215	content = r.json()
	216
02bbf9eb	217	expected = ['responses', 'servfail-responses', 'queries', 'acl-drops', 'block-filter',
dd46e5e3	218	'rule-drop', 'rule-nxdomain', 'rule-refused', 'self-answered', 'downstream-timeouts',
02bbf9eb RG	219	'downstream-send-errors', 'trunc-failures', 'no-policy', 'latency0-1',
	220	'latency1-10', 'latency10-50', 'latency50-100', 'latency100-1000',
	221	'latency-slow', 'latency-avg100', 'latency-avg1000', 'latency-avg10000',
	222	'latency-avg1000000', 'uptime', 'real-memory-usage', 'noncompliant-queries',
	223	'noncompliant-responses', 'rdqueries', 'empty-queries', 'cache-hits',
	224	'cache-misses', 'cpu-user-msec', 'cpu-sys-msec', 'fd-usage', 'dyn-blocked',
dd46e5e3 RG	225	'dyn-block-nmg-size', 'packetcache-hits', 'packetcache-misses', 'over-capacity-drops',
dd46e5e3 RG	226	'too-old-drops']
02bbf9eb RG	227
	228	for key in expected:
	229	self.assertIn(key, content)
	230	self.assertTrue(content[key] >= 0)
	231
	232	def testJsonstatDynblocklist(self):
	233	"""
	234	API: /jsonstat?command=dynblocklist
	235	"""
	236	headers = {'x-api-key': self._webServerAPIKey}
	237	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/jsonstat?command=dynblocklist'
	238	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	239	self.assertTrue(r)
	240	self.assertEquals(r.status_code, 200)
	241
	242	content = r.json()
	243
	244	if content:
	245	for key in ['reason', 'seconds', 'blocks']:
	246	self.assertIn(key, content)
	247
	248	for key in ['blocks']:
	249	self.assertTrue(content[key] >= 0)
56d68fad	250
36927800 RG	251	class TestAPIServerDown(DNSDistTest):
	252
	253	_webTimeout = 2.0
	254	_webServerPort = 8083
	255	_webServerBasicAuthPassword = 'secret'
	256	_webServerAPIKey = 'apisecret'
	257	# paths accessible using the API key
	258	_config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey']
	259	_config_template = """
	260	setACL({"127.0.0.1/32", "::1/128"})
	261	newServer{address="127.0.0.1:%s"}
	262	getServer(0):setDown()
	263	webserver("127.0.0.1:%s", "%s", "%s")
	264	"""
	265
	266	def testServerDownNoLatencyLocalhost(self):
	267	"""
	268	API: /api/v1/servers/localhost, no latency for a down server
	269	"""
	270	headers = {'x-api-key': self._webServerAPIKey}
	271	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost'
	272	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	273	self.assertTrue(r)
	274	self.assertEquals(r.status_code, 200)
	275	self.assertTrue(r.json())
	276	content = r.json()
	277
	278	self.assertEquals(content['servers'][0]['latency'], None)
	279
56d68fad RG	280	class TestAPIWritable(DNSDistTest):
	281
	282	_webTimeout = 2.0
	283	_webServerPort = 8083
	284	_webServerBasicAuthPassword = 'secret'
	285	_webServerAPIKey = 'apisecret'
	286	_APIWriteDir = '/tmp'
	287	_config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey', '_APIWriteDir']
	288	_config_template = """
	289	setACL({"127.0.0.1/32", "::1/128"})
	290	newServer{address="127.0.0.1:%s"}
	291	webserver("127.0.0.1:%s", "%s", "%s")
	292	setAPIWritable(true, "%s")
	293	"""
	294
	295	def testSetACL(self):
	296	"""
	297	API: Set ACL
	298	"""
	299	headers = {'x-api-key': self._webServerAPIKey}
	300	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config/allow-from'
	301	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	302	self.assertTrue(r)
	303	self.assertEquals(r.status_code, 200)
	304	self.assertTrue(r.json())
	305	content = r.json()
	306	self.assertEquals(content['value'], ["127.0.0.1/32", "::1/128"])
	307
	308	newACL = ["192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"]
	309	payload = json.dumps({"name": "allow-from",
	310	"type": "ConfigSetting",
	311	"value": newACL})
	312	r = requests.put(url, headers=headers, timeout=self._webTimeout, data=payload)
	313	self.assertTrue(r)
	314	self.assertEquals(r.status_code, 200)
	315	self.assertTrue(r.json())
	316	content = r.json()
	317	self.assertEquals(content['value'], newACL)
	318
	319	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	320	self.assertTrue(r)
	321	self.assertEquals(r.status_code, 200)
	322	self.assertTrue(r.json())
	323	content = r.json()
	324	self.assertEquals(content['value'], newACL)
	325
	326	configFile = self._APIWriteDir + '/' + 'acl.conf'
	327	self.assertTrue(os.path.isfile(configFile))
	328	fileContent = None
	329	with file(configFile) as f:
	330	fileContent = f.read()
	331
	332	self.assertEquals(fileContent, """-- Generated by the REST API, DO NOT EDIT
	333	setACL({"192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"})
	334	""")