chore: sync GovOPlaN module split state
This commit is contained in:
167
dev/mail-testbed/run_transport_smoke.py
Normal file
167
dev/mail-testbed/run_transport_smoke.py
Normal file
@@ -0,0 +1,167 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user