6 from dnsdisttests
import DNSDistTest
8 @unittest.skipIf('SKIP_PROMETHEUS_TESTS' in os
.environ
, 'Prometheus tests are disabled')
9 class TestPrometheus(DNSDistTest
):
13 _webServerBasicAuthPassword
= 'secret'
14 _webServerAPIKey
= 'apisecret'
15 _config_params
= ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey']
16 _config_template
= """
17 newServer{address="127.0.0.1:%s"}
18 webserver("127.0.0.1:%s")
19 setWebserverConfig({password="%s", apiKey="%s"})
22 def checkPrometheusContentBasic(self
, content
):
23 for line
in content
.splitlines():
24 if line
.startswith('# HELP'):
25 tokens
= line
.split(' ')
26 self
.assertGreaterEqual(len(tokens
), 4)
27 elif line
.startswith('# TYPE'):
28 tokens
= line
.split(' ')
29 self
.assertEqual(len(tokens
), 4)
30 self
.assertIn(tokens
[3], ['counter', 'gauge', 'histogram'])
31 elif not line
.startswith('#'):
32 tokens
= line
.split(' ')
33 self
.assertEqual(len(tokens
), 2)
34 if not line
.startswith('dnsdist_'):
35 raise AssertionError('Expecting prometheus metric to be prefixed by \'dnsdist_\', got: "%s"' % (line
))
37 def checkPrometheusContentPromtool(self
, content
):
40 testcmd
= ['promtool', 'check', 'metrics']
41 process
= subprocess
.Popen(testcmd
, stdout
=subprocess
.PIPE
, stdin
=subprocess
.PIPE
, stderr
=subprocess
.STDOUT
, close_fds
=True)
42 output
= process
.communicate(input=content
)
43 except subprocess
.CalledProcessError
as exc
:
44 raise AssertionError('%s failed (%d): %s' % (testcmd
, process
.returncode
, process
.output
))
46 # commented out because promtool returns 3 because of the "_total" suffix warnings
47 #if process.returncode != 0:
48 # raise AssertionError('%s failed (%d): %s' % (testcmd, process.returncode, output))
50 for line
in output
[0].splitlines():
51 if line
.endswith(b
"should have \"_total\" suffix"):
53 raise AssertionError('%s returned an unexpected output. Faulty line is "%s", complete content is "%s"' % (testcmd
, line
, output
))
55 def testMetrics(self
):
57 Prometheus: Retrieve metrics
59 url
= 'http://127.0.0.1:' + str(self
._webServerPort
) + '/metrics'
60 r
= requests
.get(url
, auth
=('whatever', self
._webServerBasicAuthPassword
), timeout
=self
._webTimeout
)
62 self
.assertEqual(r
.status_code
, 200)
63 self
.checkPrometheusContentBasic(r
.text
)
64 self
.checkPrometheusContentPromtool(r
.content
)