saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
257 lines
8.8 KiB
Python
257 lines
8.8 KiB
Python
"""Notification provider unit tests: console (always succeeds), SMTP (stdlib
|
|
smtplib mocked, never touches a real socket), Twilio (respx-mocked REST
|
|
calls), the factory's type routing, and the message builder's per-channel
|
|
formatting."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from app.core.config import Settings
|
|
from app.models.alert import Alert
|
|
from app.models.company import Company
|
|
from app.models.enums import NotificationType, SeverityLevel
|
|
from app.notifications.base import NotificationMessage
|
|
from app.notifications.console import ConsoleProvider
|
|
from app.notifications.factory import get_notification_provider
|
|
from app.notifications.message_builder import build_alert_message
|
|
from app.notifications.smtp_email import SmtpEmailProvider
|
|
from app.notifications.telnyx_sms import TelnyxSmsProvider
|
|
from app.notifications.twilio_sms import TwilioSmsProvider
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_console_provider_always_succeeds():
|
|
result = await ConsoleProvider().send(
|
|
NotificationMessage(destination_value="dev@local", subject="Test", body_text="hi")
|
|
)
|
|
assert result.success is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_smtp_provider_sends_via_smtplib(monkeypatch):
|
|
sent = {}
|
|
|
|
class FakeSmtp:
|
|
def __init__(self, host, port, timeout=10):
|
|
sent["host"] = host
|
|
sent["port"] = port
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def starttls(self):
|
|
sent["starttls"] = True
|
|
|
|
def login(self, username, password):
|
|
sent["login"] = (username, password)
|
|
|
|
def sendmail(self, from_addr, to_addrs, message):
|
|
sent["from_addr"] = from_addr
|
|
sent["to_addrs"] = to_addrs
|
|
sent["message"] = message
|
|
|
|
monkeypatch.setattr("app.notifications.smtp_email.smtplib.SMTP", FakeSmtp)
|
|
|
|
settings = Settings(
|
|
smtp_host="mailpit", smtp_port=1025, smtp_from_email="[email protected]"
|
|
)
|
|
provider = SmtpEmailProvider(settings)
|
|
result = await provider.send(
|
|
NotificationMessage(
|
|
destination_value="[email protected]",
|
|
subject="Alert: Pricing change",
|
|
body_text="Plain text body",
|
|
body_html="<p>HTML body</p>",
|
|
)
|
|
)
|
|
|
|
assert result.success is True
|
|
assert sent["host"] == "mailpit"
|
|
assert sent["to_addrs"] == ["[email protected]"]
|
|
assert "starttls" not in sent # smtp_use_tls defaults False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_smtp_provider_reports_failure_without_raising(monkeypatch):
|
|
class FailingSmtp:
|
|
def __init__(self, host, port, timeout=10):
|
|
raise ConnectionRefusedError("no mailpit running")
|
|
|
|
monkeypatch.setattr("app.notifications.smtp_email.smtplib.SMTP", FailingSmtp)
|
|
|
|
provider = SmtpEmailProvider(Settings())
|
|
result = await provider.send(
|
|
NotificationMessage(destination_value="[email protected]", subject="s", body_text="b")
|
|
)
|
|
|
|
assert result.success is False
|
|
assert "no mailpit running" in result.error
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_twilio_provider_not_configured_fails_cleanly():
|
|
settings = Settings(twilio_account_sid="", twilio_auth_token="", twilio_from_number="")
|
|
result = await TwilioSmsProvider(settings).send(
|
|
NotificationMessage(destination_value="+15551234567", subject="s", body_text="b")
|
|
)
|
|
assert result.success is False
|
|
assert "not configured" in result.error
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_twilio_provider_sends_via_rest_api():
|
|
settings = Settings(
|
|
twilio_account_sid="ACxxxx", twilio_auth_token="secret", twilio_from_number="+15559990000"
|
|
)
|
|
with respx.mock:
|
|
respx.post("https://api.twilio.com/2010-04-01/Accounts/ACxxxx/Messages.json").mock(
|
|
return_value=httpx.Response(201, json={"sid": "SMxxxxx"})
|
|
)
|
|
result = await TwilioSmsProvider(settings).send(
|
|
NotificationMessage(
|
|
destination_value="+15551234567", subject="s", body_text="Alert text"
|
|
)
|
|
)
|
|
|
|
assert result.success is True
|
|
assert result.external_message_id == "SMxxxxx"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_twilio_provider_surfaces_api_error():
|
|
settings = Settings(
|
|
twilio_account_sid="ACxxxx", twilio_auth_token="secret", twilio_from_number="+15559990000"
|
|
)
|
|
with respx.mock:
|
|
respx.post("https://api.twilio.com/2010-04-01/Accounts/ACxxxx/Messages.json").mock(
|
|
return_value=httpx.Response(400, text="Invalid 'To' number")
|
|
)
|
|
result = await TwilioSmsProvider(settings).send(
|
|
NotificationMessage(destination_value="not-a-number", subject="s", body_text="b")
|
|
)
|
|
|
|
assert result.success is False
|
|
assert "400" in result.error
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telnyx_provider_not_configured_fails_cleanly():
|
|
settings = Settings(telnyx_api_key="", telnyx_from_number="")
|
|
result = await TelnyxSmsProvider(settings).send(
|
|
NotificationMessage(destination_value="+15551234567", subject="s", body_text="b")
|
|
)
|
|
assert result.success is False
|
|
assert "not configured" in result.error
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telnyx_provider_sends_via_rest_api():
|
|
settings = Settings(telnyx_api_key="KEYxxxx", telnyx_from_number="+15559990000")
|
|
with respx.mock:
|
|
respx.post("https://api.telnyx.com/v2/messages").mock(
|
|
return_value=httpx.Response(200, json={"data": {"id": "msg-abc123"}})
|
|
)
|
|
result = await TelnyxSmsProvider(settings).send(
|
|
NotificationMessage(
|
|
destination_value="+15551234567", subject="s", body_text="Alert text"
|
|
)
|
|
)
|
|
|
|
assert result.success is True
|
|
assert result.external_message_id == "msg-abc123"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_telnyx_provider_surfaces_api_error():
|
|
settings = Settings(telnyx_api_key="KEYxxxx", telnyx_from_number="+15559990000")
|
|
with respx.mock:
|
|
respx.post("https://api.telnyx.com/v2/messages").mock(
|
|
return_value=httpx.Response(
|
|
403,
|
|
json={
|
|
"errors": [
|
|
{
|
|
"code": "40300",
|
|
"title": "Forbidden",
|
|
"detail": "The from number is not assigned to a messaging profile.",
|
|
}
|
|
]
|
|
},
|
|
)
|
|
)
|
|
result = await TelnyxSmsProvider(settings).send(
|
|
NotificationMessage(destination_value="+15551234567", subject="s", body_text="b")
|
|
)
|
|
|
|
assert result.success is False
|
|
assert "403" in result.error
|
|
assert "messaging profile" in result.error
|
|
|
|
|
|
def test_factory_routes_by_notification_type():
|
|
settings = Settings()
|
|
assert get_notification_provider(NotificationType.EMAIL, settings).provider_name == "smtp"
|
|
assert get_notification_provider(NotificationType.SMS, settings).provider_name == "twilio_sms"
|
|
assert get_notification_provider(NotificationType.CONSOLE, settings).provider_name == "console"
|
|
|
|
|
|
def test_factory_routes_sms_by_sms_provider_setting():
|
|
twilio_settings = Settings(sms_provider="twilio")
|
|
assert (
|
|
get_notification_provider(NotificationType.SMS, twilio_settings).provider_name
|
|
== "twilio_sms"
|
|
)
|
|
|
|
telnyx_settings = Settings(sms_provider="telnyx")
|
|
assert (
|
|
get_notification_provider(NotificationType.SMS, telnyx_settings).provider_name
|
|
== "telnyx_sms"
|
|
)
|
|
|
|
|
|
def _sample_alert() -> Alert:
|
|
return Alert(
|
|
id=uuid.uuid4(),
|
|
company_id=uuid.uuid4(),
|
|
detected_change_id=uuid.uuid4(),
|
|
user_id=uuid.uuid4(),
|
|
title="New leadership hire announced",
|
|
summary="A new VP of Engineering was announced on the careers page.",
|
|
why_it_matters="Signals a scaling push in engineering.",
|
|
severity=SeverityLevel.HIGH,
|
|
confidence=0.82,
|
|
)
|
|
|
|
|
|
def test_message_builder_sms_is_short_and_truncated():
|
|
company = Company(id=uuid.uuid4(), user_id=uuid.uuid4(), name="Acme Mobility", slug="acme")
|
|
message = build_alert_message(
|
|
NotificationType.SMS, "+15551234567", company, _sample_alert(), Settings()
|
|
)
|
|
assert len(message.body_text) <= 480
|
|
assert "HIGH" in message.body_text
|
|
assert message.body_html is None
|
|
|
|
|
|
def test_message_builder_email_includes_context_and_links():
|
|
company = Company(id=uuid.uuid4(), user_id=uuid.uuid4(), name="Acme Mobility", slug="acme")
|
|
alert = _sample_alert()
|
|
message = build_alert_message(
|
|
NotificationType.EMAIL, "[email protected]", company, alert, Settings()
|
|
)
|
|
assert "Acme Mobility" in message.subject
|
|
assert alert.summary in message.body_text
|
|
assert alert.why_it_matters in message.body_text
|
|
assert "/alerts" in message.body_text
|
|
assert "/settings" in message.body_text
|
|
assert message.body_html is not None
|
|
assert alert.summary in message.body_html
|