From ff7da29638764e5beb9a60bff4b620985e748838 Mon Sep 17 00:00:00 2001 From: Egutierrez Date: Mon, 13 Apr 2026 02:03:12 +0200 Subject: [PATCH] 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 --- python/functions/infra/email_build_html.md | 54 ++++++ python/functions/infra/email_build_html.py | 48 +++++ .../functions/infra/email_build_html_test.py | 44 +++++ python/functions/infra/smtp_send.md | 67 +++++++ python/functions/infra/smtp_send.py | 169 ++++++++++++++++++ python/functions/infra/smtp_send_test.py | 149 +++++++++++++++ 6 files changed, 531 insertions(+) create mode 100644 python/functions/infra/email_build_html.md create mode 100644 python/functions/infra/email_build_html.py create mode 100644 python/functions/infra/email_build_html_test.py create mode 100644 python/functions/infra/smtp_send.md create mode 100644 python/functions/infra/smtp_send.py create mode 100644 python/functions/infra/smtp_send_test.py diff --git a/python/functions/infra/email_build_html.md b/python/functions/infra/email_build_html.md new file mode 100644 index 00000000..5cb470b8 --- /dev/null +++ b/python/functions/infra/email_build_html.md @@ -0,0 +1,54 @@ +--- +name: email_build_html +kind: function +lang: py +domain: infra +version: "1.0.0" +purity: pure +signature: "email_build_html(from_addr: str, to: list[str], subject: str, body_html: str) -> EmailMessagePy" +description: "Construye un EmailMessagePy inmutable con cuerpo HTML. El campo body_text queda vacio. Funcion pura sin efectos secundarios." +tags: [email, html, builder, pure, python, dataclass] +uses_functions: [] +uses_types: [] +returns: [] +returns_optional: false +error_type: "" +imports: ["dataclasses"] +params: + - name: from_addr + desc: "direccion del remitente (ej: 'alice@example.com')" + - name: to + desc: "lista de direcciones de destinatarios principales" + - name: subject + desc: "asunto del correo" + - name: body_html + desc: "contenido HTML del cuerpo del mensaje" +output: "EmailMessagePy inmutable con from_addr, to (como tupla), subject y body_html; body_text vacio" +tested: true +tests: + - "construye mensaje html con campos correctos" + - "convierte lista to a tupla inmutable" + - "body_text queda vacio" +test_file_path: "python/functions/infra/email_build_html_test.py" +file_path: "python/functions/infra/email_build_html.py" +--- + +## Ejemplo + +```python +from infra.email_build_html import email_build_html + +msg = email_build_html( + from_addr="alice@example.com", + to=["bob@example.com", "carol@example.com"], + subject="Reporte mensual", + body_html="

Hola

Ver adjunto.

", +) +assert msg.body_html == "

Hola

Ver adjunto.

