- 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/')
-
- 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')
+ if bool(self.config.get("alert", "enhanced_alarms")):
+ self.loop.run_until_complete(
+ service.handle_alarm("test_id", "alarm", self.payload)
+ )
+ requests_post.assert_called_once_with(
+ url="http://alarm-url/",
+ data='{"notify_details": {"alarm_number": 1}}',
+ headers={"content-type": "application/json"},
+ verify=False,
+ timeout=alert_timeout,
+ )
+ else:
+ self.loop.run_until_complete(service.handle_alarm("test_id", "alarm", {}))
+ requests_post.assert_called_once_with(
+ json="{}", url="http://alarm-url/", timeout=alert_timeout
+ )
+
+ @mock.patch.object(VnfAlarmRepository, "get")
+ @mock.patch("requests.post")
+ @mock.patch("osm_policy_module.core.database.db")
+ def test_handle_ok_suppression(self, database, requests_post, get_alarm):
+ alert_timeout = int(self.config.get("alert", "timeout"))
+ mock_alarm = self._build_mock_alarm("test_id", last_action="alarm")
+ get_alarm.return_value = mock_alarm
+ service = AlarmingService(self.config)
+ if bool(self.config.get("alert", "enhanced_alarms")):
+ self.loop.run_until_complete(
+ service.handle_alarm("test_id", "ok", self.payload)
+ )
+ requests_post.assert_called_once_with(
+ url="http://ok-url/",
+ data='{"notify_details": {"alarm_number": 0}}',
+ headers={"content-type": "application/json"},
+ verify=False,
+ timeout=alert_timeout,
+ )
+ else:
+ self.loop.run_until_complete(service.handle_alarm("test_id", "ok", {}))
+ requests_post.assert_called_once_with(
+ json="{}", url="http://ok-url/", timeout=alert_timeout
+ )
+
+ @mock.patch.object(VnfAlarmRepository, "get")
+ @mock.patch("requests.post")
+ @mock.patch("osm_policy_module.core.database.db")
+ def test_handle_insufficientalarm(self, database, requests_post, get_alarm):
+ alert_timeout = int(self.config.get("alert", "timeout"))
+ mock_alarm = self._build_mock_alarm("test_id")
+ get_alarm.return_value = mock_alarm
+ service = AlarmingService(self.config)
+ if bool(self.config.get("alert", "enhanced_alarms")):
+ self.loop.run_until_complete(
+ service.handle_alarm("test_id", "insufficient-data", self.payload)
+ )
+ requests_post.assert_called_once_with(
+ url="http://insufficient-data-url/",
+ data='{"notify_details": {"alarm_number": 0}}',
+ headers={"content-type": "application/json"},
+ verify=False,
+ timeout=alert_timeout,
+ )
+ else:
+ self.loop.run_until_complete(
+ service.handle_alarm("test_id", "insufficient-data", {})
+ )
+ requests_post.assert_called_once_with(
+ json="{}", url="http://insufficient-data-url/", timeout=alert_timeout
+ )
+
+ @mock.patch.object(VnfAlarmRepository, "get")
+ @mock.patch("requests.post")
+ @mock.patch("osm_policy_module.core.database.db")