regression-tests.ixfrdist/test_Stats.py

   1 from ixfrdisttests import IXFRDistTest
   2 import time
   3 import requests
   4
   5 xfrServerPort = 4244
   6
   7 class IXFRDistStatsTest(IXFRDistTest):
   8     """
   9     This test makes sure we have statistics in ixfrdist
  10     """
  11
  12     webserver_address = '127.0.0.1:8080'
  13
  14     _config_params = ['_ixfrDistPort', 'webserver_address']
  15
  16     _config_template = """
  17 listen:
  18   - '127.0.0.1:%d'
  19 acl:
  20   - '127.0.0.0/8'
  21 axfr-timeout: 20
  22 keep: 20
  23 tcp-in-threads: 1
  24 work-dir: 'ixfrdist.dir'
  25 failed-soa-retry: 3
  26 webserver-address: %s
  27 """
  28
  29     _config_domains = {'example': '127.0.0.1:' + str(xfrServerPort)}
  30
  31     metric_prog_stats = ["ixfrdist_uptime_seconds", "ixfrdist_domains"]
  32     metric_domain_stats = ["ixfrdist_soa_serial", "ixfrdist_soa_checks",
  33                            "ixfrdist_soa_checks_failed",
  34                            "ixfrdist_soa_inqueries",
  35                            "ixfrdist_axfr_inqueries", "ixfrdist_axfr_failures",
  36                            "ixfrdist_ixfr_inqueries", "ixfrdist_ixfr_failures"]
  37
  38     @classmethod
  39     def setUpClass(cls):
  40         cls.startIXFRDist()
  41         cls.setUpSockets()
  42         time.sleep(3)
  43
  44     @classmethod
  45     def tearDownClass(cls):
  46         cls.tearDownIXFRDist()
  47
  48     def test_program_stats_exist(self):
  49         res = requests.get('http://{}/metrics'.format(self.webserver_address))
  50         self.assertEqual(res.status_code, 200)
  51         for line in res.text.splitlines():
  52             if line[0] == "#":
  53                 continue
  54             if "{" in line:
  55                 continue
  56             self.assertIn(line.split(" ")[0],
  57                           self.metric_prog_stats + self.metric_domain_stats)
  58
  59     def test_registered(self):
  60         res = requests.get('http://{}/metrics'.format(self.webserver_address))
  61         self.assertEqual(res.status_code, 200)
  62         for line in res.text.splitlines():
  63             if line.startswith('ixfrdist_domains'):
  64                 self.assertEqual(line, 'ixfrdist_domains 1')
  65                 continue
  66             if line[0] == "#":
  67                 continue
  68             if "{" not in line:
  69                 continue
  70             self.assertIn('{domain=example}', line)
  71             self.assertIn(line.split("{")[0], self.metric_domain_stats)
  72
  73     def test_metrics_have_help(self):
  74         res = requests.get('http://{}/metrics'.format(self.webserver_address))
  75         self.assertEqual(res.status_code, 200)
  76         for s in self.metric_prog_stats + self.metric_domain_stats:
  77             self.assertIn('# HELP {}'.format(s), res.text)
  78
  79     def test_metrics_have_type(self):
  80         res = requests.get('http://{}/metrics'.format(self.webserver_address))
  81         self.assertEqual(res.status_code, 200)
  82         for s in self.metric_prog_stats + self.metric_domain_stats:
  83             self.assertIn('# TYPE {}'.format(s), res.text)
  84
  85     def test_missing_metrics(self):
  86         all_metrics = set()
  87
  88         res = requests.get('http://{}/metrics'.format(self.webserver_address))
  89         self.assertEqual(res.status_code, 200)
  90
  91         for line in res.text.splitlines():
  92             if line[0] == "#":
  93                 all_metrics.add(line.split(" ")[2])
  94                 continue
  95             if "{" in line:
  96                 all_metrics.add(line.split("{")[0])
  97                 continue
  98             all_metrics.add(line.split(" ")[0])
  99
 100         should_have_metrics = set(self.metric_prog_stats +
 101                                   self.metric_domain_stats)
 102
 103         unknown_metrics = all_metrics - should_have_metrics
 104
 105         self.assertSetEqual(unknown_metrics, set())