56 lines
2.0 KiB
Python
56 lines
2.0 KiB
Python
from services import alert_service
|
|
from repositories.monitoring_repository import monitoring_repository
|
|
from repositories.alert_repository import alert_repository
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, status_code=200):
|
|
self.status_code = status_code
|
|
|
|
|
|
def test_send_gotify_logs_delivered_true_on_success(monkeypatch):
|
|
monkeypatch.setattr(alert_service.requests, 'post', lambda *a, **k: _FakeResponse(200))
|
|
alert_service.send_gotify('Test title', 'Test message', priority=5)
|
|
events = list(monitoring_repository._records.values())
|
|
assert events[-1]['alert_sent'] is True
|
|
assert events[-1]['severity'] == 'advisory'
|
|
|
|
|
|
def test_send_gotify_maps_priority_to_severity():
|
|
import requests as real_requests
|
|
|
|
def _fake_post(*a, **k):
|
|
return _FakeResponse(200)
|
|
|
|
alert_service.requests.post = _fake_post
|
|
alert_service.send_gotify('Critical', 'msg', priority=10)
|
|
events = list(monitoring_repository._records.values())
|
|
assert events[-1]['severity'] == 'critical'
|
|
|
|
alert_service.send_gotify('Warning', 'msg', priority=7)
|
|
events = list(monitoring_repository._records.values())
|
|
assert events[-1]['severity'] == 'warning'
|
|
alert_service.requests.post = real_requests.post
|
|
|
|
|
|
def test_send_gotify_never_raises_on_network_failure(monkeypatch):
|
|
def _raise(*a, **k):
|
|
raise Exception('connection refused')
|
|
|
|
monkeypatch.setattr(alert_service.requests, 'post', _raise)
|
|
# Must not raise — side-effect failure only.
|
|
alert_service.send_gotify('Title', 'Message', priority=5)
|
|
events = list(monitoring_repository._records.values())
|
|
assert events[-1]['alert_sent'] is False
|
|
|
|
|
|
def test_create_alert_defaults_dashboard_delivered_true():
|
|
alert = alert_service.create_alert(
|
|
tenant_id='t1', competitor_id='c1', alert_type='price_change',
|
|
message='Price dropped', change_percent=-12
|
|
)
|
|
assert alert['delivered_dashboard'] is True
|
|
assert alert['delivered_email'] is False
|
|
assert alert['read'] is False
|
|
assert alert_repository.list_unread('t1') == [alert]
|