regression-tests.recursor-dnssec/test_API.py

   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):
  42         self.waitForTCPSocket("127.0.0.1", self._wsPort)
  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)
  47         self.assertEqual(r.status_code, 200)
  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):
  67         self.waitForTCPSocket("127.0.0.1", self._wsPort)
  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