]>
git.ipfire.org Git - thirdparty/pdns.git/blob - regression-tests.recursor-dnssec/test_Prometheus.py
6 from recursortests
import RecursorTest
8 class RecPrometheusTest(RecursorTest
):
13 # we don't need all the auth stuff
17 confdir
= os
.path
.join('configs', cls
._confdir
)
18 cls
.createConfigDir(confdir
)
20 cls
.generateRecursorConfig(confdir
)
21 cls
.startRecursor(confdir
, cls
._recursorPort
)
24 def tearDownClass(cls
):
25 cls
.tearDownRecursor()
27 def checkPrometheusContentBasic(self
, content
):
28 for line
in content
.splitlines():
29 if line
.startswith('# HELP'):
30 tokens
= line
.split(' ')
31 self
.assertGreaterEqual(len(tokens
), 4)
32 elif line
.startswith('# TYPE'):
33 tokens
= line
.split(' ')
34 self
.assertEqual(len(tokens
), 4)
35 self
.assertIn(tokens
[3], ['counter', 'gauge', 'histogram'])
36 elif not line
.startswith('#'):
37 tokens
= line
.split(' ')
38 self
.assertEqual(len(tokens
), 2)
39 if not line
.startswith('pdns_recursor_'):
40 raise AssertionError('Expecting prometheus metric to be prefixed by \'pdns_recursor_\', got: "%s"' % (line
))
42 def checkPrometheusContentPromtool(self
, content
):
45 testcmd
= ['promtool', 'check', 'metrics']
46 process
= subprocess
.Popen(testcmd
, stdout
=subprocess
.PIPE
, stdin
=subprocess
.PIPE
, stderr
=subprocess
.STDOUT
, close_fds
=True)
47 output
= process
.communicate(input=content
)
48 except subprocess
.CalledProcessError
as exc
:
49 raise AssertionError('%s failed (%d): %s' % (testcmd
, process
.returncode
, process
.output
))
51 # commented out because promtool returns 3 because of the "_total" suffix warnings
52 #if process.returncode != 0:
53 # raise AssertionError('%s failed (%d): %s' % (testcmd, process.returncode, output))
55 for line
in output
[0].splitlines():
56 if line
.endswith(b
"should have \"_total\" suffix"):
58 raise AssertionError('%s returned an unexpected output. Faulty line is "%s", complete content is "%s"' % (testcmd
, line
, output
))
60 class BasicPrometheusTest(RecPrometheusTest
):
61 _confdir
= 'Prometheus'
64 _wsPassword
= 'secretpassword'
65 _apiKey
= 'secretapikey'
67 _config_template
= """
70 webserver-address=127.0.0.1
72 webserver-allow-from=127.0.0.1
74 """ % (_wsPort
, _wsPassword
, _apiKey
)
76 def testPrometheus(self
):
77 self
.waitForTCPSocket("127.0.0.1", self
._wsPort
)
78 url
= 'http://user:' + self
._wsPassword
+ '@127.0.0.1:' + str(self
._wsPort
) + '/metrics'
79 r
= requests
.get(url
, timeout
=self
._wsTimeout
)
81 self
.assertEqual(r
.status_code
, 200)
82 self
.checkPrometheusContentBasic(r
.text
)
83 self
.checkPrometheusContentPromtool(r
.content
)