govoplan-campaign/dev/mail-testbed/run_transport_smoke.py

#!/usr/bin/env python3
from __future__ import annotations

import imaplib
import mimetypes
import os
import tempfile
import time
from email.message import EmailMessage
from email.utils import formatdate, make_msgid
from pathlib import Path

from govoplan_mail.backend.config import ImapConfig, SmtpConfig, TransportSecurity
from govoplan_mail.backend.sending.imap import append_message_to_sent, list_imap_folders, list_imap_messages, test_imap_login
from govoplan_mail.backend.sending.smtp import send_email_message, test_smtp_login

from govoplan_campaign.backend.services.zip_service import create_zip_archive


def _env(name: str, default: str) -> str:
    return os.environ.get(name, default).strip() or default


SMTP_HOST = _env("GOVOPLAN_MAIL_TEST_SMTP_HOST", "127.0.0.1")
IMAP_HOST = _env("GOVOPLAN_MAIL_TEST_IMAP_HOST", "127.0.0.1")
SMTP_PORT = int(_env("GOVOPLAN_MAIL_TEST_SMTP_PORT", "3025"))
IMAP_PORT = int(_env("GOVOPLAN_MAIL_TEST_IMAP_PORT", "3143"))
USERNAME = _env("GOVOPLAN_MAIL_TEST_USER", "campaign-test@govoplan.test")
PASSWORD = _env("GOVOPLAN_MAIL_TEST_PASSWORD", "campaign-test-password")
SENDER = _env("GOVOPLAN_MAIL_TEST_FROM", USERNAME)
RECIPIENT = _env("GOVOPLAN_MAIL_TEST_RECIPIENT", USERNAME)
SENT_FOLDER = _env("GOVOPLAN_MAIL_TEST_SENT_FOLDER", "Sent")
ZIP_PASSWORD = _env("GOVOPLAN_MAIL_TEST_ZIP_PASSWORD", "zip-test-password")
SERVICE_READY_TIMEOUT_SECONDS = int(_env("GOVOPLAN_MAIL_TEST_READY_TIMEOUT_SECONDS", "45"))


def _smtp_config() -> SmtpConfig:
    return SmtpConfig(
        host=SMTP_HOST,
        port=SMTP_PORT,
        security=TransportSecurity.PLAIN,
        username=USERNAME,
        password=PASSWORD,
        timeout_seconds=10,
    )


def _imap_config() -> ImapConfig:
    return ImapConfig(
        host=IMAP_HOST,
        port=IMAP_PORT,
        security=TransportSecurity.PLAIN,
        username=USERNAME,
        password=PASSWORD,
        sent_folder=SENT_FOLDER,
        timeout_seconds=10,
    )


def _message(subject: str, body: str, attachments: list[Path] | None = None) -> EmailMessage:
    msg = EmailMessage()
    msg["From"] = SENDER
    msg["To"] = RECIPIENT
    msg["Subject"] = subject
    msg["Date"] = formatdate(localtime=False)
    msg["Message-ID"] = make_msgid(domain="govoplan.test")
    msg.set_content(body)

    for attachment in attachments or []:
        content_type = mimetypes.guess_type(attachment.name)[0] or "application/octet-stream"
        maintype, subtype = content_type.split("/", 1)
        msg.add_attachment(attachment.read_bytes(), maintype=maintype, subtype=subtype, filename=attachment.name)
    return msg


def _ensure_folder(config: ImapConfig, folder: str) -> None:
    client = imaplib.IMAP4(host=config.host or "", port=config.port or 143, timeout=config.timeout_seconds)
    try:
        if config.username and config.password:
            client.login(config.username, config.password)
        typ, data = client.create(folder)
        if typ not in {"OK", "NO"}:
            raise RuntimeError(f"Could not create IMAP folder {folder!r}: {data!r}")
    finally:
        try:
            client.logout()
        except Exception:
            pass


def _wait_for_delivery(config: ImapConfig, *, before_count: int, expected_increment: int) -> None:
    deadline = time.monotonic() + 15
    last_count = before_count
    while time.monotonic() < deadline:
        result = list_imap_messages(imap_config=config, folder="INBOX", limit=10)
        last_count = result.total_count
        if last_count >= before_count + expected_increment:
            return
        time.sleep(0.5)
    raise RuntimeError(
        f"Expected at least {expected_increment} new INBOX message(s), "
        f"but count moved from {before_count} to {last_count}."
    )


def _wait_for_service(label: str, check):
    deadline = time.monotonic() + SERVICE_READY_TIMEOUT_SECONDS
    last_error: Exception | None = None
    while time.monotonic() < deadline:
        try:
            return check()
        except Exception as exc:  # GreenMail can bind before SMTP/IMAP is ready.
            last_error = exc
            time.sleep(1)
    raise RuntimeError(f"{label} did not become ready within {SERVICE_READY_TIMEOUT_SECONDS}s: {last_error}") from last_error


def main() -> int:
    smtp = _smtp_config()
    imap = _imap_config()

    smtp_login = _wait_for_service("SMTP", lambda: test_smtp_login(smtp_config=smtp))
    imap_login = _wait_for_service("IMAP", lambda: test_imap_login(imap_config=imap))
    print(f"SMTP login OK: {smtp_login.host}:{smtp_login.port} authenticated={smtp_login.authenticated}")
    print(f"IMAP login OK: {imap_login.host}:{imap_login.port} authenticated={imap_login.authenticated}")

    _wait_for_service("IMAP folder creation", lambda: _ensure_folder(imap, SENT_FOLDER))
    folders = _wait_for_service("IMAP folder listing", lambda: list_imap_folders(imap_config=imap))
    print("IMAP folders: " + ", ".join(folder.name for folder in folders.folders))

    before_inbox = list_imap_messages(imap_config=imap, folder="INBOX", limit=10).total_count

    with tempfile.TemporaryDirectory(prefix="govoplan-campaign-mail-test-") as tmp:
        root = Path(tmp)
        attachment = root / "single-attachment.txt"
        attachment.write_text("GovOPlaN single attachment smoke test\n", encoding="utf-8")
        zip_source = root / "zip-source.txt"
        zip_source.write_text("GovOPlaN password-protected ZIP smoke test\n", encoding="utf-8")
        zip_path = create_zip_archive(root / "password-protected.zip", [zip_source], ZIP_PASSWORD)

        messages = [
            _message("[GovOPlaN test] no attachment", "No attachment SMTP/IMAP smoke."),
            _message("[GovOPlaN test] one attachment", "One attachment SMTP/IMAP smoke.", [attachment]),
            _message("[GovOPlaN test] password ZIP", "Password-protected ZIP SMTP/IMAP smoke.", [zip_path]),
        ]

        for message in messages:
            result = send_email_message(
                message,
                smtp_config=smtp,
                envelope_from=SENDER,
                envelope_recipients=[RECIPIENT],
            )
            print(f"SMTP accepted {result.accepted_count} recipient(s): {message['Subject']}")
            append = append_message_to_sent(bytes(message), imap_config=imap, folder=SENT_FOLDER)
            print(f"IMAP appended {append.bytes_appended} byte(s) to {append.folder}: {message['Subject']}")

    _wait_for_delivery(imap, before_count=before_inbox, expected_increment=3)
    sent_count = list_imap_messages(imap_config=imap, folder=SENT_FOLDER, limit=10).total_count
    if sent_count < 3:
        raise RuntimeError(f"Expected at least 3 messages in {SENT_FOLDER!r}, found {sent_count}.")
    print("Transport smoke OK: no attachment, one attachment, and password-protected ZIP messages verified.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())