@mock.patch.object(MonClient, "__init__", lambda *args, **kwargs: None)
@mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
class TestAlarmingService(TestCase):
-
def setUp(self):
self.config = Config()
self.loop = asyncio.new_event_loop()
- @mock.patch.object(VnfAlarmRepository, 'get')
- @mock.patch('requests.post')
- @mock.patch('osm_policy_module.core.database.db')
+ @mock.patch.object(VnfAlarmRepository, "get")
+ @mock.patch("requests.post")
+ @mock.patch("osm_policy_module.core.database.db")
def test_handle_alarm(self, database, requests_post, get_alarm):
- mock_alarm = self._build_mock_alarm('test_id')
+ mock_alarm = self._build_mock_alarm("test_id")
get_alarm.return_value = mock_alarm
service = AlarmingService(self.config)
- self.loop.run_until_complete(service.handle_alarm('test_id', 'alarm', {}))
- requests_post.assert_called_once_with(json='{}', url='http://alarm-url/')
+ self.loop.run_until_complete(service.handle_alarm("test_id", "alarm", {}))
+ requests_post.assert_called_once_with(json="{}", url="http://alarm-url/")
requests_post.reset_mock()
- self.loop.run_until_complete(service.handle_alarm('test_id', 'ok', {}))
- requests_post.assert_called_once_with(json='{}', url='http://ok-url/')
+ self.loop.run_until_complete(service.handle_alarm("test_id", "ok", {}))
+ requests_post.assert_called_once_with(json="{}", url="http://ok-url/")
requests_post.reset_mock()
- self.loop.run_until_complete(service.handle_alarm('test_id', 'insufficient-data', {}))
- requests_post.assert_called_once_with(json='{}', url='http://insufficient-data-url/')
-
- @mock.patch.object(VnfAlarmRepository, 'get')
- @mock.patch('requests.post')
- @mock.patch('osm_policy_module.core.database.db')
+ self.loop.run_until_complete(
+ service.handle_alarm("test_id", "insufficient-data", {})
+ )
+ requests_post.assert_called_once_with(
+ json="{}", url="http://insufficient-data-url/"
+ )
+
+ @mock.patch.object(VnfAlarmRepository, "get")
+ @mock.patch("requests.post")
+ @mock.patch("osm_policy_module.core.database.db")
def test_handle_alarm_unknown_status(self, database, requests_post, get_alarm):
- mock_alarm = self._build_mock_alarm('test_id')
+ mock_alarm = self._build_mock_alarm("test_id")
get_alarm.return_value = mock_alarm
service = AlarmingService(self.config)
- self.loop.run_until_complete(service.handle_alarm('test_id', 'unknown', {}))
+ self.loop.run_until_complete(service.handle_alarm("test_id", "unknown", {}))
requests_post.assert_not_called()
- def _build_mock_alarm(self,
- alarm_id='test_id',
- alarm_url='http://alarm-url/',
- insufficient_data_url='http://insufficient-data-url/',
- ok_url='http://ok-url/'):
+ def _build_mock_alarm(
+ self,
+ alarm_id="test_id",
+ alarm_url="http://alarm-url/",
+ insufficient_data_url="http://insufficient-data-url/",
+ ok_url="http://ok-url/",
+ ):
mock_alarm = mock.Mock()
mock_alarm.alarm_id = alarm_id
insufficient_data_action = mock.Mock()
- insufficient_data_action.type = 'insufficient-data'
+ insufficient_data_action.type = "insufficient-data"
insufficient_data_action.url = insufficient_data_url
alarm_action = mock.Mock()
- alarm_action.type = 'alarm'
+ alarm_action.type = "alarm"
alarm_action.url = alarm_url
ok_action = mock.Mock()
- ok_action.type = 'ok'
+ ok_action.type = "ok"
ok_action.url = ok_url
mock_alarm.actions = [insufficient_data_action, alarm_action, ok_action]
return mock_alarm