feat: funciones email SMTP en Python (infra)
smtp_send: conecta+envia+cierra en un paso via smtplib (TLS/STARTTLS/plain). email_build_html: construye EmailMessagePy frozen dataclass con cuerpo HTML. Solo stdlib Python: smtplib, email.mime. Tests con mock SMTP server threading. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,149 @@
|
||||
"""Tests para smtp_send usando un mock SMTP server con smtpd/aiosmtpd o threading."""
|
||||
|
||||
import smtplib
|
||||
import socket
|
||||
import sys
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
from io import StringIO
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
from infra.smtp_send import smtp_send, SMTPConfigPy, EmailAttachmentPy
|
||||
|
||||
|
||||
def _find_free_port() -> int:
|
||||
"""Encuentra un puerto libre en 127.0.0.1."""
|
||||
with socket.socket() as s:
|
||||
s.bind(("127.0.0.1", 0))
|
||||
return s.getsockname()[1]
|
||||
|
||||
|
||||
class _MockSMTPHandler:
|
||||
"""Handler de conexion SMTP minimal para tests."""
|
||||
|
||||
def __init__(self, conn: socket.socket):
|
||||
self.conn = conn
|
||||
|
||||
def run(self):
|
||||
f = self.conn.makefile("rb")
|
||||
try:
|
||||
self._send(b"220 mock SMTP\r\n")
|
||||
while True:
|
||||
line = f.readline()
|
||||
if not line:
|
||||
break
|
||||
cmd = line.decode("utf-8", errors="replace").strip().upper()
|
||||
if cmd.startswith("EHLO") or cmd.startswith("HELO"):
|
||||
self._send(b"250 mock\r\n")
|
||||
elif cmd.startswith("MAIL FROM"):
|
||||
self._send(b"250 OK\r\n")
|
||||
elif cmd.startswith("RCPT TO"):
|
||||
self._send(b"250 OK\r\n")
|
||||
elif cmd.startswith("DATA"):
|
||||
self._send(b"354 End with .\r\n")
|
||||
while True:
|
||||
dl = f.readline()
|
||||
if dl.strip() == b".":
|
||||
break
|
||||
self._send(b"250 OK\r\n")
|
||||
elif cmd.startswith("QUIT"):
|
||||
self._send(b"221 Bye\r\n")
|
||||
break
|
||||
else:
|
||||
self._send(b"502 Not recognized\r\n")
|
||||
finally:
|
||||
self.conn.close()
|
||||
|
||||
def _send(self, data: bytes):
|
||||
self.conn.sendall(data)
|
||||
|
||||
|
||||
class _MockSMTPServer:
|
||||
"""SMTP server minimal corriendo en thread para tests."""
|
||||
|
||||
def __init__(self):
|
||||
self.port = _find_free_port()
|
||||
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
self._sock.bind(("127.0.0.1", self.port))
|
||||
self._sock.listen(5)
|
||||
self._sock.settimeout(2.0)
|
||||
self._stop = threading.Event()
|
||||
self._thread = threading.Thread(target=self._serve, daemon=True)
|
||||
|
||||
def start(self):
|
||||
self._thread.start()
|
||||
return self
|
||||
|
||||
def stop(self):
|
||||
self._stop.set()
|
||||
self._sock.close()
|
||||
|
||||
def _serve(self):
|
||||
while not self._stop.is_set():
|
||||
try:
|
||||
conn, _ = self._sock.accept()
|
||||
t = threading.Thread(target=_MockSMTPHandler(conn).run, daemon=True)
|
||||
t.start()
|
||||
except OSError:
|
||||
break
|
||||
|
||||
def cfg(self) -> SMTPConfigPy:
|
||||
return SMTPConfigPy(host="127.0.0.1", port=self.port, tls_mode="")
|
||||
|
||||
|
||||
def test_envia_texto_plano_via_mock_smtpd():
|
||||
srv = _MockSMTPServer().start()
|
||||
try:
|
||||
smtp_send(
|
||||
srv.cfg(),
|
||||
from_addr="alice@example.com",
|
||||
to=["bob@example.com"],
|
||||
subject="Test plain",
|
||||
body_text="Hello Bob",
|
||||
)
|
||||
finally:
|
||||
srv.stop()
|
||||
|
||||
|
||||
def test_envia_html_via_mock_smtpd():
|
||||
srv = _MockSMTPServer().start()
|
||||
try:
|
||||
smtp_send(
|
||||
srv.cfg(),
|
||||
from_addr="alice@example.com",
|
||||
to=["bob@example.com"],
|
||||
subject="Test HTML",
|
||||
body_html="<b>Hello</b>",
|
||||
)
|
||||
finally:
|
||||
srv.stop()
|
||||
|
||||
|
||||
def test_envia_con_adjunto_via_mock_smtpd():
|
||||
srv = _MockSMTPServer().start()
|
||||
try:
|
||||
att = EmailAttachmentPy(filename="f.txt", content_type="text/plain", data=b"file data")
|
||||
smtp_send(
|
||||
srv.cfg(),
|
||||
from_addr="alice@example.com",
|
||||
to=["bob@example.com"],
|
||||
subject="Test attachment",
|
||||
body_text="See attachment",
|
||||
attachments=[att],
|
||||
)
|
||||
finally:
|
||||
srv.stop()
|
||||
|
||||
|
||||
def test_error_si_host_no_existe():
|
||||
port = _find_free_port()
|
||||
# Port is free but nothing listens — should get connection refused
|
||||
cfg = SMTPConfigPy(host="127.0.0.1", port=port, tls_mode="")
|
||||
try:
|
||||
smtp_send(cfg, "a@x.com", ["b@x.com"], "sub", body_text="hi")
|
||||
assert False, "Expected RuntimeError"
|
||||
except RuntimeError:
|
||||
pass
|
||||
Reference in New Issue
Block a user