from osm_mon.core.message_bus_client import MessageBusClient
from osm_mon.evaluator.backends.prometheus import PrometheusBackend
from osm_mon.evaluator.evaluator import AlarmStatus
from osm_mon.core.message_bus_client import MessageBusClient
from osm_mon.evaluator.backends.prometheus import PrometheusBackend
from osm_mon.evaluator.evaluator import AlarmStatus
- @mock.patch('osm_mon.core.database.db')
- def test_evaluate_metric(self, db, get_metric_value):
+ def test_evaluate_metric(self, get_metric_value):
service.queue.put.assert_called_with((mock_alarm, AlarmStatus.ALARM))
service.queue.reset_mock()
mock_alarm.operation = 'lt'
service.queue.put.assert_called_with((mock_alarm, AlarmStatus.ALARM))
service.queue.reset_mock()
mock_alarm.operation = 'lt'
service.queue.put.assert_called_with((mock_alarm, AlarmStatus.OK))
service.queue.reset_mock()
get_metric_value.return_value = None
service.queue.put.assert_called_with((mock_alarm, AlarmStatus.OK))
service.queue.reset_mock()
get_metric_value.return_value = None
service.queue.put.assert_called_with((mock_alarm, AlarmStatus.INSUFFICIENT))
@mock.patch('multiprocessing.Process')
@mock.patch.object(EvaluatorService, "_evaluate_metric")
@mock.patch.object(CommonDbClient, "get_vnfd")
@mock.patch.object(CommonDbClient, "get_vnfr")
service.queue.put.assert_called_with((mock_alarm, AlarmStatus.INSUFFICIENT))
@mock.patch('multiprocessing.Process')
@mock.patch.object(EvaluatorService, "_evaluate_metric")
@mock.patch.object(CommonDbClient, "get_vnfd")
@mock.patch.object(CommonDbClient, "get_vnfr")
- @mock.patch.object(AlarmRepository, "list")
- @mock.patch('osm_mon.core.database.db')
- def test_evaluate(self, db, alarm_list, get_vnfr, get_vnfd, evaluate_metric, process):
+ @mock.patch.object(CommonDbClient, "get_alarms")
+ def test_evaluate_alarms(self, alarm_list, get_vnfr, get_vnfd, evaluate_metric, process):
mock_alarm = mock.Mock()
mock_alarm.vdur_name = 'cirros_ns-1-cirros_vnfd-VM-1'
mock_alarm.monitoring_param = 'cirros_vnf_memory_util'
mock_alarm = mock.Mock()
mock_alarm.vdur_name = 'cirros_ns-1-cirros_vnfd-VM-1'
mock_alarm.monitoring_param = 'cirros_vnf_memory_util'
alarm_list.return_value = [mock_alarm]
get_vnfr.return_value = vnfr_record_mock
get_vnfd.return_value = vnfd_record_mock
alarm_list.return_value = [mock_alarm]
get_vnfr.return_value = vnfr_record_mock
get_vnfd.return_value = vnfd_record_mock
- @mock.patch('osm_mon.core.database.db')
- def test_get_metric_value_prometheus(self, db, get_metric_value):
+ def test_get_metric_value_prometheus(self, get_metric_value):
self.config.set('evaluator', 'backend', 'prometheus')
evaluator = EvaluatorService(self.config)
evaluator._get_metric_value('test', {})
self.config.set('evaluator', 'backend', 'prometheus')
evaluator = EvaluatorService(self.config)
evaluator._get_metric_value('test', {})