ceph/src/pybind/mgr/dashboard/tests/test_prometheus.py

   1 # -*- coding: utf-8 -*-
   2 # pylint: disable=protected-access
   3 try:
   4     from mock import patch
   5 except ImportError:
   6     from unittest.mock import patch
   7
   8 from .. import mgr
   9 from ..controllers.prometheus import Prometheus, PrometheusNotifications, PrometheusReceiver
  10 from ..tests import ControllerTestCase
  11
  12
  13 class PrometheusControllerTest(ControllerTestCase):
  14     alert_host = 'http://alertmanager:9093/mock'
  15     alert_host_api = alert_host + '/api/v1'
  16
  17     prometheus_host = 'http://prometheus:9090/mock'
  18     prometheus_host_api = prometheus_host + '/api/v1'
  19
  20     @classmethod
  21     def setup_server(cls):
  22         settings = {
  23             'ALERTMANAGER_API_HOST': cls.alert_host,
  24             'PROMETHEUS_API_HOST': cls.prometheus_host
  25         }
  26         mgr.get_module_option.side_effect = settings.get
  27         cls.setup_controllers([Prometheus, PrometheusNotifications, PrometheusReceiver])
  28
  29     @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c: False)
  30     def test_rules(self):
  31         with patch('requests.request') as mock_request:
  32             self._get('/api/prometheus/rules')
  33             mock_request.assert_called_with('GET', self.prometheus_host_api + '/rules',
  34                                             json=None, params={}, verify=True, auth=None)
  35
  36     @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c: False)
  37     def test_list(self):
  38         with patch('requests.request') as mock_request:
  39             self._get('/api/prometheus')
  40             mock_request.assert_called_with('GET', self.alert_host_api + '/alerts',
  41                                             json=None, params={}, verify=True, auth=None)
  42
  43     @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c: False)
  44     def test_get_silences(self):
  45         with patch('requests.request') as mock_request:
  46             self._get('/api/prometheus/silences')
  47             mock_request.assert_called_with('GET', self.alert_host_api + '/silences',
  48                                             json=None, params={}, verify=True, auth=None)
  49
  50     @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c: False)
  51     def test_add_silence(self):
  52         with patch('requests.request') as mock_request:
  53             self._post('/api/prometheus/silence', {'id': 'new-silence'})
  54             mock_request.assert_called_with('POST', self.alert_host_api + '/silences',
  55                                             params=None, json={'id': 'new-silence'},
  56                                             verify=True, auth=None)
  57
  58     @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c: False)
  59     def test_update_silence(self):
  60         with patch('requests.request') as mock_request:
  61             self._post('/api/prometheus/silence', {'id': 'update-silence'})
  62             mock_request.assert_called_with('POST', self.alert_host_api + '/silences',
  63                                             params=None, json={'id': 'update-silence'},
  64                                             verify=True, auth=None)
  65
  66     @patch("dashboard.controllers.prometheus.mgr.get_module_option_ex", lambda a, b, c: False)
  67     def test_expire_silence(self):
  68         with patch('requests.request') as mock_request:
  69             self._delete('/api/prometheus/silence/0')
  70             mock_request.assert_called_with('DELETE', self.alert_host_api + '/silence/0',
  71                                             json=None, params=None, verify=True, auth=None)
  72
  73     def test_silences_empty_delete(self):
  74         with patch('requests.request') as mock_request:
  75             self._delete('/api/prometheus/silence')
  76             mock_request.assert_not_called()
  77
  78     def test_post_on_receiver(self):
  79         PrometheusReceiver.notifications = []
  80         self._post('/api/prometheus_receiver', {'name': 'foo'})
  81         self.assertEqual(len(PrometheusReceiver.notifications), 1)
  82         notification = PrometheusReceiver.notifications[0]
  83         self.assertEqual(notification['name'], 'foo')
  84         self.assertTrue(len(notification['notified']) > 20)
  85
  86     def test_get_empty_list_with_no_notifications(self):
  87         PrometheusReceiver.notifications = []
  88         self._get('/api/prometheus/notifications')
  89         self.assertStatus(200)
  90         self.assertJsonBody([])
  91         self._get('/api/prometheus/notifications?from=last')
  92         self.assertStatus(200)
  93         self.assertJsonBody([])
  94
  95     def test_get_all_notification(self):
  96         PrometheusReceiver.notifications = []
  97         self._post('/api/prometheus_receiver', {'name': 'foo'})
  98         self._post('/api/prometheus_receiver', {'name': 'bar'})
  99         self._get('/api/prometheus/notifications')
 100         self.assertStatus(200)
 101         self.assertJsonBody(PrometheusReceiver.notifications)
 102
 103     def test_get_last_notification_with_use_of_last_keyword(self):
 104         PrometheusReceiver.notifications = []
 105         self._post('/api/prometheus_receiver', {'name': 'foo'})
 106         self._post('/api/prometheus_receiver', {'name': 'bar'})
 107         self._get('/api/prometheus/notifications?from=last')
 108         self.assertStatus(200)
 109         last = PrometheusReceiver.notifications[1]
 110         self.assertJsonBody([last])
 111
 112     def test_get_no_notification_with_unknown_id(self):
 113         PrometheusReceiver.notifications = []
 114         self._post('/api/prometheus_receiver', {'name': 'foo'})
 115         self._post('/api/prometheus_receiver', {'name': 'bar'})
 116         self._get('/api/prometheus/notifications?from=42')
 117         self.assertStatus(200)
 118         self.assertJsonBody([])
 119
 120     def test_get_no_notification_since_with_last_notification(self):
 121         PrometheusReceiver.notifications = []
 122         self._post('/api/prometheus_receiver', {'name': 'foo'})
 123         notification = PrometheusReceiver.notifications[0]
 124         self._get('/api/prometheus/notifications?from=' + notification['id'])
 125         self.assertStatus(200)
 126         self.assertJsonBody([])
 127
 128     def test_get_notifications_since_last_notification(self):
 129         PrometheusReceiver.notifications = []
 130         self._post('/api/prometheus_receiver', {'name': 'foobar'})
 131         next_to_last = PrometheusReceiver.notifications[0]
 132         self._post('/api/prometheus_receiver', {'name': 'foo'})
 133         self._post('/api/prometheus_receiver', {'name': 'bar'})
 134         self._get('/api/prometheus/notifications?from=' + next_to_last['id'])
 135         forelast = PrometheusReceiver.notifications[1]
 136         last = PrometheusReceiver.notifications[2]
 137         self.assertEqual(self.json_body(), [forelast, last])