egutierrez
ff7da29638
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>
150 lines
4.3 KiB
Python
150 lines
4.3 KiB
Python
"""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
|