[ceph.git] / ceph / src / pybind / mgr / dashboard / tests / test_prometheus.py

# -*- coding: utf-8 -*-
# pylint: disable=protected-access
try:
    from mock import patch
except ImportError:
    from unittest.mock import patch

from .. import mgr
from ..controllers.prometheus import Prometheus, PrometheusNotifications, PrometheusReceiver
from ..tests import ControllerTestCase


class PrometheusControllerTest(ControllerTestCase):
    alert_host = 'http://alertmanager:9093/mock'
    alert_host_api = alert_host + '/api/v1'

    prometheus_host = 'http://prometheus:9090/mock'
    prometheus_host_api = prometheus_host + '/api/v1'

    @classmethod
    def setup_server(cls):
        settings = {
            'ALERTMANAGER_API_HOST': cls.alert_host,
            'PROMETHEUS_API_HOST': cls.prometheus_host
        }
        mgr.get_module_option.side_effect = settings.get
        cls.setup_controllers([Prometheus, PrometheusNotifications, PrometheusReceiver])

    @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", return_value='cephadm')
    @patch("dashboard.controllers.prometheus.mgr.mon_command", return_value=(1, {}, None))
    @patch('requests.request')
    def test_rules_cephadm(self, mock_request, mock_mon_command, mock_get_module_option_ex):
        # in this test we use:
        # in the first call to get_module_option_ex we return 'cephadm' as backend
        # in the second call we return 'True' for 'secure_monitoring_stack' option
        mock_get_module_option_ex.side_effect = lambda module, key, default=None: 'cephadm' \
            if module == 'orchestrator' else True
        self._get('/api/prometheus/rules')
        mock_request.assert_called_with('GET',
                                        self.prometheus_host_api + '/rules',
                                        json=None, params={},
                                        verify=True, auth=None)
        assert mock_mon_command.called

    @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", return_value='cephadm')
    @patch("dashboard.controllers.prometheus.mgr.mon_command", return_value=(1, {}, None))
    @patch('requests.request')
    def test_rules_rook(self, mock_request, mock_mon_command, mock_get_module_option_ex):
        # in this test we use:
        # in the first call to get_module_option_ex we return 'rook' as backend
        mock_get_module_option_ex.side_effect = lambda module, key, default=None: 'rook' \
            if module == 'orchestrator' else None
        self._get('/api/prometheus/rules')
        mock_request.assert_called_with('GET',
                                        self.prometheus_host_api + '/rules',
                                        json=None,
                                        params={},
                                        verify=True, auth=None)
        assert not mock_mon_command.called

    @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c=None: None)
    def test_list(self):
        with patch('requests.request') as mock_request:
            self._get('/api/prometheus')
            mock_request.assert_called_with('GET', self.alert_host_api + '/alerts',
                                            json=None, params={}, verify=True, auth=None)

    @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c=None: None)
    def test_get_silences(self):
        with patch('requests.request') as mock_request:
            self._get('/api/prometheus/silences')
            mock_request.assert_called_with('GET', self.alert_host_api + '/silences',
                                            json=None, params={}, verify=True, auth=None)

    @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c=None: None)
    def test_add_silence(self):
        with patch('requests.request') as mock_request:
            self._post('/api/prometheus/silence', {'id': 'new-silence'})
            mock_request.assert_called_with('POST', self.alert_host_api + '/silences',
                                            params=None, json={'id': 'new-silence'},
                                            verify=True, auth=None)

    @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c=None: None)
    def test_update_silence(self):
        with patch('requests.request') as mock_request:
            self._post('/api/prometheus/silence', {'id': 'update-silence'})
            mock_request.assert_called_with('POST', self.alert_host_api + '/silences',
                                            params=None, json={'id': 'update-silence'},
                                            verify=True, auth=None)

    @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c=None: None)
    def test_expire_silence(self):
        with patch('requests.request') as mock_request:
            self._delete('/api/prometheus/silence/0')
            mock_request.assert_called_with('DELETE', self.alert_host_api + '/silence/0',
                                            json=None, params=None, verify=True, auth=None)

    def test_silences_empty_delete(self):
        with patch('requests.request') as mock_request:
            self._delete('/api/prometheus/silence')
            mock_request.assert_not_called()

    def test_post_on_receiver(self):
        PrometheusReceiver.notifications = []
        self._post('/api/prometheus_receiver', {'name': 'foo'})
        self.assertEqual(len(PrometheusReceiver.notifications), 1)
        notification = PrometheusReceiver.notifications[0]
        self.assertEqual(notification['name'], 'foo')
        self.assertTrue(len(notification['notified']) > 20)

    def test_get_empty_list_with_no_notifications(self):
        PrometheusReceiver.notifications = []
        self._get('/api/prometheus/notifications')
        self.assertStatus(200)
        self.assertJsonBody([])
        self._get('/api/prometheus/notifications?from=last')
        self.assertStatus(200)
        self.assertJsonBody([])

    def test_get_all_notification(self):
        PrometheusReceiver.notifications = []
        self._post('/api/prometheus_receiver', {'name': 'foo'})
        self._post('/api/prometheus_receiver', {'name': 'bar'})
        self._get('/api/prometheus/notifications')
        self.assertStatus(200)
        self.assertJsonBody(PrometheusReceiver.notifications)

    def test_get_last_notification_with_use_of_last_keyword(self):
        PrometheusReceiver.notifications = []
        self._post('/api/prometheus_receiver', {'name': 'foo'})
        self._post('/api/prometheus_receiver', {'name': 'bar'})
        self._get('/api/prometheus/notifications?from=last')
        self.assertStatus(200)
        last = PrometheusReceiver.notifications[1]
        self.assertJsonBody([last])

    def test_get_no_notification_with_unknown_id(self):
        PrometheusReceiver.notifications = []
        self._post('/api/prometheus_receiver', {'name': 'foo'})
        self._post('/api/prometheus_receiver', {'name': 'bar'})
        self._get('/api/prometheus/notifications?from=42')
        self.assertStatus(200)
        self.assertJsonBody([])

    def test_get_no_notification_since_with_last_notification(self):
        PrometheusReceiver.notifications = []
        self._post('/api/prometheus_receiver', {'name': 'foo'})
        notification = PrometheusReceiver.notifications[0]
        self._get('/api/prometheus/notifications?from=' + notification['id'])
        self.assertStatus(200)
        self.assertJsonBody([])

    def test_get_notifications_since_last_notification(self):
        PrometheusReceiver.notifications = []
        self._post('/api/prometheus_receiver', {'name': 'foobar'})
        next_to_last = PrometheusReceiver.notifications[0]
        self._post('/api/prometheus_receiver', {'name': 'foo'})
        self._post('/api/prometheus_receiver', {'name': 'bar'})
        self._get('/api/prometheus/notifications?from=' + next_to_last['id'])
        forelast = PrometheusReceiver.notifications[1]
        last = PrometheusReceiver.notifications[2]
        self.assertEqual(self.json_body(), [forelast, last])
