You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
100 lines
4.5 KiB
Python
100 lines
4.5 KiB
Python
import os
|
|
import unittest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
os.environ.setdefault("DEBUG", "false")
|
|
|
|
from app.core.settings import settings
|
|
from app.services.integrations.providers import BrevoEmailProvider
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, *, status_code: int = 201, payload: dict | None = None, text: str = "") -> None:
|
|
self.status_code = status_code
|
|
self._payload = payload or {}
|
|
self.text = text
|
|
|
|
def json(self) -> dict:
|
|
return self._payload
|
|
|
|
|
|
class BrevoEmailProviderTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_send_email_applies_provider_config_to_payload(self):
|
|
response = _FakeResponse(payload={"messageId": "brevo-123"}, text='{"messageId":"brevo-123"}')
|
|
post_mock = AsyncMock(return_value=response)
|
|
fake_client = type("FakeClient", (), {"post": post_mock})()
|
|
|
|
with patch.object(settings, "brevo_api_key", "brevo-key"), patch.object(
|
|
settings,
|
|
"brevo_sender_email",
|
|
"sender@empresa.com",
|
|
), patch.object(settings, "brevo_sender_name", "Orquestrador"), patch(
|
|
"app.services.integrations.providers.httpx.AsyncClient"
|
|
) as client_cls:
|
|
client_cls.return_value.__aenter__ = AsyncMock(return_value=fake_client)
|
|
client_cls.return_value.__aexit__ = AsyncMock(return_value=None)
|
|
provider = BrevoEmailProvider()
|
|
result = await provider.send_email(
|
|
to_email="destinatario@empresa.com",
|
|
to_name="Operacoes",
|
|
subject="Pedido criado",
|
|
body="Pedido PED-1 criado com sucesso.",
|
|
tags=["order.created"],
|
|
provider_config={
|
|
"sender": {"email": "noreply@empresa.com", "name": "Operacoes"},
|
|
"reply_to": {"email": "atendimento@empresa.com", "name": "Atendimento"},
|
|
"cc": ["financeiro@empresa.com", {"email": "gestor@empresa.com", "name": "Gestor"}],
|
|
"bcc": ["auditoria@empresa.com"],
|
|
"tags": ["ops", "order.created"],
|
|
"headers": {"X-Canal": "orquestrador"},
|
|
"html_content": "<strong>Pedido PED-1 criado com sucesso.</strong>",
|
|
},
|
|
)
|
|
|
|
payload = post_mock.await_args.kwargs["json"]
|
|
self.assertEqual(payload["sender"], {"email": "noreply@empresa.com", "name": "Operacoes"})
|
|
self.assertEqual(payload["replyTo"], {"email": "atendimento@empresa.com", "name": "Atendimento"})
|
|
self.assertEqual(
|
|
payload["cc"],
|
|
[
|
|
{"email": "financeiro@empresa.com"},
|
|
{"email": "gestor@empresa.com", "name": "Gestor"},
|
|
],
|
|
)
|
|
self.assertEqual(payload["bcc"], [{"email": "auditoria@empresa.com"}])
|
|
self.assertEqual(payload["tags"], ["order.created", "ops"])
|
|
self.assertEqual(payload["headers"], {"X-Canal": "orquestrador"})
|
|
self.assertEqual(payload["htmlContent"], "<strong>Pedido PED-1 criado com sucesso.</strong>")
|
|
self.assertEqual(result["message_id"], "brevo-123")
|
|
|
|
async def test_send_email_accepts_route_sender_without_global_sender_email(self):
|
|
response = _FakeResponse(payload={"messageId": "brevo-456"}, text='{"messageId":"brevo-456"}')
|
|
post_mock = AsyncMock(return_value=response)
|
|
fake_client = type("FakeClient", (), {"post": post_mock})()
|
|
|
|
with patch.object(settings, "brevo_api_key", "brevo-key"), patch.object(
|
|
settings,
|
|
"brevo_sender_email",
|
|
"",
|
|
), patch.object(settings, "brevo_sender_name", "Orquestrador"), patch(
|
|
"app.services.integrations.providers.httpx.AsyncClient"
|
|
) as client_cls:
|
|
client_cls.return_value.__aenter__ = AsyncMock(return_value=fake_client)
|
|
client_cls.return_value.__aexit__ = AsyncMock(return_value=None)
|
|
provider = BrevoEmailProvider()
|
|
await provider.send_email(
|
|
to_email="destinatario@empresa.com",
|
|
to_name=None,
|
|
subject="Pedido criado",
|
|
body="Pedido PED-1 criado com sucesso.",
|
|
provider_config={
|
|
"sender": {"email": "noreply@empresa.com", "name": "Operacoes"},
|
|
},
|
|
)
|
|
|
|
payload = post_mock.await_args.kwargs["json"]
|
|
self.assertEqual(payload["sender"], {"email": "noreply@empresa.com", "name": "Operacoes"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |