112 lines
4.0 KiB
Python
112 lines
4.0 KiB
Python
from services import alert_service, email_service
|
|
from repositories.monitoring_repository import monitoring_repository
|
|
from repositories.alert_repository import alert_repository
|
|
from repositories.tenant_repository import tenant_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]
|
|
|
|
|
|
def _make_tenant(**overrides):
|
|
data = {
|
|
'name': 'G Adventures', 'email': 'pm@gadventures.com', 'domain': 'gadventures.com',
|
|
'tier': 'analyse', 'status': 'active', 'max_seats': 5,
|
|
}
|
|
data.update(overrides)
|
|
return tenant_repository.create(data)
|
|
|
|
|
|
def test_send_daily_digest_skips_when_no_undelivered_alerts(monkeypatch):
|
|
tenant = _make_tenant()
|
|
sent_calls = []
|
|
monkeypatch.setattr(email_service, 'send_daily_digest_email', lambda *a, **k: sent_calls.append(1) or True)
|
|
|
|
result = alert_service.send_daily_digest(tenant['id'])
|
|
|
|
assert result is False
|
|
assert sent_calls == []
|
|
|
|
|
|
def test_send_daily_digest_sends_and_marks_delivered(monkeypatch):
|
|
tenant = _make_tenant()
|
|
alert_service.create_alert(tenant['id'], 'c1', 'price_change', 'Price dropped', change_percent=-12)
|
|
alert_service.create_alert(tenant['id'], 'c1', 'new_product', 'New trip added')
|
|
|
|
captured = {}
|
|
monkeypatch.setattr(
|
|
email_service, 'send_daily_digest_email',
|
|
lambda email, name, alerts: captured.update(email=email, name=name, count=len(alerts)) or True
|
|
)
|
|
|
|
result = alert_service.send_daily_digest(tenant['id'])
|
|
|
|
assert result is True
|
|
assert captured['email'] == 'pm@gadventures.com'
|
|
assert captured['count'] == 2
|
|
for alert in alert_repository.list_recent(tenant['id']):
|
|
assert alert['delivered_email'] is True
|
|
|
|
|
|
def test_send_daily_digest_does_not_mark_delivered_on_send_failure(monkeypatch):
|
|
tenant = _make_tenant()
|
|
alert_service.create_alert(tenant['id'], 'c1', 'price_change', 'Price dropped', change_percent=-12)
|
|
monkeypatch.setattr(email_service, 'send_daily_digest_email', lambda *a, **k: False)
|
|
|
|
result = alert_service.send_daily_digest(tenant['id'])
|
|
|
|
assert result is False
|
|
assert alert_repository.list_undelivered_email(tenant['id']) != []
|
|
|
|
|
|
def test_send_daily_digest_unknown_tenant_returns_false():
|
|
assert alert_service.send_daily_digest('does-not-exist') is False
|