mirror of
https://github.com/suitenumerique/messages.git
synced 2026-08-17 21:25:41 +02:00
✨(inbound) add prototype Brevo inbound hook for email delivery
This commit is contained in:
Sylvain Zimmer
@@ -0,0 +1,297 @@
|
||||
"""Brevo inbound channel implementation for handling email from Brevo parse webhooks."""
|
||||
|
||||
import hashlib
|
||||
import logging
|
||||
import secrets
|
||||
from email.utils import parsedate_to_datetime
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.core.validators import validate_email
|
||||
from django.utils import timezone
|
||||
|
||||
from drf_spectacular.utils import extend_schema
|
||||
from rest_framework import status, viewsets
|
||||
from rest_framework.authentication import BaseAuthentication
|
||||
from rest_framework.decorators import action
|
||||
from rest_framework.exceptions import AuthenticationFailed
|
||||
from rest_framework.permissions import IsAuthenticated
|
||||
from rest_framework.response import Response
|
||||
|
||||
from core import models
|
||||
from core.mda.inbound import check_local_recipients, deliver_inbound_message
|
||||
from core.mda.rfc5322 import compose_email
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BrevoAuthentication(BaseAuthentication):
|
||||
"""
|
||||
Custom authentication for Brevo webhook endpoints.
|
||||
|
||||
Supports two authentication methods:
|
||||
1. Channel ID authentication (X-Channel-ID header) - for channel-specific webhooks
|
||||
2. HMAC signature validation (X-Brevo-Signature header) - for global webhooks
|
||||
|
||||
Returns None or (user, auth)
|
||||
"""
|
||||
|
||||
def authenticate(self, request):
|
||||
channel_id = request.headers.get("X-Channel-ID")
|
||||
if channel_id:
|
||||
return self._authenticate_by_channel_id(channel_id)
|
||||
|
||||
signature = request.headers.get("X-Brevo-Signature")
|
||||
if signature:
|
||||
return self._authenticate_by_signature(request, signature)
|
||||
|
||||
raise AuthenticationFailed("Missing authentication credentials")
|
||||
|
||||
def _authenticate_by_channel_id(self, channel_id: str):
|
||||
try:
|
||||
channel = models.Channel.objects.get(id=channel_id, type="brevo")
|
||||
except (models.Channel.DoesNotExist, ValidationError) as e:
|
||||
raise AuthenticationFailed("Invalid channel_id") from e
|
||||
return (None, {"channel": channel, "auth_type": "channel_id"})
|
||||
|
||||
def _authenticate_by_signature(self, request, signature: str):
|
||||
secret = getattr(settings, "BREVO_WEBHOOK_SECRET", None)
|
||||
if not secret:
|
||||
raise AuthenticationFailed("Brevo webhook secret not configured")
|
||||
|
||||
body = request.body
|
||||
expected_signature = hashlib.sha256(secret.encode() + body).hexdigest()
|
||||
if not secrets.compare_digest(signature, expected_signature):
|
||||
raise AuthenticationFailed("Invalid signature")
|
||||
|
||||
return (None, {"auth_type": "hmac"})
|
||||
|
||||
def authenticate_header(self, request):
|
||||
return 'Bearer realm="Brevo"'
|
||||
|
||||
|
||||
def convert_brevo_payload_to_parsed_email(item: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Convert Brevo parsed email payload to internal email format."""
|
||||
from_email = item.get("From", {})
|
||||
to_emails = item.get("To", [])
|
||||
cc_emails = item.get("Cc", [])
|
||||
reply_to = item.get("ReplyTo")
|
||||
|
||||
subject = item.get("Subject", "(no subject)")
|
||||
extracted_message = item.get("ExtractedMarkdownMessage", "")
|
||||
raw_html_body = item.get("RawHtmlBody")
|
||||
raw_text_body = item.get("RawTextBody")
|
||||
in_reply_to = item.get("InReplyTo")
|
||||
sent_at = item.get("SentAtDate")
|
||||
|
||||
html_body = ""
|
||||
if raw_html_body:
|
||||
html_body = raw_html_body
|
||||
elif extracted_message:
|
||||
html_body = extracted_message.replace("\n", "<br/>")
|
||||
|
||||
text_body = raw_text_body or extracted_message or ""
|
||||
|
||||
parsed_email: Dict[str, Any] = {
|
||||
"subject": subject,
|
||||
"from": {
|
||||
"email": from_email.get("Address", ""),
|
||||
"name": from_email.get("Name"),
|
||||
},
|
||||
"to": [
|
||||
{
|
||||
"email": to.get("Address", ""),
|
||||
"name": to.get("Name"),
|
||||
}
|
||||
for to in to_emails
|
||||
],
|
||||
"cc": [
|
||||
{
|
||||
"email": cc.get("Address", ""),
|
||||
"name": cc.get("Name"),
|
||||
}
|
||||
for cc in cc_emails
|
||||
],
|
||||
"headers": {},
|
||||
}
|
||||
|
||||
if reply_to:
|
||||
parsed_email["reply_to"] = {
|
||||
"email": reply_to.get("Address", ""),
|
||||
"name": reply_to.get("Name"),
|
||||
}
|
||||
|
||||
if in_reply_to:
|
||||
parsed_email["in_reply_to"] = in_reply_to
|
||||
parsed_email["references"] = in_reply_to
|
||||
|
||||
if sent_at:
|
||||
try:
|
||||
parsed_email["date"] = parsedate_to_datetime(sent_at)
|
||||
except (ValueError, TypeError):
|
||||
parsed_email["date"] = timezone.now()
|
||||
else:
|
||||
parsed_email["date"] = timezone.now()
|
||||
|
||||
if html_body:
|
||||
parsed_email["htmlBody"] = [{"content": html_body}]
|
||||
if text_body:
|
||||
parsed_email["textBody"] = [{"content": text_body}]
|
||||
|
||||
return parsed_email
|
||||
|
||||
|
||||
class InboundBrevoViewSet(viewsets.GenericViewSet):
|
||||
"""Handles incoming email messages from Brevo inbound parse webhooks."""
|
||||
|
||||
CHANNEL_TYPE = "brevo"
|
||||
CHANNEL_DESCRIPTION = "Brevo inbound email parsing"
|
||||
|
||||
permission_classes = [IsAuthenticated]
|
||||
authentication_classes = [BrevoAuthentication]
|
||||
|
||||
@extend_schema(exclude=True)
|
||||
@action(
|
||||
detail=False,
|
||||
methods=["post"],
|
||||
url_path="webhook",
|
||||
url_name="inbound-brevo-webhook",
|
||||
)
|
||||
def webhook(self, request):
|
||||
"""Handle incoming Brevo webhook with parsed email(s)."""
|
||||
|
||||
data = request.data
|
||||
items = data.get("items", [])
|
||||
|
||||
if not items:
|
||||
return Response(
|
||||
{"detail": "No items in payload"},
|
||||
status=status.HTTP_400_BAD_REQUEST,
|
||||
)
|
||||
|
||||
if not isinstance(items, list):
|
||||
return Response(
|
||||
{"detail": "Items must be a list"},
|
||||
status=status.HTTP_400_BAD_REQUEST,
|
||||
)
|
||||
|
||||
auth_data = request.auth
|
||||
channel = auth_data.get("channel")
|
||||
|
||||
results = []
|
||||
success_count = 0
|
||||
failure_count = 0
|
||||
|
||||
for item in items:
|
||||
result = self._process_brevo_item(item, channel)
|
||||
if result["success"]:
|
||||
success_count += 1
|
||||
else:
|
||||
failure_count += 1
|
||||
results.append(result)
|
||||
|
||||
if failure_count > 0 and success_count == 0:
|
||||
return Response(
|
||||
{
|
||||
"status": "error",
|
||||
"detail": "Failed to process all messages",
|
||||
"results": results,
|
||||
},
|
||||
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
)
|
||||
|
||||
if failure_count > 0:
|
||||
return Response(
|
||||
{
|
||||
"status": "partial_success",
|
||||
"processed": success_count,
|
||||
"failed": failure_count,
|
||||
"results": results,
|
||||
},
|
||||
status=status.HTTP_207_MULTI_STATUS,
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"Successfully processed %d Brevo inbound messages",
|
||||
success_count,
|
||||
)
|
||||
return Response(
|
||||
{
|
||||
"status": "ok",
|
||||
"processed": success_count,
|
||||
"results": results,
|
||||
}
|
||||
)
|
||||
|
||||
def _process_brevo_item(self, item: Dict[str, Any], channel: Optional[models.Channel]) -> Dict[str, Any]:
|
||||
"""Process a single Brevo email item."""
|
||||
try:
|
||||
from_email = item.get("From", {}).get("Address", "")
|
||||
if not from_email:
|
||||
return {"success": False, "error": "Missing From address"}
|
||||
|
||||
recipients = item.get("Recipients", [])
|
||||
if not recipients:
|
||||
recipients = [r.get("Address") for r in item.get("To", []) if r.get("Address")]
|
||||
if not recipients:
|
||||
return {"success": False, "error": "No recipients found"}
|
||||
|
||||
recipient_emails = [r for r in recipients if r]
|
||||
if not recipient_emails:
|
||||
return {"success": False, "error": "No valid recipient addresses"}
|
||||
|
||||
local_recipients = check_local_recipients(recipient_emails)
|
||||
|
||||
parsed_email = convert_brevo_payload_to_parsed_email(item)
|
||||
prepend_headers = [
|
||||
("X-Brevo-Webhook", "inbound"),
|
||||
("Received", "from brevo-inbound"),
|
||||
]
|
||||
|
||||
raw_email = compose_email(parsed_email, prepend_headers=prepend_headers)
|
||||
|
||||
delivered_count = 0
|
||||
for recipient in recipient_emails:
|
||||
if recipient in local_recipients:
|
||||
delivered = deliver_inbound_message(
|
||||
recipient,
|
||||
parsed_email,
|
||||
raw_email,
|
||||
channel=channel,
|
||||
skip_inbound_queue=True,
|
||||
)
|
||||
if delivered:
|
||||
delivered_count += 1
|
||||
|
||||
return {
|
||||
"success": delivered_count > 0,
|
||||
"message_id": item.get("MessageId"),
|
||||
"delivered": delivered_count,
|
||||
"recipients": len(recipient_emails),
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("Error processing Brevo item: %s", e)
|
||||
return {"success": False, "error": str(e)}
|
||||
|
||||
@extend_schema(exclude=True)
|
||||
@action(
|
||||
detail=False,
|
||||
methods=["post"],
|
||||
url_path="check",
|
||||
url_name="inbound-brevo-check",
|
||||
)
|
||||
def check(self, request):
|
||||
"""Check which recipients are locally deliverable."""
|
||||
data = request.data
|
||||
addresses = data.get("addresses", [])
|
||||
if not addresses or not isinstance(addresses, list):
|
||||
return Response(
|
||||
{"detail": "Missing addresses"},
|
||||
status=status.HTTP_400_BAD_REQUEST,
|
||||
)
|
||||
|
||||
local_addresses = check_local_recipients(addresses)
|
||||
results = {address: address in local_addresses for address in addresses}
|
||||
return Response(results)
|
||||
@@ -0,0 +1,487 @@
|
||||
"""Tests for Brevo inbound API endpoints."""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from django.test import override_settings
|
||||
from rest_framework import status
|
||||
from rest_framework.exceptions import AuthenticationFailed
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from core import factories, models
|
||||
from core.api.viewsets.inbound.brevo import (
|
||||
BrevoAuthentication,
|
||||
convert_brevo_payload_to_parsed_email,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(name="api_client")
|
||||
def fixture_api_client():
|
||||
"""Return an API client."""
|
||||
return APIClient()
|
||||
|
||||
|
||||
@pytest.fixture(name="channel")
|
||||
def fixture_channel():
|
||||
"""Create a test channel for Brevo."""
|
||||
mailbox = factories.MailboxFactory()
|
||||
return factories.ChannelFactory(
|
||||
type="brevo",
|
||||
mailbox=mailbox,
|
||||
settings={
|
||||
"config": {"enabled": True},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(name="channel_with_settings")
|
||||
def fixture_channel_with_settings():
|
||||
"""Create a test channel with settings."""
|
||||
mailbox = factories.MailboxFactory()
|
||||
return factories.ChannelFactory(
|
||||
type="brevo",
|
||||
mailbox=mailbox,
|
||||
settings={
|
||||
"config": {"enabled": True},
|
||||
"tags": [],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(name="brevo_payload")
|
||||
def fixture_brevo_payload():
|
||||
"""Return a sample Brevo webhook payload."""
|
||||
return {
|
||||
"items": [
|
||||
{
|
||||
"Uuid": ["1a825d56-029b-4a41-b8e4-61670463431b"],
|
||||
"MessageId": "<test-message-id@example.com>",
|
||||
"InReplyTo": None,
|
||||
"From": {
|
||||
"Name": "Test Sender",
|
||||
"Address": "sender@example.com",
|
||||
},
|
||||
"To": [
|
||||
{
|
||||
"Name": "Test Recipient",
|
||||
"Address": "recipient@test.com",
|
||||
}
|
||||
],
|
||||
"Recipients": ["recipient@test.com"],
|
||||
"Cc": [],
|
||||
"ReplyTo": None,
|
||||
"SentAtDate": "Mon, 15 Jan 2024 10:00:00 +0000",
|
||||
"Subject": "Test Subject",
|
||||
"RawHtmlBody": "<p>Test message body</p>",
|
||||
"RawTextBody": "Test message body",
|
||||
"ExtractedMarkdownMessage": "Test message body",
|
||||
"ExtractedMarkdownSignature": "-- \nTest Sender",
|
||||
"SpamScore": 1.0,
|
||||
"Attachments": [],
|
||||
"Headers": {},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
class TestConvertBrevoPayload:
|
||||
"""Test the Brevo payload conversion function."""
|
||||
|
||||
def test_convert_full_payload(self):
|
||||
"""Test conversion of a full Brevo payload."""
|
||||
item = {
|
||||
"From": {"Address": "sender@example.com", "Name": "Sender"},
|
||||
"To": [{"Address": "recipient@example.com", "Name": "Recipient"}],
|
||||
"Cc": [{"Address": "cc@example.com", "Name": "CC"}],
|
||||
"Subject": "Test Subject",
|
||||
"ExtractedMarkdownMessage": "Hello World",
|
||||
"RawHtmlBody": "<p>Hello World</p>",
|
||||
"RawTextBody": "Hello World",
|
||||
"SentAtDate": "Mon, 15 Jan 2024 10:00:00 +0000",
|
||||
"InReplyTo": "<original@example.com>",
|
||||
}
|
||||
|
||||
result = convert_brevo_payload_to_parsed_email(item)
|
||||
|
||||
assert result["subject"] == "Test Subject"
|
||||
assert result["from"]["email"] == "sender@example.com"
|
||||
assert result["from"]["name"] == "Sender"
|
||||
assert len(result["to"]) == 1
|
||||
assert result["to"][0]["email"] == "recipient@example.com"
|
||||
assert result["in_reply_to"] == "<original@example.com>"
|
||||
assert "htmlBody" in result
|
||||
assert "textBody" in result
|
||||
|
||||
def test_convert_minimal_payload(self):
|
||||
"""Test conversion with minimal payload."""
|
||||
item = {
|
||||
"From": {"Address": "sender@example.com"},
|
||||
"Subject": "Test",
|
||||
}
|
||||
|
||||
result = convert_brevo_payload_to_parsed_email(item)
|
||||
|
||||
assert result["subject"] == "Test"
|
||||
assert result["from"]["email"] == "sender@example.com"
|
||||
assert result["from"]["name"] is None
|
||||
|
||||
def test_convert_without_html_uses_markdown(self):
|
||||
"""Test that markdown is used when HTML is not available."""
|
||||
item = {
|
||||
"From": {"Address": "sender@example.com"},
|
||||
"Subject": "Test",
|
||||
"ExtractedMarkdownMessage": "Markdown **content**",
|
||||
}
|
||||
|
||||
result = convert_brevo_payload_to_parsed_email(item)
|
||||
|
||||
assert "htmlBody" in result
|
||||
assert "**content**" in result["htmlBody"][0]["content"]
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestBrevoAuthentication:
|
||||
"""Test the BrevoAuthentication class."""
|
||||
|
||||
def test_authenticate_with_valid_channel_id(self, channel):
|
||||
"""Test authentication with valid channel ID."""
|
||||
auth = BrevoAuthentication()
|
||||
|
||||
class MockRequest:
|
||||
"""Mock request."""
|
||||
|
||||
def __init__(self, channel_id):
|
||||
self.headers = {"X-Channel-ID": str(channel_id)}
|
||||
|
||||
request = MockRequest(channel.id)
|
||||
user, auth_data = auth.authenticate(request)
|
||||
|
||||
assert user is None
|
||||
assert auth_data["channel"] == channel
|
||||
assert auth_data["auth_type"] == "channel_id"
|
||||
|
||||
def test_authenticate_with_missing_channel_id(self):
|
||||
"""Test authentication fails with missing credentials."""
|
||||
auth = BrevoAuthentication()
|
||||
|
||||
class MockRequest:
|
||||
"""Mock request."""
|
||||
|
||||
def __init__(self):
|
||||
self.headers = {}
|
||||
|
||||
request = MockRequest()
|
||||
|
||||
with pytest.raises(AuthenticationFailed, match="Missing authentication"):
|
||||
auth.authenticate(request)
|
||||
|
||||
def test_authenticate_with_invalid_channel_id(self):
|
||||
"""Test authentication fails with invalid channel ID."""
|
||||
auth = BrevoAuthentication()
|
||||
|
||||
class MockRequest:
|
||||
"""Mock request."""
|
||||
|
||||
def __init__(self):
|
||||
self.headers = {"X-Channel-ID": "invalid-uuid"}
|
||||
|
||||
request = MockRequest()
|
||||
|
||||
with pytest.raises(AuthenticationFailed, match="Invalid channel_id"):
|
||||
auth.authenticate(request)
|
||||
|
||||
@override_settings(BREVO_WEBHOOK_SECRET="test-secret")
|
||||
def test_authenticate_with_valid_hmac(self):
|
||||
"""Test authentication with valid HMAC signature."""
|
||||
auth = BrevoAuthentication()
|
||||
|
||||
secret = "test-secret"
|
||||
body = b'{"items": []}'
|
||||
signature = hashlib.sha256(secret.encode() + body).hexdigest()
|
||||
|
||||
class MockRequest:
|
||||
"""Mock request."""
|
||||
|
||||
def __init__(self):
|
||||
self.headers = {"X-Brevo-Signature": signature}
|
||||
self.body = body
|
||||
|
||||
request = MockRequest()
|
||||
user, auth_data = auth.authenticate(request)
|
||||
|
||||
assert user is None
|
||||
assert auth_data["auth_type"] == "hmac"
|
||||
|
||||
@override_settings(BREVO_WEBHOOK_SECRET="test-secret")
|
||||
def test_authenticate_with_invalid_hmac(self):
|
||||
"""Test authentication fails with invalid HMAC signature."""
|
||||
auth = BrevoAuthentication()
|
||||
|
||||
class MockRequest:
|
||||
"""Mock request."""
|
||||
|
||||
def __init__(self):
|
||||
self.headers = {"X-Brevo-Signature": "invalid-signature"}
|
||||
self.body = b'{"items": []}'
|
||||
|
||||
request = MockRequest()
|
||||
|
||||
with pytest.raises(AuthenticationFailed, match="Invalid signature"):
|
||||
auth.authenticate(request)
|
||||
|
||||
def test_authenticate_without_secret(self):
|
||||
"""Test authentication fails when secret is not configured."""
|
||||
auth = BrevoAuthentication()
|
||||
|
||||
class MockRequest:
|
||||
"""Mock request."""
|
||||
|
||||
def __init__(self):
|
||||
self.headers = {"X-Brevo-Signature": "some-signature"}
|
||||
self.body = b'{"items": []}'
|
||||
|
||||
request = MockRequest()
|
||||
|
||||
with pytest.raises(AuthenticationFailed, match="not configured"):
|
||||
auth.authenticate(request)
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestInboundBrevoWebhook:
|
||||
"""Test the webhook endpoint."""
|
||||
|
||||
@patch("core.api.viewsets.inbound.brevo.deliver_inbound_message")
|
||||
def test_webhook_success(self, mock_deliver, api_client, channel, brevo_payload):
|
||||
"""Test successful message delivery."""
|
||||
mock_deliver.return_value = True
|
||||
|
||||
response = api_client.post(
|
||||
"/api/v1.0/inbound/brevo/webhook/",
|
||||
data=brevo_payload,
|
||||
format="json",
|
||||
HTTP_X_CHANNEL_ID=str(channel.id),
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["status"] == "ok"
|
||||
assert data["processed"] == 1
|
||||
|
||||
def test_webhook_empty_items(self, api_client, channel):
|
||||
"""Test webhook with empty items list."""
|
||||
response = api_client.post(
|
||||
"/api/v1.0/inbound/brevo/webhook/",
|
||||
data={"items": []},
|
||||
format="json",
|
||||
HTTP_X_CHANNEL_ID=str(channel.id),
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
assert response.json()["detail"] == "No items in payload"
|
||||
|
||||
def test_webhook_missing_items(self, api_client, channel):
|
||||
"""Test webhook with missing items field."""
|
||||
response = api_client.post(
|
||||
"/api/v1.0/inbound/brevo/webhook/",
|
||||
data={},
|
||||
format="json",
|
||||
HTTP_X_CHANNEL_ID=str(channel.id),
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
|
||||
def test_webhook_invalid_items_type(self, api_client, channel):
|
||||
"""Test webhook with invalid items type."""
|
||||
response = api_client.post(
|
||||
"/api/v1.0/inbound/brevo/webhook/",
|
||||
data={"items": "not a list"},
|
||||
format="json",
|
||||
HTTP_X_CHANNEL_ID=str(channel.id),
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
assert response.json()["detail"] == "Items must be a list"
|
||||
|
||||
def test_webhook_without_authentication(self, api_client, brevo_payload):
|
||||
"""Test webhook endpoint without authentication."""
|
||||
response = api_client.post(
|
||||
"/api/v1.0/inbound/brevo/webhook/",
|
||||
data=brevo_payload,
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
def test_webhook_item_missing_from(self, api_client, channel):
|
||||
"""Test webhook with item missing From address."""
|
||||
payload = {
|
||||
"items": [
|
||||
{
|
||||
"To": [{"Address": "recipient@test.com"}],
|
||||
"Subject": "Test",
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
response = api_client.post(
|
||||
"/api/v1.0/inbound/brevo/webhook/",
|
||||
data=payload,
|
||||
format="json",
|
||||
HTTP_X_CHANNEL_ID=str(channel.id),
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["processed"] == 0
|
||||
assert data["results"][0]["success"] is False
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestInboundBrevoCheck:
|
||||
"""Test the check endpoint."""
|
||||
|
||||
def test_check_success(self, api_client, channel):
|
||||
"""Test successful recipient check."""
|
||||
response = api_client.post(
|
||||
"/api/v1.0/inbound/brevo/check/",
|
||||
data={"addresses": ["recipient@test.com"]},
|
||||
format="json",
|
||||
HTTP_X_CHANNEL_ID=str(channel.id),
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["recipient@test.com"] is True
|
||||
|
||||
def test_check_missing_addresses(self, api_client, channel):
|
||||
"""Test check with missing addresses."""
|
||||
response = api_client.post(
|
||||
"/api/v1.0/inbound/brevo/check/",
|
||||
data={},
|
||||
format="json",
|
||||
HTTP_X_CHANNEL_ID=str(channel.id),
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
|
||||
def test_check_empty_addresses(self, api_client, channel):
|
||||
"""Test check with empty addresses list."""
|
||||
response = api_client.post(
|
||||
"/api/v1.0/inbound/brevo/check/",
|
||||
data={"addresses": []},
|
||||
format="json",
|
||||
HTTP_X_CHANNEL_ID=str(channel.id),
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestInboundBrevoE2E:
|
||||
"""End-to-end tests for Brevo inbound channel."""
|
||||
|
||||
def test_webhook_message_e2e(self, api_client):
|
||||
"""Test that message is properly created from Brevo payload."""
|
||||
assert models.Message.objects.count() == 0
|
||||
|
||||
mailbox = factories.MailboxFactory()
|
||||
label = factories.LabelFactory(mailbox=mailbox, name="Brevo")
|
||||
channel = factories.ChannelFactory(
|
||||
type="brevo",
|
||||
mailbox=mailbox,
|
||||
settings={
|
||||
"config": {"enabled": True},
|
||||
"tags": [str(label.id)],
|
||||
},
|
||||
)
|
||||
|
||||
payload = {
|
||||
"items": [
|
||||
{
|
||||
"MessageId": "<brevo-test-123@example.com>",
|
||||
"From": {
|
||||
"Name": "External Sender",
|
||||
"Address": "external@example.com",
|
||||
},
|
||||
"To": [
|
||||
{
|
||||
"Name": mailbox.address,
|
||||
"Address": mailbox.address,
|
||||
}
|
||||
],
|
||||
"Recipients": [mailbox.address],
|
||||
"Cc": [],
|
||||
"SentAtDate": "Mon, 15 Jan 2024 10:00:00 +0000",
|
||||
"Subject": "Brevo Test Email",
|
||||
"RawHtmlBody": "<p>Hello from Brevo</p>",
|
||||
"RawTextBody": "Hello from Brevo",
|
||||
"ExtractedMarkdownMessage": "Hello from Brevo",
|
||||
"Attachments": [],
|
||||
"Headers": {},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
response = api_client.post(
|
||||
"/api/v1.0/inbound/brevo/webhook/",
|
||||
data=payload,
|
||||
format="json",
|
||||
HTTP_X_CHANNEL_ID=str(channel.id),
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
assert models.Message.objects.count() == 1
|
||||
message = models.Message.objects.first()
|
||||
|
||||
assert message.thread.accesses.first().mailbox == mailbox
|
||||
assert message.sender.email == "external@example.com"
|
||||
assert message.subject == "Brevo Test Email"
|
||||
|
||||
thread_label_ids = set(message.thread.labels.values_list("id", flat=True))
|
||||
assert label.id in thread_label_ids
|
||||
|
||||
def test_webhook_with_reply(self, api_client):
|
||||
"""Test that reply is properly threaded."""
|
||||
mailbox = factories.MailboxFactory()
|
||||
channel = factories.ChannelFactory(
|
||||
type="brevo",
|
||||
mailbox=mailbox,
|
||||
)
|
||||
|
||||
existing_message = factories.MessageFactory(
|
||||
thread__mailbox=mailbox,
|
||||
mime_id="original-message-id@example.com",
|
||||
)
|
||||
|
||||
payload = {
|
||||
"items": [
|
||||
{
|
||||
"MessageId": "<reply-message-id@example.com>",
|
||||
"InReplyTo": "<original-message-id@example.com>",
|
||||
"From": {
|
||||
"Address": "reply@example.com",
|
||||
},
|
||||
"To": [{"Address": mailbox.address}],
|
||||
"Recipients": [mailbox.address],
|
||||
"Subject": "Re: Original Subject",
|
||||
"ExtractedMarkdownMessage": "This is a reply",
|
||||
"SentAtDate": "Mon, 15 Jan 2024 11:00:00 +0000",
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
response = api_client.post(
|
||||
"/api/v1.0/inbound/brevo/webhook/",
|
||||
data=payload,
|
||||
format="json",
|
||||
HTTP_X_CHANNEL_ID=str(channel.id),
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
assert models.Message.objects.count() == 2
|
||||
new_message = models.Message.objects.exclude(id=existing_message.id).first()
|
||||
|
||||
assert new_message.thread == existing_message.thread
|
||||
@@ -16,6 +16,7 @@ from core.api.viewsets.image_proxy import ImageProxyViewSet
|
||||
from core.api.viewsets.import_message import ImportViewSet, MessagesArchiveUploadViewSet
|
||||
from core.api.viewsets.inbound.mta import InboundMTAViewSet
|
||||
from core.api.viewsets.inbound.widget import InboundWidgetViewSet
|
||||
from core.api.viewsets.inbound.brevo import InboundBrevoViewSet
|
||||
from core.api.viewsets.label import LabelViewSet
|
||||
from core.api.viewsets.mailbox import MailboxViewSet
|
||||
from core.api.viewsets.mailbox_access import MailboxAccessViewSet
|
||||
@@ -98,6 +99,9 @@ inbound_nested_router.register(r"mta", InboundMTAViewSet, basename="inbound-mta"
|
||||
inbound_nested_router.register(
|
||||
r"widget", InboundWidgetViewSet, basename="inbound-widget"
|
||||
)
|
||||
inbound_nested_router.register(
|
||||
r"brevo", InboundBrevoViewSet, basename="inbound-brevo"
|
||||
)
|
||||
|
||||
|
||||
# Router for /maildomains/{maildomain_id}/message-templates/
|
||||
|
||||
Reference in New Issue
Block a user