def setUp(self):
self.config = Config()
self.loop = asyncio.new_event_loop()
- self.payload = {
- "notify_details": {
- "alarm_number": 0
- }
- }
+ self.payload = {"notify_details": {"alarm_number": 0}}
self.headers = {"content-type": "application/json"}
@mock.patch.object(VnfAlarmRepository, "get")
@mock.patch("requests.post")
@mock.patch("osm_policy_module.core.database.db")
def test_handle_alarm_suppression(self, database, requests_post, get_alarm):
- alert_timeout = int(self.config.get('alert', 'timeout'))
+ alert_timeout = int(self.config.get("alert", "timeout"))
mock_alarm = self._build_mock_alarm("test_id", last_action="ok")
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", "alarm", self.payload))
+ 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
+ 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)
+ 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'))
+ 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
+ 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)
+ 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'))
+ 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')):
+ 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
+ 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
+ json="{}", url="http://insufficient-data-url/", timeout=alert_timeout
)
@mock.patch.object(VnfAlarmRepository, "get")
insufficient_data_url="http://insufficient-data-url/",
ok_url="http://ok-url/",
id_suffix=0,
- last_action="insufficient-data"
+ last_action="insufficient-data",
):
mock_alarm = mock.Mock()
mock_alarm.alarm_id = alarm_id