email-connect/tests/harness/__init__.py

"""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 deliver_raw(recipient: str, raw_bytes: bytes, sender: str = DEFAULT_SENDER) -> None:
    """Deliver a crafted message verbatim to a harness mailbox.

    The harness accepts everything, so it cannot *generate* a bounce, complaint,
    or deferral. Injecting a crafted `.eml` is how those classes are exercised:
    the message still travels through a real MTA and picks up real `Received:`
    headers, which a fixture read off disk never does.
    """

    import smtplib

    with smtplib.SMTP(HOST, SMTP_PORT, timeout=10) as smtp:
        smtp.sendmail(sender, [recipient], raw_bytes)


def inject_maildir(maildir_dir, raw_bytes: bytes, *, name: str | None = None, subdir: str = "new"):
    """Write a crafted message straight into a Maildir tree.

    The offline counterpart to `deliver_raw`: no MTA involved, so the message
    arrives byte-for-byte with no added headers.
    """

    import time
    from pathlib import Path

    root = Path(maildir_dir)
    for required in ("new", "cur", "tmp"):
        (root / required).mkdir(parents=True, exist_ok=True)
    filename = name or f"{int(time.time())}.M{len(raw_bytes)}P{id(raw_bytes) % 100000}.harness"
    path = root / subdir / filename
    path.write_bytes(raw_bytes)
    return path


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),
    )