tegwick
442a574cf9
Adds the tests/harness helper package: address(), available(), reset(),
users(), smtp_provider() and mailbox_config().
Tests address mailboxes as harness.address("<test name>"), which slugs the name
under the RFC 2606 reserved harness.email-connect.test domain, so distinct test
names cannot collide and a stray send cannot leave the host. No provisioning
call is needed -- the auth-disabled harness creates the mailbox on first login.
reset() drops every user and message via GreenMail's service reset; the next
login recreates the mailbox empty. It is documented to run at test start so a
crashed test cannot leak state forward, and it is global, so resetting tests
cannot run in parallel against one harness.
Suite: 64 passed with the harness up, 61 passed + 3 skipped with it down.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
139 lines
4.2 KiB
Python
139 lines
4.2 KiB
Python
"""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),
|
|
)
|