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>
2026-04-13 02:03:12 +02:00
parent 6019f2aafa
commit be081c68f2
6 changed files with 531 additions and 0 deletions
@@ -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="<h1>Hola</h1><p>Ver adjunto.</p>",
+)
+assert msg.body_html == "<h1>Hola</h1><p>Ver adjunto.</p>"
+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.
@@ -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="",
+    )
@@ -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="<b>Hola Bob</b>",
+    )
+    assert msg.from_addr == "alice@example.com"
+    assert msg.subject == "Hola"
+    assert msg.body_html == "<b>Hola Bob</b>"
+    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", "<p>html</p>")
+    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
@@ -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="<h1>Hola</h1>",
+    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.
@@ -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
@@ -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