1 # -*- coding: utf-8 -*-
2 # pylint: disable=protected-access
6 from unittest
.mock
import patch
9 from ..controllers
.prometheus
import Prometheus
, PrometheusNotifications
, PrometheusReceiver
10 from ..tests
import ControllerTestCase
13 class PrometheusControllerTest(ControllerTestCase
):
14 alert_host
= 'http://alertmanager:9093/mock'
15 alert_host_api
= alert_host
+ '/api/v1'
17 prometheus_host
= 'http://prometheus:9090/mock'
18 prometheus_host_api
= prometheus_host
+ '/api/v1'
21 def setup_server(cls
):
23 'ALERTMANAGER_API_HOST': cls
.alert_host
,
24 'PROMETHEUS_API_HOST': cls
.prometheus_host
26 mgr
.get_module_option
.side_effect
= settings
.get
27 cls
.setup_controllers([Prometheus
, PrometheusNotifications
, PrometheusReceiver
])
29 @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", return_value
='cephadm')
30 @patch("dashboard.controllers.prometheus.mgr.mon_command", return_value
=(1, {}, None))
31 @patch('requests.request')
32 def test_rules_cephadm(self
, mock_request
, mock_mon_command
, mock_get_module_option_ex
):
33 # in this test we use:
34 # in the first call to get_module_option_ex we return 'cephadm' as backend
35 # in the second call we return 'True' for 'secure_monitoring_stack' option
36 mock_get_module_option_ex
.side_effect
= lambda module
, key
, default
=None: 'cephadm' \
37 if module
== 'orchestrator' else True
38 self
._get
('/api/prometheus/rules')
39 mock_request
.assert_called_with('GET',
40 self
.prometheus_host_api
+ '/rules',
42 verify
=True, auth
=None)
43 assert mock_mon_command
.called
45 @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", return_value
='cephadm')
46 @patch("dashboard.controllers.prometheus.mgr.mon_command", return_value
=(1, {}, None))
47 @patch('requests.request')
48 def test_rules_rook(self
, mock_request
, mock_mon_command
, mock_get_module_option_ex
):
49 # in this test we use:
50 # in the first call to get_module_option_ex we return 'rook' as backend
51 mock_get_module_option_ex
.side_effect
= lambda module
, key
, default
=None: 'rook' \
52 if module
== 'orchestrator' else None
53 self
._get
('/api/prometheus/rules')
54 mock_request
.assert_called_with('GET',
55 self
.prometheus_host_api
+ '/rules',
58 verify
=True, auth
=None)
59 assert not mock_mon_command
.called
61 @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a
, b
, c
=None: None)
63 with
patch('requests.request') as mock_request
:
64 self
._get
('/api/prometheus')
65 mock_request
.assert_called_with('GET', self
.alert_host_api
+ '/alerts',
66 json
=None, params
={}, verify
=True, auth
=None)
68 @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a
, b
, c
=None: None)
69 def test_get_silences(self
):
70 with
patch('requests.request') as mock_request
:
71 self
._get
('/api/prometheus/silences')
72 mock_request
.assert_called_with('GET', self
.alert_host_api
+ '/silences',
73 json
=None, params
={}, verify
=True, auth
=None)
75 @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a
, b
, c
=None: None)
76 def test_add_silence(self
):
77 with
patch('requests.request') as mock_request
:
78 self
._post
('/api/prometheus/silence', {'id': 'new-silence'})
79 mock_request
.assert_called_with('POST', self
.alert_host_api
+ '/silences',
80 params
=None, json
={'id': 'new-silence'},
81 verify
=True, auth
=None)
83 @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a
, b
, c
=None: None)
84 def test_update_silence(self
):
85 with
patch('requests.request') as mock_request
:
86 self
._post
('/api/prometheus/silence', {'id': 'update-silence'})
87 mock_request
.assert_called_with('POST', self
.alert_host_api
+ '/silences',
88 params
=None, json
={'id': 'update-silence'},
89 verify
=True, auth
=None)
91 @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a
, b
, c
=None: None)
92 def test_expire_silence(self
):
93 with
patch('requests.request') as mock_request
:
94 self
._delete
('/api/prometheus/silence/0')
95 mock_request
.assert_called_with('DELETE', self
.alert_host_api
+ '/silence/0',
96 json
=None, params
=None, verify
=True, auth
=None)
98 def test_silences_empty_delete(self
):
99 with
patch('requests.request') as mock_request
:
100 self
._delete
('/api/prometheus/silence')
101 mock_request
.assert_not_called()
103 def test_post_on_receiver(self
):
104 PrometheusReceiver
.notifications
= []
105 self
._post
('/api/prometheus_receiver', {'name': 'foo'})
106 self
.assertEqual(len(PrometheusReceiver
.notifications
), 1)
107 notification
= PrometheusReceiver
.notifications
[0]
108 self
.assertEqual(notification
['name'], 'foo')
109 self
.assertTrue(len(notification
['notified']) > 20)
111 def test_get_empty_list_with_no_notifications(self
):
112 PrometheusReceiver
.notifications
= []
113 self
._get
('/api/prometheus/notifications')
114 self
.assertStatus(200)
115 self
.assertJsonBody([])
116 self
._get
('/api/prometheus/notifications?from=last')
117 self
.assertStatus(200)
118 self
.assertJsonBody([])
120 def test_get_all_notification(self
):
121 PrometheusReceiver
.notifications
= []
122 self
._post
('/api/prometheus_receiver', {'name': 'foo'})
123 self
._post
('/api/prometheus_receiver', {'name': 'bar'})
124 self
._get
('/api/prometheus/notifications')
125 self
.assertStatus(200)
126 self
.assertJsonBody(PrometheusReceiver
.notifications
)
128 def test_get_last_notification_with_use_of_last_keyword(self
):
129 PrometheusReceiver
.notifications
= []
130 self
._post
('/api/prometheus_receiver', {'name': 'foo'})
131 self
._post
('/api/prometheus_receiver', {'name': 'bar'})
132 self
._get
('/api/prometheus/notifications?from=last')
133 self
.assertStatus(200)
134 last
= PrometheusReceiver
.notifications
[1]
135 self
.assertJsonBody([last
])
137 def test_get_no_notification_with_unknown_id(self
):
138 PrometheusReceiver
.notifications
= []
139 self
._post
('/api/prometheus_receiver', {'name': 'foo'})
140 self
._post
('/api/prometheus_receiver', {'name': 'bar'})
141 self
._get
('/api/prometheus/notifications?from=42')
142 self
.assertStatus(200)
143 self
.assertJsonBody([])
145 def test_get_no_notification_since_with_last_notification(self
):
146 PrometheusReceiver
.notifications
= []
147 self
._post
('/api/prometheus_receiver', {'name': 'foo'})
148 notification
= PrometheusReceiver
.notifications
[0]
149 self
._get
('/api/prometheus/notifications?from=' + notification
['id'])
150 self
.assertStatus(200)
151 self
.assertJsonBody([])
153 def test_get_notifications_since_last_notification(self
):
154 PrometheusReceiver
.notifications
= []
155 self
._post
('/api/prometheus_receiver', {'name': 'foobar'})
156 next_to_last
= PrometheusReceiver
.notifications
[0]
157 self
._post
('/api/prometheus_receiver', {'name': 'foo'})
158 self
._post
('/api/prometheus_receiver', {'name': 'bar'})
159 self
._get
('/api/prometheus/notifications?from=' + next_to_last
['id'])
160 forelast
= PrometheusReceiver
.notifications
[1]
161 last
= PrometheusReceiver
.notifications
[2]
162 self
.assertEqual(self
.json_body(), [forelast
, last
])