]>
Commit | Line | Data |
---|---|---|
b1b7facf OM |
1 | import os |
2 | import requests | |
3 | ||
4 | from recursortests import RecursorTest | |
5 | ||
6 | class APIRecursorTest(RecursorTest): | |
7 | ||
8 | @classmethod | |
9 | def setUpClass(cls): | |
10 | ||
11 | # we don't need all the auth stuff | |
12 | cls.setUpSockets() | |
13 | cls.startResponders() | |
14 | ||
15 | confdir = os.path.join('configs', cls._confdir) | |
16 | cls.createConfigDir(confdir) | |
17 | ||
18 | cls.generateRecursorConfig(confdir) | |
19 | cls.startRecursor(confdir, cls._recursorPort) | |
20 | ||
21 | @classmethod | |
22 | def tearDownClass(cls): | |
23 | cls.tearDownRecursor() | |
24 | ||
25 | class APIAllowedRecursorTest(APIRecursorTest): | |
26 | _confdir = 'API' | |
27 | _wsPort = 8042 | |
28 | _wsTimeout = 2 | |
29 | _wsPassword = 'secretpassword' | |
30 | _apiKey = 'secretapikey' | |
31 | ||
32 | _config_template = """ | |
33 | webserver=yes | |
34 | webserver-port=%d | |
35 | webserver-address=127.0.0.1 | |
36 | webserver-password=%s | |
37 | webserver-allow-from=127.0.0.1 | |
38 | api-key=%s | |
39 | """ % (_wsPort, _wsPassword, _apiKey) | |
40 | ||
41 | def testAPI(self): | |
32a17637 | 42 | self.waitForTCPSocket("127.0.0.1", self._wsPort) |
b1b7facf OM |
43 | headers = {'x-api-key': self._apiKey} |
44 | url = 'http://127.0.0.1:' + str(self._wsPort) + '/api/v1/servers/localhost/statistics' | |
45 | r = requests.get(url, headers=headers, timeout=self._wsTimeout) | |
46 | self.assertTrue(r) | |
4bfebc93 | 47 | self.assertEqual(r.status_code, 200) |
b1b7facf OM |
48 | self.assertTrue(r.json()) |
49 | ||
50 | class APIDeniedRecursorTest(APIRecursorTest): | |
51 | _confdir = 'API' | |
52 | _wsPort = 8042 | |
53 | _wsTimeout = 2 | |
54 | _wsPassword = 'secretpassword' | |
55 | _apiKey = 'secretapikey' | |
56 | ||
57 | _config_template = """ | |
58 | webserver=yes | |
59 | webserver-port=%d | |
60 | webserver-address=127.0.0.1 | |
61 | webserver-password=%s | |
62 | webserver-allow-from=192.0.2.1 | |
63 | api-key=%s | |
64 | """ % (_wsPort, _wsPassword, _apiKey) | |
65 | ||
66 | def testAPI(self): | |
32a17637 | 67 | self.waitForTCPSocket("127.0.0.1", self._wsPort) |
b1b7facf OM |
68 | headers = {'x-api-key': self._apiKey} |
69 | url = 'http://127.0.0.1:' + str(self._wsPort) + '/api/v1/servers/localhost/statistics' | |
70 | try: | |
71 | r = requests.get(url, headers=headers, timeout=self._wsTimeout) | |
72 | self.assertTrue(False) | |
73 | except requests.exceptions.ConnectionError as exp: | |
74 | pass |