Commit	Line	Data
11fdf7f2 TL	1	# -- coding: utf-8 --
11fdf7f2 TL	2	# pylint: disable=protected-access
9f95a23c TL	3	try:
	4	from mock import patch
	5	except ImportError:
	6	from unittest.mock import patch
494da23a	7
11fdf7f2	8	from .. import mgr
f67539c2	9	from ..controllers.prometheus import Prometheus, PrometheusNotifications, PrometheusReceiver
a4b75251	10	from ..tests import ControllerTestCase
11fdf7f2 TL	11
11fdf7f2 TL	12
494da23a TL	13	class PrometheusControllerTest(ControllerTestCase):
	14	alert_host = 'http://alertmanager:9093/mock'
	15	alert_host_api = alert_host + '/api/v1'
11fdf7f2	16
494da23a TL	17	prometheus_host = 'http://prometheus:9090/mock'
494da23a TL	18	prometheus_host_api = prometheus_host + '/api/v1'
11fdf7f2	19
11fdf7f2 TL	20	@classmethod
	21	def setup_server(cls):
	22	settings = {
494da23a TL	23	'ALERTMANAGER_API_HOST': cls.alert_host,
494da23a TL	24	'PROMETHEUS_API_HOST': cls.prometheus_host
11fdf7f2 TL	25	}
11fdf7f2 TL	26	mgr.get_module_option.side_effect = settings.get
494da23a TL	27	cls.setup_controllers([Prometheus, PrometheusNotifications, PrometheusReceiver])
494da23a TL	28
aee94f69 TL	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',
	41	json=None, params={},
	42	verify=True, auth=None)
	43	assert mock_mon_command.called
	44
	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',
	56	json=None,
	57	params={},
	58	verify=True, auth=None)
	59	assert not mock_mon_command.called
11fdf7f2	60
aee94f69	61	@patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c=None: None)
11fdf7f2	62	def test_list(self):
494da23a TL	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',
1e59de90	66	json=None, params={}, verify=True, auth=None)
494da23a	67
aee94f69	68	@patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c=None: None)
494da23a TL	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',
1e59de90	73	json=None, params={}, verify=True, auth=None)
494da23a	74
aee94f69	75	@patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c=None: None)
494da23a TL	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',
cd265ab1	80	params=None, json={'id': 'new-silence'},
1e59de90	81	verify=True, auth=None)
494da23a	82
aee94f69	83	@patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c=None: None)
494da23a TL	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',
cd265ab1	88	params=None, json={'id': 'update-silence'},
1e59de90	89	verify=True, auth=None)
494da23a	90
aee94f69	91	@patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c=None: None)
494da23a TL	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',
1e59de90	96	json=None, params=None, verify=True, auth=None)
494da23a TL	97
	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()
11fdf7f2 TL	102
	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)
	110
	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([])
	119
	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)
	127
	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])
	136
	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([])
	144
	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([])
	152
	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'])
494da23a	160	forelast = PrometheusReceiver.notifications[1]
11fdf7f2	161	last = PrometheusReceiver.notifications[2]
9f95a23c	162	self.assertEqual(self.json_body(), [forelast, last])