Раис Юсупалиев
167 lines
4.4 KiB
Python
167 lines
4.4 KiB
Python
import asyncio
|
|
from email.message import EmailMessage
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from app.adapters.email import SMTPEmailSender, SMTPEmailSenderError
|
|
|
|
|
|
class StubSend:
|
|
def __init__(self, raise_exc: Exception | None = None) -> None:
|
|
self.raise_exc = raise_exc
|
|
self.calls: list[dict[str, Any]] = []
|
|
|
|
async def __call__(
|
|
self,
|
|
message: EmailMessage,
|
|
*,
|
|
hostname: str,
|
|
port: int,
|
|
username: str | None,
|
|
password: str | None,
|
|
use_tls: bool,
|
|
start_tls: bool,
|
|
timeout: float,
|
|
) -> object:
|
|
self.calls.append(
|
|
{
|
|
"message": message,
|
|
"hostname": hostname,
|
|
"port": port,
|
|
"username": username,
|
|
"password": password,
|
|
"use_tls": use_tls,
|
|
"start_tls": start_tls,
|
|
"timeout": timeout,
|
|
}
|
|
)
|
|
if self.raise_exc is not None:
|
|
raise self.raise_exc
|
|
return None
|
|
|
|
|
|
def _make_sender(send: StubSend, **overrides: Any) -> SMTPEmailSender:
|
|
kwargs: dict[str, Any] = {
|
|
"smtp_host": "smtp.test",
|
|
"smtp_port": 587,
|
|
"username": "user",
|
|
"password": "pass",
|
|
"from_address": "no-reply@test",
|
|
"use_tls": True,
|
|
"timeout_seconds": 5.0,
|
|
"send": send,
|
|
}
|
|
kwargs.update(overrides)
|
|
return SMTPEmailSender(**kwargs)
|
|
|
|
|
|
def test_send_email_builds_multipart_message_with_pdf_attachment() -> None:
|
|
send = StubSend()
|
|
sender = _make_sender(send)
|
|
|
|
asyncio.run(
|
|
sender.send_email(
|
|
to="client@example.com",
|
|
subject="Накладная по заказу o-1",
|
|
body="hello",
|
|
attachment_bytes=b"%PDF",
|
|
attachment_filename="waybill_o-1.pdf",
|
|
)
|
|
)
|
|
|
|
assert len(send.calls) == 1
|
|
call = send.calls[0]
|
|
assert call["hostname"] == "smtp.test"
|
|
assert call["port"] == 587
|
|
assert call["username"] == "user"
|
|
assert call["password"] == "pass"
|
|
assert call["use_tls"] is False
|
|
assert call["start_tls"] is True
|
|
assert call["timeout"] == 5.0
|
|
|
|
message: EmailMessage = call["message"]
|
|
assert message["From"] == "no-reply@test"
|
|
assert message["To"] == "client@example.com"
|
|
assert message["Subject"] == "Накладная по заказу o-1"
|
|
|
|
parts = list(message.iter_attachments())
|
|
assert len(parts) == 1
|
|
attachment = parts[0]
|
|
assert attachment.get_content_type() == "application/pdf"
|
|
assert attachment.get_filename() == "waybill_o-1.pdf"
|
|
assert attachment.get_payload(decode=True) == b"%PDF"
|
|
|
|
|
|
def test_send_email_uses_tls_for_port_465() -> None:
|
|
send = StubSend()
|
|
sender = _make_sender(send, smtp_port=465)
|
|
|
|
asyncio.run(
|
|
sender.send_email(
|
|
to="x@example.com",
|
|
subject="s",
|
|
body="b",
|
|
attachment_bytes=b"",
|
|
attachment_filename="f.pdf",
|
|
)
|
|
)
|
|
|
|
call = send.calls[0]
|
|
assert call["use_tls"] is True
|
|
assert call["start_tls"] is False
|
|
|
|
|
|
def test_send_email_disables_tls_when_use_tls_false() -> None:
|
|
send = StubSend()
|
|
sender = _make_sender(send, use_tls=False)
|
|
|
|
asyncio.run(
|
|
sender.send_email(
|
|
to="x@example.com",
|
|
subject="s",
|
|
body="b",
|
|
attachment_bytes=b"",
|
|
attachment_filename="f.pdf",
|
|
)
|
|
)
|
|
|
|
call = send.calls[0]
|
|
assert call["use_tls"] is False
|
|
assert call["start_tls"] is False
|
|
|
|
|
|
def test_send_email_wraps_send_errors_in_sender_error() -> None:
|
|
send = StubSend(raise_exc=RuntimeError("connection refused"))
|
|
sender = _make_sender(send)
|
|
|
|
with pytest.raises(SMTPEmailSenderError, match="connection refused"):
|
|
asyncio.run(
|
|
sender.send_email(
|
|
to="x@example.com",
|
|
subject="s",
|
|
body="b",
|
|
attachment_bytes=b"",
|
|
attachment_filename="f.pdf",
|
|
)
|
|
)
|
|
|
|
|
|
def test_send_email_passes_none_when_credentials_blank() -> None:
|
|
send = StubSend()
|
|
sender = _make_sender(send, username="", password="")
|
|
|
|
asyncio.run(
|
|
sender.send_email(
|
|
to="x@example.com",
|
|
subject="s",
|
|
body="b",
|
|
attachment_bytes=b"",
|
|
attachment_filename="f.pdf",
|
|
)
|
|
)
|
|
|
|
call = send.calls[0]
|
|
assert call["username"] is None
|
|
assert call["password"] is None
|