"""Test-user addressing and reset for the local GreenMail harness. Start the harness with: docker compose -f tests/harness/docker-compose.yml up -d Everything here is test-only. The harness runs with authentication disabled on loopback, so the credentials below are placeholders rather than secrets; real provider material is routed through OpenBao (`.claude/rules/credential-routing.md`) and must never be pointed at this harness. """ from __future__ import annotations import os import re import socket import urllib.error import urllib.request HOST = "127.0.0.1" SMTP_PORT = 3025 IMAP_PORT = 3143 API_PORT = 8080 #: `.test` is reserved by RFC 2606, so a stray send can never leave the host. DOMAIN = "harness.email-connect.test" #: Accepted and ignored — the harness disables authentication. It still has to #: be non-empty because the scanner requires both credential env vars to be set. PASSWORD = "harness" DEFAULT_SENDER = f"noreply@{DOMAIN}" IMAP_USER_ENV = "EMAIL_CONNECT_HARNESS_IMAP_USER" IMAP_PASSWORD_ENV = "EMAIL_CONNECT_HARNESS_IMAP_PASSWORD" _SLUG_RE = re.compile(r"[^a-z0-9]+") def address(name: str) -> str: """Deterministic mailbox address for a named test user. The harness creates the mailbox on first login, so there is no provisioning step: give each test its own name and the mailboxes cannot collide. >>> address("invitation resend") 'invitation-resend@harness.email-connect.test' """ slug = _SLUG_RE.sub("-", name.strip().lower()).strip("-") if not slug: raise ValueError(f"Test user name produced an empty slug: {name!r}") return f"{slug}@{DOMAIN}" def available(timeout: float = 0.5) -> bool: """True when both mail ports accept connections.""" for port in (SMTP_PORT, IMAP_PORT): try: with socket.create_connection((HOST, port), timeout=timeout): pass except OSError: return False return True def reset() -> None: """Drop every harness user and message. GreenMail's reset clears users *and* their mail; a later login recreates the mailbox empty. Call it in setup rather than teardown so a crashed test cannot leave state for the next one. """ request = urllib.request.Request(f"http://{HOST}:{API_PORT}/api/service/reset", method="POST") try: with urllib.request.urlopen(request, timeout=5) as response: if response.status != 200: raise RuntimeError(f"Harness reset failed with status {response.status}") except urllib.error.URLError as exc: raise RuntimeError(f"Harness reset failed: {exc}") from exc def users() -> list[dict]: """Users the harness has created so far, newest first in GreenMail's order.""" import json with urllib.request.urlopen(f"http://{HOST}:{API_PORT}/api/user", timeout=5) as response: return json.loads(response.read().decode("utf-8")) def smtp_provider(sender: str = DEFAULT_SENDER): """SMTPProvider wired to the harness. GreenMail offers no STARTTLS, so this uses the plaintext transport mode, which `SMTPProvider` permits only for loopback hosts. """ from email_connect.transactional import SMTPProvider return SMTPProvider(HOST, SMTP_PORT, sender, PASSWORD, sender, security="plaintext") def mailbox_config( user_address: str, *, storage_path: str, reports_dir: str, mailbox_id: str = "harness-mailbox", scan=None, ): """AppConfig for scanning one harness mailbox over IMAP. The scanner reads IMAP credentials from named environment variables, so this sets the harness pair as a side effect. """ from email_connect.config import AppConfig, MailboxConfig, ReportsConfig, ScanConfig, StorageConfig os.environ[IMAP_USER_ENV] = user_address os.environ[IMAP_PASSWORD_ENV] = PASSWORD return AppConfig( mailbox=MailboxConfig( id=mailbox_id, protocol="imap", host=HOST, port=IMAP_PORT, tls=False, username_env=IMAP_USER_ENV, password_env=IMAP_PASSWORD_ENV, folder="INBOX", ), scan=scan or ScanConfig(), storage=StorageConfig(path=storage_path), reports=ReportsConfig(output_dir=reports_dir), )