" +assert msg.body_text == "" +assert isinstance(msg.to, tuple) +``` + +## Notas + +Funcion pura. `EmailMessagePy` es un dataclass frozen — inmutable por construccion. La lista `to` se convierte a tupla para evitar mutacion externa. Para usar con smtp_send, pasar los campos del mensaje como argumentos individuales. diff --git a/python/functions/infra/email_build_html.py b/python/functions/infra/email_build_html.py new file mode 100644 index 00000000..1131a82a --- /dev/null +++ b/python/functions/infra/email_build_html.py @@ -0,0 +1,48 @@ +"""Construccion de EmailMessage Python con cuerpo HTML — funcion pura.""" + +from dataclasses import dataclass, field + + +@dataclass(frozen=True) +class EmailMessagePy: + """Mensaje de email inmutable listo para enviar. + + Construir con email_build_html() o email_build_text(). + """ + from_addr: str + to: tuple[str, ...] + subject: str + body_html: str = "" + body_text: str = "" + cc: tuple[str, ...] = field(default_factory=tuple) + bcc: tuple[str, ...] = field(default_factory=tuple) + headers: tuple[tuple[str, str], ...] = field(default_factory=tuple) + + +def email_build_html( + from_addr: str, + to: list[str], + subject: str, + body_html: str, +) -> EmailMessagePy: + """Construye un EmailMessagePy con cuerpo HTML. + + Funcion pura — no tiene efectos secundarios. El slice `to` se convierte a + tupla para garantizar inmutabilidad. + + Args: + from_addr: Direccion del remitente. + to: Lista de destinatarios principales. + subject: Asunto del correo. + body_html: Contenido HTML del cuerpo del mensaje. + + Returns: + EmailMessagePy inmutable con cuerpo HTML. body_text queda vacio. + """ + return EmailMessagePy( + from_addr=from_addr, + to=tuple(to), + subject=subject, + body_html=body_html, + body_text="", + ) diff --git a/python/functions/infra/email_build_html_test.py b/python/functions/infra/email_build_html_test.py new file mode 100644 index 00000000..7f767c6f --- /dev/null +++ b/python/functions/infra/email_build_html_test.py @@ -0,0 +1,44 @@ +"""Tests para email_build_html.""" + +import sys +import os + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from infra.email_build_html import email_build_html, EmailMessagePy + + +def test_construye_mensaje_html_con_campos_correctos(): + msg = email_build_html( + from_addr="alice@example.com", + to=["bob@example.com"], + subject="Hola", + body_html="Hola Bob", + ) + assert msg.from_addr == "alice@example.com" + assert msg.subject == "Hola" + assert msg.body_html == "Hola Bob" + assert "bob@example.com" in msg.to + + +def test_convierte_lista_to_a_tupla_inmutable(): + to_list = ["a@example.com", "b@example.com"] + msg = email_build_html("f@x.com", to_list, "s", "h") + assert isinstance(msg.to, tuple) + # Mutating the original list must not affect the message + to_list[0] = "mutated@example.com" + assert msg.to[0] == "a@example.com" + + +def test_body_text_queda_vacio(): + msg = email_build_html("f@x.com", ["t@x.com"], "s", "

html

") + assert msg.body_text == "" + + +def test_mensaje_es_inmutable(): + msg = email_build_html("f@x.com", ["t@x.com"], "s", "h") + try: + msg.subject = "changed" + assert False, "Should have raised FrozenInstanceError" + except Exception: + pass # frozen dataclass raises FrozenInstanceError diff --git a/python/functions/infra/smtp_send.md b/python/functions/infra/smtp_send.md new file mode 100644 index 00000000..5173faa3 --- /dev/null +++ b/python/functions/infra/smtp_send.md @@ -0,0 +1,67 @@ +--- +name: smtp_send +kind: function +lang: py +domain: infra +version: "1.0.0" +purity: impure +signature: "smtp_send(cfg: SMTPConfigPy, from_addr: str, to: list[str], subject: str, body_html: str = '', body_text: str = '', cc: list[str] | None = None, bcc: list[str] | None = None, attachments: list[EmailAttachmentPy] | None = None, headers: dict[str, str] | None = None) -> None" +description: "Conecta al servidor SMTP, construye el mensaje MIME y envia el email en un solo paso. Soporta TLS directo (port 465), STARTTLS (port 587) y sin cifrado (port 25). Cierra la conexion automaticamente." +tags: [email, smtp, send, python, smtplib, mime, tls] +uses_functions: [] +uses_types: [] +returns: [] +returns_optional: false +error_type: "error_go_core" +imports: ["smtplib", "email.mime.multipart", "email.mime.text", "email.mime.base", "email.encoders", "dataclasses"] +params: + - name: cfg + desc: "configuracion SMTP: host, port, username, password, tls_mode ('tls', 'starttls' o '')" + - name: from_addr + desc: "direccion del remitente" + - name: to + desc: "lista de destinatarios principales" + - name: subject + desc: "asunto del correo" + - name: body_html + desc: "cuerpo HTML (opcional; puede estar vacio si body_text esta presente)" + - name: body_text + desc: "cuerpo de texto plano (opcional; puede estar vacio si body_html esta presente)" + - name: cc + desc: "lista de destinatarios en copia visible (opcional)" + - name: bcc + desc: "lista de destinatarios en copia oculta (opcional)" + - name: attachments + desc: "lista de EmailAttachmentPy con filename, content_type y data binarios (opcional)" + - name: headers + desc: "diccionario de headers MIME adicionales como X-Mailer (opcional)" +output: "None si el envio fue exitoso; lanza RuntimeError con descripcion del fallo SMTP" +tested: true +tests: + - "envia texto plano via mock smtpd" + - "envia html via mock smtpd" + - "envia con adjunto via mock smtpd" + - "error si host no existe" +test_file_path: "python/functions/infra/smtp_send_test.py" +file_path: "python/functions/infra/smtp_send.py" +--- + +## Ejemplo + +```python +from infra.smtp_send import smtp_send, SMTPConfigPy, EmailAttachmentPy + +cfg = SMTPConfigPy(host="smtp.gmail.com", port=587, username="u@gmail.com", password="app-pw") +smtp_send( + cfg, + from_addr="u@gmail.com", + to=["dest@example.com"], + subject="Reporte", + body_html="

Hola

", + body_text="Hola", +) +``` + +## Notas + +Funcion impura — abre conexion TCP real. Usa solo stdlib Python (smtplib, email). Para TLS directo (port 465) usa `SMTP_SSL`; para STARTTLS (port 587) usa `SMTP` + `starttls()`. Los adjuntos se codifican en base64. BCC se incluye en `sendmail` pero no en las cabeceras MIME visibles. diff --git a/python/functions/infra/smtp_send.py b/python/functions/infra/smtp_send.py new file mode 100644 index 00000000..8d530511 --- /dev/null +++ b/python/functions/infra/smtp_send.py @@ -0,0 +1,169 @@ +"""Envio de email via SMTP con smtplib — sin dependencias externas.""" + +import smtplib +from dataclasses import dataclass, field +from email.mime.base import MIMEBase +from email.mime.multipart import MIMEMultipart +from email.mime.text import MIMEText +from email import encoders + + +@dataclass(frozen=True) +class EmailAttachmentPy: + """Archivo adjunto a un email.""" + filename: str + content_type: str # MIME type, ej: "application/pdf" + data: bytes + + +@dataclass(frozen=True) +class SMTPConfigPy: + """Parametros de conexion SMTP. + + tls_mode puede ser: + "tls" — TLS directo (port 465) + "starttls" — upgrade STARTTLS (port 587) + "" — sin cifrado (port 25) + """ + host: str + port: int + username: str = "" + password: str = "" + tls_mode: str = "starttls" # "tls", "starttls" o "" + + +def smtp_send( + cfg: SMTPConfigPy, + from_addr: str, + to: list[str], + subject: str, + body_html: str = "", + body_text: str = "", + cc: list[str] | None = None, + bcc: list[str] | None = None, + attachments: list[EmailAttachmentPy] | None = None, + headers: dict[str, str] | None = None, +) -> None: + """Conecta al servidor SMTP, envia el email y cierra la conexion. + + Construye el mensaje MIME segun el contenido: + - Solo texto: text/plain + - Solo HTML: text/html + - Ambos: multipart/alternative + - Con adjuntos: multipart/mixed + + Args: + cfg: Configuracion SMTP (host, port, usuario, password, tls_mode). + from_addr: Direccion del remitente. + to: Lista de destinatarios principales. + subject: Asunto del correo. + body_html: Cuerpo HTML (opcional). + body_text: Cuerpo texto plano (opcional). + cc: Lista de destinatarios en copia (opcional). + bcc: Lista de destinatarios en copia oculta (opcional). + attachments: Lista de adjuntos EmailAttachmentPy (opcional). + headers: Headers MIME adicionales como diccionario (opcional). + + Raises: + RuntimeError: Si la conexion SMTP falla, la autenticacion es incorrecta + o el envio no se puede completar. + """ + cc = cc or [] + bcc = bcc or [] + attachments = attachments or [] + headers = headers or {} + + msg = _build_mime(from_addr, to, cc, subject, body_html, body_text, attachments, headers) + + all_recipients = list(to) + list(cc) + list(bcc) + + try: + if cfg.tls_mode == "tls": + smtp = smtplib.SMTP_SSL(cfg.host, cfg.port) + else: + smtp = smtplib.SMTP(cfg.host, cfg.port) + + with smtp: + if cfg.tls_mode == "starttls": + smtp.starttls() + if cfg.username: + smtp.login(cfg.username, cfg.password) + smtp.sendmail(from_addr, all_recipients, msg.as_string()) + + except smtplib.SMTPException as e: + raise RuntimeError(f"smtp_send: SMTP error: {e}") from e + except OSError as e: + raise RuntimeError(f"smtp_send: connection error to {cfg.host}:{cfg.port}: {e}") from e + + +def _build_mime( + from_addr: str, + to: list[str], + cc: list[str], + subject: str, + body_html: str, + body_text: str, + attachments: list[EmailAttachmentPy], + headers: dict[str, str], +) -> MIMEMultipart | MIMEText: + """Construye la estructura MIME del mensaje.""" + has_text = bool(body_text) + has_html = bool(body_html) + has_atts = bool(attachments) + + if has_atts: + outer = MIMEMultipart("mixed") + _set_headers(outer, from_addr, to, cc, subject, headers) + body_part = _build_body_part(body_html, body_text, has_html, has_text) + outer.attach(body_part) + for att in attachments: + outer.attach(_build_attachment(att)) + return outer + + if has_html and has_text: + alt = MIMEMultipart("alternative") + _set_headers(alt, from_addr, to, cc, subject, headers) + alt.attach(MIMEText(body_text, "plain", "utf-8")) + alt.attach(MIMEText(body_html, "html", "utf-8")) + return alt + + if has_html: + msg = MIMEText(body_html, "html", "utf-8") + _set_headers(msg, from_addr, to, cc, subject, headers) + return msg + + msg = MIMEText(body_text or "", "plain", "utf-8") + _set_headers(msg, from_addr, to, cc, subject, headers) + return msg + + +def _build_body_part( + body_html: str, body_text: str, has_html: bool, has_text: bool +) -> MIMEBase: + if has_html and has_text: + alt = MIMEMultipart("alternative") + alt.attach(MIMEText(body_text, "plain", "utf-8")) + alt.attach(MIMEText(body_html, "html", "utf-8")) + return alt + if has_html: + return MIMEText(body_html, "html", "utf-8") + return MIMEText(body_text, "plain", "utf-8") + + +def _set_headers(msg: MIMEBase, from_addr: str, to: list[str], cc: list[str], subject: str, extra: dict[str, str]) -> None: + msg["From"] = from_addr + msg["To"] = ", ".join(to) + if cc: + msg["Cc"] = ", ".join(cc) + msg["Subject"] = subject + for k, v in extra.items(): + msg[k] = v + + +def _build_attachment(att: EmailAttachmentPy) -> MIMEBase: + maintype, subtype = att.content_type.split("/", 1) if "/" in att.content_type else ("application", "octet-stream") + part = MIMEBase(maintype, subtype) + part.set_payload(att.data) + encoders.encode_base64(part) + part.add_header("Content-Disposition", "attachment", filename=att.filename) + return part diff --git a/python/functions/infra/smtp_send_test.py b/python/functions/infra/smtp_send_test.py new file mode 100644 index 00000000..1e336f11 --- /dev/null +++ b/python/functions/infra/smtp_send_test.py @@ -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="Hello", + ) + 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