From d5672573112226f5442ca7fe9b2d4c4080b9c04a Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 22 Jul 2026 16:38:01 +0200 Subject: [PATCH] test(campaign): prove GreenMail delivery journey --- dev/mail-testbed/run_campaign_acceptance.py | 1312 +++++++++++++++++++ examples/greenmail-delivery/campaign.json | 88 ++ examples/greenmail-delivery/fixture.json | 22 + tests/test_mail_testbed_acceptance.py | 388 ++++++ 4 files changed, 1810 insertions(+) create mode 100644 dev/mail-testbed/run_campaign_acceptance.py create mode 100644 examples/greenmail-delivery/campaign.json create mode 100644 examples/greenmail-delivery/fixture.json create mode 100644 tests/test_mail_testbed_acceptance.py diff --git a/dev/mail-testbed/run_campaign_acceptance.py b/dev/mail-testbed/run_campaign_acceptance.py new file mode 100644 index 0000000..470cc67 --- /dev/null +++ b/dev/mail-testbed/run_campaign_acceptance.py @@ -0,0 +1,1312 @@ +#!/usr/bin/env python3 +"""Run a secret-free Campaign journey against the local GreenMail test bed. + +The runner creates an isolated temporary GovOPlaN database and storage root. It +creates a Mail-owned profile through the public API, materializes the committed +credential-free Campaign fixture with only that profile reference, validates +and builds it, sends the exact generated EML, appends it to Sent, and retains a +bounded JSON evidence projection. Optional failure drills use only the local +test bed and never include raw provider diagnostics in the evidence. +""" + +from __future__ import annotations + +import argparse +import hashlib +import imaplib +import importlib.metadata +import ipaddress +import json +import os +import shutil +import socketserver +import sys +import tempfile +import threading +import time +import tomllib +from collections import Counter +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Iterator, Mapping, Protocol +from uuid import uuid4 + + +SCRIPT_ROOT = Path(__file__).resolve().parent +REPOSITORY_ROOT = SCRIPT_ROOT.parents[1] +DEFAULT_FIXTURE = REPOSITORY_ROOT / "examples" / "greenmail-delivery" / "campaign.json" +EVIDENCE_SCHEMA = "govoplan.campaign.greenmail-acceptance.v1" +EXPECTED_AUDIT_ACTIONS = frozenset( + { + "campaign.created", + "campaign.validated", + "campaign.messages_built", + "campaign.built", + "campaign.sent_now", + "campaign.send_now_rejected", + "campaign.append_sent_enqueued", + } +) +FORBIDDEN_CAMPAIGN_KEYS = frozenset( + { + "credentials", + "imap", + "imap_password", + "password", + "secret", + "smtp", + "smtp_password", + } +) +SEND_RESULT_STATUSES = frozenset( + { + "already_accepted", + "already_claimed", + "cancelled", + "dry_run", + "failed", + "not_claimed", + "outcome_unknown", + "paused", + "smtp_accepted", + } +) +APPEND_RESULT_STATUSES = frozenset( + { + "already_appended", + "append_in_progress", + "appended", + "claim_lost", + "dry_run", + "failed", + "not_claimed", + "not_requested", + "not_sent", + "outcome_unknown", + "skipped", + } +) +REPORT_SEND_STATUSES = frozenset( + { + "cancelled", + "claimed", + "failed_permanent", + "failed_temporary", + "not_queued", + "outcome_unknown", + "queued", + "sending", + "sent", + "skipped", + "smtp_accepted", + } +) +REPORT_IMAP_STATUSES = frozenset( + { + "appended", + "appending", + "failed", + "not_requested", + "outcome_unknown", + "pending", + "skipped", + } +) + + +class AcceptanceError(RuntimeError): + """A bounded acceptance failure safe to show to an operator.""" + + +class JsonResponse(Protocol): + status_code: int + + def json(self) -> Any: ... + + +class ApiClient(Protocol): + def post(self, path: str, **kwargs: Any) -> JsonResponse: ... + + def get(self, path: str, **kwargs: Any) -> JsonResponse: ... + + +@dataclass(frozen=True, slots=True) +class TestbedSettings: + smtp_host: str + smtp_port: int + imap_host: str + imap_port: int + username: str + password: str + sender: str + recipient: str + sent_folder: str + provider_timeout_seconds: int + + @classmethod + def from_environment(cls) -> TestbedSettings: + username = _env("GOVOPLAN_MAIL_TEST_USER", "campaign-test@govoplan.test") + return cls( + smtp_host=_env("GOVOPLAN_MAIL_TEST_SMTP_HOST", "127.0.0.1"), + smtp_port=_positive_env_int("GOVOPLAN_MAIL_TEST_SMTP_PORT", 3025), + imap_host=_env("GOVOPLAN_MAIL_TEST_IMAP_HOST", "127.0.0.1"), + imap_port=_positive_env_int("GOVOPLAN_MAIL_TEST_IMAP_PORT", 3143), + username=username, + 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"), + provider_timeout_seconds=_positive_env_int("GOVOPLAN_MAIL_TEST_READY_TIMEOUT_SECONDS", 45), + ) + + def assert_local_testbed(self) -> None: + for label, host in ( + ("SMTP", self.smtp_host), + ("IMAP", self.imap_host), + ): + try: + address = ipaddress.ip_address(host) + except ValueError as exc: + raise AcceptanceError( + f"{label} acceptance host must be a literal loopback IP address" + ) from exc + if not address.is_loopback: + raise AcceptanceError( + f"{label} acceptance is restricted to the loopback GreenMail test bed; " + "target-provider runs require a separately approved acceptance profile" + ) + + +@dataclass(frozen=True, slots=True) +class CampaignBoundary: + profile_reference_only: bool + smtp_revision_frozen: bool + imap_revision_frozen: bool + resolved_transport_material_present: bool + + def as_dict(self) -> dict[str, bool]: + return { + "profile_reference_only": self.profile_reference_only, + "smtp_revision_frozen": self.smtp_revision_frozen, + "imap_revision_frozen": self.imap_revision_frozen, + "resolved_transport_material_present": self.resolved_transport_material_present, + } + + +def _env(name: str, default: str) -> str: + return os.environ.get(name, default).strip() or default + + +def _positive_env_int(name: str, default: int) -> int: + raw = _env(name, str(default)) + try: + value = int(raw) + except ValueError as exc: + raise AcceptanceError(f"{name} must be a positive integer") from exc + if value <= 0: + raise AcceptanceError(f"{name} must be a positive integer") + return value + + +def _core_package_version() -> str: + """Return the version declared by the Core package supplying this runtime.""" + + import govoplan_core + + module_path = Path(govoplan_core.__file__).resolve() + for parent in module_path.parents: + pyproject = parent / "pyproject.toml" + if not pyproject.is_file(): + continue + try: + with pyproject.open("rb") as handle: + project = tomllib.load(handle).get("project", {}) + except (OSError, tomllib.TOMLDecodeError) as exc: + raise AcceptanceError("Core package metadata could not be read") from exc + if project.get("name") != "govoplan-core": + continue + version = project.get("version") + if isinstance(version, str) and version.strip(): + return version.strip() + raise AcceptanceError("Core package metadata declares no version") + try: + version = importlib.metadata.version("govoplan-core") + except importlib.metadata.PackageNotFoundError as exc: + raise AcceptanceError("Core package version is unavailable") from exc + if not version.strip(): + raise AcceptanceError("Core package version is unavailable") + return version.strip() + + +def required_composition_versions( + fixture_path: Path, + available_versions: Mapping[str, str], +) -> dict[str, str]: + """Return the exact fixture composition or fail before any mail effect.""" + + metadata_path = fixture_path.with_name("fixture.json") + try: + metadata_payload = json.loads(metadata_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise AcceptanceError("Campaign fixture metadata is unavailable or invalid") from exc + required = metadata_payload.get("required_modules") if isinstance(metadata_payload, dict) else None + if not isinstance(required, list) or not required or any(not isinstance(item, str) or not item.strip() for item in required): + raise AcceptanceError("Campaign fixture declares no valid required composition") + normalized = [item.strip() for item in required] + if len(set(normalized)) != len(normalized) or "core" not in normalized: + raise AcceptanceError("Campaign fixture required composition is invalid") + missing = [module_id for module_id in normalized if not str(available_versions.get(module_id) or "").strip()] + if missing: + raise AcceptanceError("Required acceptance component versions are unavailable") + return { + module_id: str(available_versions[module_id]).strip() + for module_id in normalized + } + + +def _expect(response: JsonResponse, status_code: int, label: str) -> dict[str, Any]: + if response.status_code != status_code: + raise AcceptanceError(f"{label} returned HTTP {response.status_code}") + payload = response.json() + if not isinstance(payload, dict): + raise AcceptanceError(f"{label} returned an invalid response") + return payload + + +def _assert_no_forbidden_campaign_keys(value: Any, *, location: str = "campaign") -> None: + if isinstance(value, dict): + for key, nested in value.items(): + if str(key).casefold() in FORBIDDEN_CAMPAIGN_KEYS: + raise AcceptanceError(f"{location} contains forbidden transport material") + _assert_no_forbidden_campaign_keys(nested, location=location) + elif isinstance(value, list): + for nested in value: + _assert_no_forbidden_campaign_keys(nested, location=location) + + +def assert_campaign_boundary( + raw_json: Mapping[str, Any], + execution_snapshot: Mapping[str, Any], + *, + profile_id: str, +) -> CampaignBoundary: + server = raw_json.get("server") + profile_reference_only = server == {"mail_profile_id": profile_id} + _assert_no_forbidden_campaign_keys(raw_json) + _assert_no_forbidden_campaign_keys(execution_snapshot, location="execution snapshot") + resolved_present = any( + key.casefold() in FORBIDDEN_CAMPAIGN_KEYS + for key in _iter_mapping_keys(execution_snapshot) + ) + boundary = CampaignBoundary( + profile_reference_only=profile_reference_only, + smtp_revision_frozen=bool(execution_snapshot.get("smtp_transport_revision")), + imap_revision_frozen=bool(execution_snapshot.get("imap_transport_revision")), + resolved_transport_material_present=resolved_present, + ) + if not all( + ( + boundary.profile_reference_only, + boundary.smtp_revision_frozen, + boundary.imap_revision_frozen, + not boundary.resolved_transport_material_present, + ) + ): + raise AcceptanceError("Campaign did not retain the expected Mail profile boundary") + if execution_snapshot.get("mail_profile_id") != profile_id: + raise AcceptanceError("Campaign execution snapshot references a different Mail profile") + return boundary + + +def _iter_mapping_keys(value: Any) -> Iterator[str]: + if isinstance(value, dict): + for key, nested in value.items(): + yield str(key) + yield from _iter_mapping_keys(nested) + elif isinstance(value, list): + for nested in value: + yield from _iter_mapping_keys(nested) + + +def materialize_campaign_fixture( + fixture_path: Path, + *, + profile_id: str, + settings: TestbedSettings, + scenario: str, + run_token: str, +) -> tuple[dict[str, Any], str]: + raw = json.loads(fixture_path.read_text(encoding="utf-8")) + if not isinstance(raw, dict): + raise AcceptanceError("Campaign fixture must contain a JSON object") + campaign = raw.get("campaign") + entries = raw.get("entries", {}).get("inline") if isinstance(raw.get("entries"), dict) else None + recipients = raw.get("recipients") + if not isinstance(campaign, dict) or not isinstance(entries, list) or len(entries) != 1 or not isinstance(recipients, dict): + raise AcceptanceError("Campaign fixture shape is invalid") + + subject = f"[GovOPlaN acceptance {run_token}] Campaign delivery" + campaign["id"] = f"greenmail-{scenario}-{run_token}" + campaign["name"] = f"GreenMail {scenario.replace('_', ' ')} acceptance" + raw["server"] = {"mail_profile_id": profile_id} + from_addresses = recipients.get("from") + to_addresses = entries[0].get("to") if isinstance(entries[0], dict) else None + fields = entries[0].get("fields") if isinstance(entries[0], dict) else None + if not isinstance(from_addresses, list) or not from_addresses or not isinstance(to_addresses, list) or not to_addresses or not isinstance(fields, dict): + raise AcceptanceError("Campaign fixture address shape is invalid") + from_addresses[0]["email"] = settings.sender + to_addresses[0]["email"] = settings.recipient + fields["acceptance_run"] = run_token + template = raw.get("template") + if not isinstance(template, dict): + raise AcceptanceError("Campaign fixture template is invalid") + template["subject"] = subject + delivery = raw.get("delivery") + if not isinstance(delivery, dict) or not isinstance(delivery.get("imap_append_sent"), dict): + raise AcceptanceError("Campaign fixture delivery settings are invalid") + delivery["imap_append_sent"]["folder"] = settings.sent_folder + _assert_no_forbidden_campaign_keys(raw) + return raw, subject + + +def _profile_payload( + settings: TestbedSettings, + *, + name: str, + smtp_host: str | None = None, + smtp_port: int | None = None, + imap_host: str | None = None, + imap_port: int | None = None, + smtp_password: str | None = None, + imap_password: str | None = None, +) -> dict[str, Any]: + return { + "name": name, + "smtp": { + "host": smtp_host or settings.smtp_host, + "port": smtp_port or settings.smtp_port, + "security": "plain", + }, + "imap": { + "host": imap_host or settings.imap_host, + "port": imap_port or settings.imap_port, + "security": "plain", + "sent_folder": settings.sent_folder, + }, + "credentials": { + "smtp": { + "username": settings.username, + "password": settings.password if smtp_password is None else smtp_password, + }, + "imap": { + "username": settings.username, + "password": settings.password if imap_password is None else imap_password, + }, + }, + } + + +def create_mail_profile( + client: ApiClient, + headers: Mapping[str, str], + settings: TestbedSettings, + *, + name: str, + smtp_host: str | None = None, + smtp_port: int | None = None, + imap_host: str | None = None, + imap_port: int | None = None, + smtp_password: str | None = None, + imap_password: str | None = None, +) -> str: + payload = _expect( + client.post( + "/api/v1/mail/profiles", + headers=dict(headers), + json=_profile_payload( + settings, + name=name, + smtp_host=smtp_host, + smtp_port=smtp_port, + imap_host=imap_host, + imap_port=imap_port, + smtp_password=smtp_password, + imap_password=imap_password, + ), + ), + 201, + "Mail profile creation", + ) + profile_id = str(payload.get("id") or "") + if not profile_id: + raise AcceptanceError("Mail profile creation returned no identifier") + serialized = json.dumps(payload, sort_keys=True, default=str) + if settings.password in serialized: + raise AcceptanceError("Mail profile response exposed credential material") + return profile_id + + +def _status_counts( + results: Any, + *, + allowed: frozenset[str], + label: str, +) -> dict[str, int]: + if not isinstance(results, list): + raise AcceptanceError(f"{label} results are invalid") + statuses: list[str] = [] + for item in results: + if not isinstance(item, dict): + raise AcceptanceError(f"{label} result is invalid") + status = str(item.get("status") or "") + if status not in allowed: + raise AcceptanceError(f"{label} returned an unsupported status") + statuses.append(status) + counter = Counter(statuses) + return dict(sorted(counter.items())) + + +def _send_evidence(payload: Mapping[str, Any]) -> dict[str, Any]: + if payload.get("delivery_mode") != "synchronous": + raise AcceptanceError("Campaign acceptance requires synchronous delivery mode") + return { + "attempted_count": int(payload.get("attempted_count") or 0), + "sent_count": int(payload.get("sent_count") or 0), + "failed_count": int(payload.get("failed_count") or 0), + "outcome_unknown_count": int(payload.get("outcome_unknown_count") or 0), + "skipped_count": int(payload.get("skipped_count") or 0), + "delivery_mode": "synchronous", + "statuses": _status_counts( + payload.get("results"), + allowed=SEND_RESULT_STATUSES, + label="Campaign send", + ), + } + + +def _append_evidence(payload: Mapping[str, Any]) -> dict[str, Any]: + return { + "pending_count": int(payload.get("pending_count") or 0), + "processed_count": int(payload.get("processed_count") or 0), + "appended_count": int(payload.get("appended_count") or 0), + "failed_count": int(payload.get("failed_count") or 0), + "skipped_count": int(payload.get("skipped_count") or 0), + "statuses": _status_counts( + payload.get("results"), + allowed=APPEND_RESULT_STATUSES, + label="Campaign append", + ), + } + + +def _report_evidence(payload: Mapping[str, Any]) -> dict[str, Any]: + cards = payload.get("cards") if isinstance(payload.get("cards"), dict) else {} + statuses = payload.get("status_counts") if isinstance(payload.get("status_counts"), dict) else {} + send = statuses.get("send") if isinstance(statuses.get("send"), dict) else {} + imap = statuses.get("imap") if isinstance(statuses.get("imap"), dict) else {} + allowed_cards = ( + "jobs_total", + "sent", + "smtp_accepted", + "failed", + "outcome_unknown", + "retryable", + "needs_attention", + "imap_appended", + "imap_failed", + ) + return { + "cards": {key: int(cards.get(key) or 0) for key in allowed_cards}, + "send_status_counts": _allowlisted_count_map( + send, + allowed=REPORT_SEND_STATUSES, + label="Campaign report send status", + ), + "imap_status_counts": _allowlisted_count_map( + imap, + allowed=REPORT_IMAP_STATUSES, + label="Campaign report IMAP status", + ), + } + + +def _allowlisted_count_map( + value: Mapping[str, Any], + *, + allowed: frozenset[str], + label: str, +) -> dict[str, int]: + projected: dict[str, int] = {} + for key, count in sorted(value.items()): + clean_key = str(key) + if clean_key not in allowed: + raise AcceptanceError(f"{label} is unsupported") + if not isinstance(count, int) or isinstance(count, bool) or count < 0: + raise AcceptanceError(f"{label} count is invalid") + if count: + projected[clean_key] = count + return projected + + +def execute_campaign_scenario( + client: ApiClient, + headers: Mapping[str, str], + *, + fixture_path: Path, + profile_id: str, + settings: TestbedSettings, + scenario: str, + snapshot_probe: Callable[[str], tuple[Mapping[str, Any], Mapping[str, Any]]], + audit_probe: Callable[[str, str], Mapping[str, int]], + append_sent: bool, + repeat_send: bool, +) -> tuple[dict[str, Any], str]: + run_token = uuid4().hex[:12] + raw, subject = materialize_campaign_fixture( + fixture_path, + profile_id=profile_id, + settings=settings, + scenario=scenario, + run_token=run_token, + ) + created = _expect( + client.post("/api/v1/campaigns", headers=dict(headers), json={"config": raw}), + 200, + f"{scenario} Campaign creation", + ) + campaign_payload = created.get("campaign") if isinstance(created.get("campaign"), dict) else {} + version_payload = created.get("version") if isinstance(created.get("version"), dict) else {} + campaign_id = str(campaign_payload.get("id") or "") + version_id = str(version_payload.get("id") or "") + if not campaign_id or not version_id: + raise AcceptanceError(f"{scenario} Campaign creation returned incomplete identifiers") + + validation = _expect( + client.post( + f"/api/v1/campaigns/versions/{version_id}/validate", + headers=dict(headers), + json={"check_files": False}, + ), + 200, + f"{scenario} Campaign validation", + ) + if validation.get("ok") is not True: + raise AcceptanceError(f"{scenario} Campaign validation did not pass") + build = _expect( + client.post( + f"/api/v1/campaigns/versions/{version_id}/build", + headers=dict(headers), + json={"write_eml": True}, + ), + 200, + f"{scenario} Campaign build", + ) + if int(build.get("built_count") or 0) != 1 or int(build.get("build_failed_count") or 0) != 0: + raise AcceptanceError(f"{scenario} Campaign build did not produce one ready message") + + stored_raw, execution_snapshot = snapshot_probe(version_id) + boundary = assert_campaign_boundary(stored_raw, execution_snapshot, profile_id=profile_id) + send_payload = _expect( + client.post( + f"/api/v1/campaigns/{campaign_id}/send-now", + headers=dict(headers), + json={ + "version_id": version_id, + "validate_before_send": False, + "build_before_send": False, + "use_rate_limit": False, + "enqueue_imap_task": False, + }, + ), + 200, + f"{scenario} Campaign send", + ).get("result") + if not isinstance(send_payload, dict): + raise AcceptanceError(f"{scenario} Campaign send returned invalid evidence") + + repeat_evidence: dict[str, Any] | None = None + if repeat_send: + repeat_response = client.post( + f"/api/v1/campaigns/{campaign_id}/send-now", + headers=dict(headers), + json={ + "version_id": version_id, + "validate_before_send": False, + "build_before_send": False, + "use_rate_limit": False, + "enqueue_imap_task": False, + }, + ) + if repeat_response.status_code in {409, 422}: + repeat_evidence = { + "blocked_before_provider": True, + "http_status": repeat_response.status_code, + } + elif repeat_response.status_code == 200: + repeat_payload = repeat_response.json() + repeat = repeat_payload.get("result") if isinstance(repeat_payload, dict) else None + if not isinstance(repeat, dict): + raise AcceptanceError(f"{scenario} repeated Campaign send returned invalid evidence") + repeat_evidence = { + "blocked_before_provider": False, + **_send_evidence(repeat), + } + if repeat_evidence["sent_count"] or repeat_evidence["attempted_count"]: + raise AcceptanceError("Repeated ordinary send attempted an already accepted Campaign job") + else: + raise AcceptanceError( + f"{scenario} repeated Campaign send returned HTTP {repeat_response.status_code}" + ) + + append_evidence: dict[str, Any] | None = None + if append_sent: + append_payload = _expect( + client.post( + f"/api/v1/campaigns/{campaign_id}/append-sent", + headers=dict(headers), + json={"enqueue_celery": False, "run_inline": True, "dry_run": False}, + ), + 200, + f"{scenario} Campaign append-to-Sent", + ).get("result") + if not isinstance(append_payload, dict): + raise AcceptanceError(f"{scenario} Campaign append returned invalid evidence") + append_evidence = _append_evidence(append_payload) + + report = _expect( + client.get( + f"/api/v1/campaigns/{campaign_id}/report", + headers=dict(headers), + params={"version_id": version_id}, + ), + 200, + f"{scenario} Campaign report", + ) + audit_actions = dict(sorted(audit_probe(campaign_id, version_id).items())) + required_audit_actions = { + "campaign.created", + "campaign.validated", + "campaign.messages_built", + "campaign.sent_now", + } + if append_sent: + required_audit_actions.add("campaign.append_sent_enqueued") + if repeat_evidence and repeat_evidence.get("blocked_before_provider"): + required_audit_actions.add("campaign.send_now_rejected") + if not required_audit_actions.issubset(audit_actions): + raise AcceptanceError(f"{scenario} Campaign audit evidence is incomplete") + + evidence = { + "validation": { + "ok": True, + "error_count": int(validation.get("error_count") or 0), + "warning_count": int(validation.get("warning_count") or 0), + }, + "build": { + "built_count": int(build.get("built_count") or 0), + "build_failed_count": int(build.get("build_failed_count") or 0), + "queueable_count": int(build.get("queueable_count") or 0), + }, + "campaign_mail_boundary": boundary.as_dict(), + "send": _send_evidence(send_payload), + "report": _report_evidence(report), + "audit_actions": audit_actions, + } + if repeat_evidence is not None: + evidence["repeated_send"] = repeat_evidence + if append_evidence is not None: + evidence["append_sent"] = append_evidence + return evidence, subject + + +def _mailbox_state( + client: ApiClient, + headers: Mapping[str, str], + *, + profile_id: str, + folder: str, + subject: str | None, +) -> tuple[int, int]: + payload = _expect( + client.get( + f"/api/v1/mail/profiles/{profile_id}/mailbox/messages", + headers=dict(headers), + params={"folder": folder, "limit": 100, "refresh": True}, + ), + 200, + f"{folder} mailbox verification", + ) + messages = payload.get("messages") if isinstance(payload.get("messages"), list) else [] + matches = sum( + 1 + for message in messages + if isinstance(message, dict) and subject is not None and message.get("subject") == subject + ) + return int(payload.get("total_count") or 0), matches + + +def _wait_for_provider_evidence( + client: ApiClient, + headers: Mapping[str, str], + *, + profile_id: str, + sent_folder: str, + subject: str, + before_inbox: int, + before_sent: int, + timeout_seconds: int, +) -> dict[str, Any]: + deadline = time.monotonic() + timeout_seconds + inbox_count = before_inbox + sent_count = before_sent + inbox_matches = 0 + sent_matches = 0 + while time.monotonic() < deadline: + inbox_count, inbox_matches = _mailbox_state( + client, + headers, + profile_id=profile_id, + folder="INBOX", + subject=subject, + ) + sent_count, sent_matches = _mailbox_state( + client, + headers, + profile_id=profile_id, + folder=sent_folder, + subject=subject, + ) + if inbox_matches > 1 or sent_matches > 1: + raise AcceptanceError("GreenMail contains a duplicate message for this unique Campaign acceptance run") + if inbox_count >= before_inbox + 1 and sent_count >= before_sent + 1 and inbox_matches == 1 and sent_matches == 1: + return { + "inbox_increment": inbox_count - before_inbox, + "sent_increment": sent_count - before_sent, + "unique_subject_matches_in_inbox": inbox_matches, + "unique_subject_matches_in_sent": sent_matches, + } + time.sleep(0.5) + raise AcceptanceError("GreenMail did not expose the accepted Campaign message in both INBOX and Sent") + + +class _DisconnectHandler(socketserver.BaseRequestHandler): + def handle(self) -> None: + self.request.close() + + +class _LoopbackServer(socketserver.TCPServer): + allow_reuse_address = True + + +class _RejectingSmtpHandler(socketserver.StreamRequestHandler): + def handle(self) -> None: + self.wfile.write(b"220 local acceptance SMTP\r\n") + self.wfile.flush() + while line := self.rfile.readline(4096): + command = line.split(None, 1)[0].upper() + if command in {b"EHLO", b"HELO"}: + self.wfile.write(b"250-localhost\r\n250 AUTH PLAIN LOGIN\r\n") + elif command == b"AUTH": + self.wfile.write(b"535 5.7.8 Authentication rejected by acceptance drill\r\n") + elif command == b"QUIT": + self.wfile.write(b"221 bye\r\n") + self.wfile.flush() + return + else: + self.wfile.write(b"550 5.7.1 Command rejected by acceptance drill\r\n") + self.wfile.flush() + + +class _RejectingImapHandler(socketserver.StreamRequestHandler): + def handle(self) -> None: + self.wfile.write(b"* OK local acceptance IMAP\r\n") + self.wfile.flush() + while line := self.rfile.readline(4096): + parts = line.split(None, 2) + tag = parts[0] if parts else b"*" + command = parts[1].upper() if len(parts) > 1 else b"" + if command == b"CAPABILITY": + self.wfile.write(b"* CAPABILITY IMAP4rev1\r\n" + tag + b" OK capability\r\n") + elif command == b"LOGIN": + self.wfile.write(tag + b" NO authentication rejected by acceptance drill\r\n") + elif command == b"LOGOUT": + self.wfile.write(b"* BYE logout\r\n" + tag + b" OK logout\r\n") + self.wfile.flush() + return + else: + self.wfile.write(tag + b" BAD unsupported acceptance command\r\n") + self.wfile.flush() + + +@contextmanager +def disconnecting_smtp_endpoint() -> Iterator[tuple[str, int]]: + server = _LoopbackServer(("127.0.0.1", 0), _DisconnectHandler) + thread = threading.Thread(target=server.serve_forever, name="govoplan-smtp-disconnect", daemon=True) + thread.start() + try: + host, port = server.server_address + yield str(host), int(port) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + +@contextmanager +def rejecting_smtp_endpoint() -> Iterator[tuple[str, int]]: + with _scripted_endpoint(_RejectingSmtpHandler, "govoplan-smtp-reject") as endpoint: + yield endpoint + + +@contextmanager +def rejecting_imap_endpoint() -> Iterator[tuple[str, int]]: + with _scripted_endpoint(_RejectingImapHandler, "govoplan-imap-reject") as endpoint: + yield endpoint + + +@contextmanager +def _scripted_endpoint( + handler: type[socketserver.BaseRequestHandler], + thread_name: str, +) -> Iterator[tuple[str, int]]: + server = _LoopbackServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=server.serve_forever, name=thread_name, daemon=True) + thread.start() + try: + host, port = server.server_address + yield str(host), int(port) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + +def ensure_sent_folder(settings: TestbedSettings) -> None: + deadline = time.monotonic() + settings.provider_timeout_seconds + last_error: Exception | None = None + while time.monotonic() < deadline: + client: imaplib.IMAP4 | None = None + try: + client = imaplib.IMAP4( + host=settings.imap_host, + port=settings.imap_port, + timeout=10, + ) + client.login(settings.username, settings.password) + result, _data = client.create(settings.sent_folder) + if result not in {"OK", "NO"}: + raise AcceptanceError("GreenMail did not permit Sent-folder setup") + return + except Exception as exc: + last_error = exc + time.sleep(1) + finally: + if client is not None: + try: + client.logout() + except Exception: + pass + raise AcceptanceError("GreenMail IMAP did not become ready for the Campaign acceptance run") from last_error + + +def run_acceptance( + client: ApiClient, + headers: Mapping[str, str], + *, + settings: TestbedSettings, + fixture_path: Path, + snapshot_probe: Callable[[str], tuple[Mapping[str, Any], Mapping[str, Any]]], + audit_probe: Callable[[str, str], Mapping[str, int]], + include_failure_drills: bool, + module_versions: Mapping[str, str], +) -> dict[str, Any]: + composition_versions = required_composition_versions(fixture_path, module_versions) + success_profile = create_mail_profile( + client, + headers, + settings, + name="GreenMail Campaign acceptance", + ) + before_inbox, _ = _mailbox_state( + client, + headers, + profile_id=success_profile, + folder="INBOX", + subject=None, + ) + before_sent, _ = _mailbox_state( + client, + headers, + profile_id=success_profile, + folder=settings.sent_folder, + subject=None, + ) + success, subject = execute_campaign_scenario( + client, + headers, + fixture_path=fixture_path, + profile_id=success_profile, + settings=settings, + scenario="success", + snapshot_probe=snapshot_probe, + audit_probe=audit_probe, + append_sent=True, + repeat_send=True, + ) + success["provider_verification"] = _wait_for_provider_evidence( + client, + headers, + profile_id=success_profile, + sent_folder=settings.sent_folder, + subject=subject, + before_inbox=before_inbox, + before_sent=before_sent, + timeout_seconds=settings.provider_timeout_seconds, + ) + _assert_success_evidence(success) + + drills: dict[str, Any] = {} + if include_failure_drills: + with disconnecting_smtp_endpoint() as (host, port): + temporary_profile = create_mail_profile( + client, + headers, + settings, + name="GreenMail temporary SMTP failure drill", + smtp_host=host, + smtp_port=port, + ) + temporary, _ = execute_campaign_scenario( + client, + headers, + fixture_path=fixture_path, + profile_id=temporary_profile, + settings=settings, + scenario="smtp_connection_failure", + snapshot_probe=snapshot_probe, + audit_probe=audit_probe, + append_sent=False, + repeat_send=False, + ) + if temporary["report"]["send_status_counts"] != {"failed_temporary": 1}: + raise AcceptanceError("SMTP connection failure was not classified as temporary") + drills["smtp_connection_failure"] = temporary + + with rejecting_smtp_endpoint() as (host, port): + permanent_profile = create_mail_profile( + client, + headers, + settings, + name="Permanent SMTP failure drill", + smtp_host=host, + smtp_port=port, + ) + permanent, _ = execute_campaign_scenario( + client, + headers, + fixture_path=fixture_path, + profile_id=permanent_profile, + settings=settings, + scenario="smtp_authentication_failure", + snapshot_probe=snapshot_probe, + audit_probe=audit_probe, + append_sent=False, + repeat_send=False, + ) + if permanent["report"]["send_status_counts"] != {"failed_permanent": 1}: + raise AcceptanceError("SMTP authentication failure was not classified as permanent") + drills["smtp_authentication_failure"] = permanent + + with rejecting_imap_endpoint() as (host, port): + imap_profile = create_mail_profile( + client, + headers, + settings, + name="IMAP failure drill", + imap_host=host, + imap_port=port, + ) + imap_failure, _ = execute_campaign_scenario( + client, + headers, + fixture_path=fixture_path, + profile_id=imap_profile, + settings=settings, + scenario="imap_authentication_failure", + snapshot_probe=snapshot_probe, + audit_probe=audit_probe, + append_sent=True, + repeat_send=True, + ) + if imap_failure["report"]["imap_status_counts"] != {"failed": 1}: + raise AcceptanceError("IMAP authentication failure was not retained as a failed append") + drills["imap_authentication_failure"] = imap_failure + + evidence = { + "schema_version": EVIDENCE_SCHEMA, + "generated_at": datetime.now(timezone.utc).isoformat(), + "fixture_sha256": hashlib.sha256(fixture_path.read_bytes()).hexdigest(), + "declared_module_versions": composition_versions, + "target": { + "kind": "local_greenmail", + "smtp_security": "plain", + "imap_security": "plain", + "isolated_temporary_database": True, + }, + "success": success, + "failure_drills": drills, + "coverage": { + "smtp_acceptance": True, + "imap_append": True, + "repeat_ordinary_send_no_duplicate": True, + "smtp_connection_failure": include_failure_drills, + "smtp_temporary_response": False, + "smtp_permanent_authentication_failure": include_failure_drills, + "partial_envelope_refusal": False, + "imap_authentication_failure": include_failure_drills, + "post_data_connection_loss_outcome_unknown": False, + "source_artifact_provenance": False, + "worker_restart_interruption": False, + }, + } + _assert_evidence_safe(evidence, settings=settings) + return evidence + + +def _assert_success_evidence(evidence: Mapping[str, Any]) -> None: + expected_send = { + "attempted_count": 1, + "sent_count": 1, + "failed_count": 0, + "outcome_unknown_count": 0, + "skipped_count": 0, + "delivery_mode": "synchronous", + "statuses": {"smtp_accepted": 1}, + } + if evidence.get("send") != expected_send: + raise AcceptanceError("Campaign success evidence does not show exactly one SMTP acceptance") + expected_append = { + "pending_count": 1, + "processed_count": 1, + "appended_count": 1, + "failed_count": 0, + "skipped_count": 0, + "statuses": {"appended": 1}, + } + if evidence.get("append_sent") != expected_append: + raise AcceptanceError("Campaign success evidence does not show exactly one Sent-folder append") + report = evidence.get("report") if isinstance(evidence.get("report"), dict) else {} + cards = report.get("cards") if isinstance(report.get("cards"), dict) else {} + if ( + cards.get("jobs_total") != 1 + or cards.get("smtp_accepted") != 1 + or cards.get("sent") != 1 + or cards.get("imap_appended") != 1 + or any(cards.get(key) for key in ("failed", "outcome_unknown", "imap_failed", "needs_attention")) + or report.get("send_status_counts") != {"smtp_accepted": 1} + or report.get("imap_status_counts") != {"appended": 1} + ): + raise AcceptanceError("Campaign report does not agree with the successful SMTP/IMAP effects") + provider = evidence.get("provider_verification") if isinstance(evidence.get("provider_verification"), dict) else {} + if ( + int(provider.get("inbox_increment") or 0) < 1 + or int(provider.get("sent_increment") or 0) < 1 + or provider.get("unique_subject_matches_in_inbox") != 1 + or provider.get("unique_subject_matches_in_sent") != 1 + ): + raise AcceptanceError("Provider mailbox evidence does not show exactly one Campaign delivery and append") + boundary = evidence.get("campaign_mail_boundary") + if boundary != { + "profile_reference_only": True, + "smtp_revision_frozen": True, + "imap_revision_frozen": True, + "resolved_transport_material_present": False, + }: + raise AcceptanceError("Campaign success evidence does not prove the Mail profile boundary") + + +def _assert_evidence_safe(evidence: Mapping[str, Any], *, settings: TestbedSettings) -> None: + serialized = json.dumps(evidence, sort_keys=True, default=str) + forbidden_values = { + settings.password, + settings.username, + settings.sender, + settings.recipient, + settings.smtp_host, + settings.imap_host, + } + if any(value and value in serialized for value in forbidden_values): + raise AcceptanceError("Acceptance evidence contains target or credential material") + forbidden_keys = { + "address", + "campaign_id", + "email", + "job_id", + "password", + "profile_id", + "provider_message", + "provider_response", + "recipient", + "recipients", + "version_id", + } + if any( + key.casefold() in forbidden_keys + for key in _iter_mapping_keys(evidence) + ): + raise AcceptanceError("Acceptance evidence contains a forbidden diagnostic field") + + +def _bootstrap_and_run( + *, + settings: TestbedSettings, + fixture_path: Path, + include_failure_drills: bool, +) -> dict[str, Any]: + """Build the isolated application in this command's dedicated process. + + Importing a FastAPI composition configures process-global module and + database runtime. Tests therefore exercise the pure orchestration helpers; + live acceptance invokes this function only through the CLI subprocess. + """ + runtime_root = Path(tempfile.mkdtemp(prefix="govoplan-campaign-greenmail-", dir="/tmp")) + database = None + try: + os.environ.update( + { + "APP_ENV": "test", + "DATABASE_URL": f"sqlite:///{runtime_root / 'acceptance.db'}", + "FILE_STORAGE_BACKEND": "local", + "FILE_STORAGE_LOCAL_ROOT": str(runtime_root / "files"), + "MOCK_MAILBOX_DIR": str(runtime_root / "mock-mailbox"), + "DEV_BOOTSTRAP_ENABLED": "false", + "CELERY_ENABLED": "false", + "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true", + } + ) + from fastapi.testclient import TestClient + + from govoplan_core.db.base import Base + from govoplan_core.db.bootstrap import bootstrap_dev_data + from govoplan_core.db.session import configure_database, set_database + from govoplan_core.settings import Settings, settings as core_settings + from govoplan_core.tenancy.scope import create_scope_tables + + isolated_settings = Settings() + for field_name in Settings.model_fields: + setattr(core_settings, field_name, getattr(isolated_settings, field_name)) + database = configure_database(os.environ["DATABASE_URL"]) + set_database(database) + + from govoplan_core.server.app import app + + create_scope_tables(database.engine) + Base.metadata.create_all(bind=database.engine) + with database.SessionLocal() as session: + bootstrap_dev_data( + session, + api_key_secret="greenmail-acceptance-unused-api-key", + user_password="greenmail-acceptance-admin", + ) + + def snapshot_probe(version_id: str) -> tuple[Mapping[str, Any], Mapping[str, Any]]: + from govoplan_campaign.backend.db.models import CampaignVersion + + with database.SessionLocal() as session: + version = session.get(CampaignVersion, version_id) + if version is None: + raise AcceptanceError("Campaign execution snapshot is unavailable") + raw = version.raw_json if isinstance(version.raw_json, dict) else {} + snapshot = version.execution_snapshot if isinstance(version.execution_snapshot, dict) else {} + return raw, snapshot + + def audit_probe(campaign_id: str, version_id: str) -> Mapping[str, int]: + from govoplan_audit.backend.db.models import AuditLog + + with database.SessionLocal() as session: + actions = [ + row[0] + for row in session.query(AuditLog.action) + .filter(AuditLog.object_id.in_([campaign_id, version_id])) + .all() + if row[0] in EXPECTED_AUDIT_ACTIONS + ] + return dict(Counter(actions)) + + with TestClient(app) as client: + login = _expect( + client.post( + "/api/v1/auth/login", + json={"email": "admin@example.local", "password": "greenmail-acceptance-admin"}, + ), + 200, + "Acceptance login", + ) + access_token = str(login.get("access_token") or "") + if not access_token: + raise AcceptanceError("Acceptance login returned no access token") + from govoplan_core.core.runtime import get_registry + + registry = get_registry() + if registry is None: + raise AcceptanceError("The GovOPlaN module registry is unavailable") + module_versions = {"core": _core_package_version()} + module_versions.update({ + manifest.id: manifest.version + for manifest in registry.manifests() + if manifest.id in {"access", "audit", "campaigns", "mail"} + }) + return run_acceptance( + client, + {"Authorization": f"Bearer {access_token}"}, + settings=settings, + fixture_path=fixture_path, + snapshot_probe=snapshot_probe, + audit_probe=audit_probe, + include_failure_drills=include_failure_drills, + module_versions=module_versions, + ) + finally: + if database is not None: + database.engine.dispose() + shutil.rmtree(runtime_root, ignore_errors=True) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--fixture", type=Path, default=DEFAULT_FIXTURE) + parser.add_argument("--success-only", action="store_true", help="skip the local failure drills") + parser.add_argument("--evidence", type=Path, help="write the bounded JSON evidence to this path") + args = parser.parse_args(argv) + + try: + settings = TestbedSettings.from_environment() + settings.assert_local_testbed() + ensure_sent_folder(settings) + evidence = _bootstrap_and_run( + settings=settings, + fixture_path=args.fixture.resolve(), + include_failure_drills=not args.success_only, + ) + rendered = json.dumps(evidence, ensure_ascii=False, indent=2, sort_keys=True) + "\n" + if args.evidence: + args.evidence.parent.mkdir(parents=True, exist_ok=True) + args.evidence.write_text(rendered, encoding="utf-8") + else: + sys.stdout.write(rendered) + return 0 + except AcceptanceError as exc: + print(f"Campaign acceptance failed: {exc}", file=sys.stderr) + return 1 + except Exception as exc: + print( + f"Campaign acceptance failed unexpectedly ({type(exc).__name__}); inspect local service logs.", + file=sys.stderr, + ) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/greenmail-delivery/campaign.json b/examples/greenmail-delivery/campaign.json new file mode 100644 index 0000000..858f59c --- /dev/null +++ b/examples/greenmail-delivery/campaign.json @@ -0,0 +1,88 @@ +{ + "version": "1.0", + "campaign": { + "id": "greenmail-delivery", + "name": "GreenMail delivery acceptance", + "description": "Credential-free Campaign fixture for the local SMTP/IMAP acceptance test bed.", + "mode": "test" + }, + "fields": [ + { + "name": "display_name", + "type": "string", + "label": "Display name", + "required": true + }, + { + "name": "acceptance_run", + "type": "string", + "label": "Acceptance run", + "required": true + } + ], + "server": { + "mail_profile_id": "00000000-0000-4000-8000-000000000001" + }, + "recipients": { + "from": [ + { + "email": "campaign-test@govoplan.test", + "name": "GovOPlaN acceptance", + "type": "to" + } + ], + "allow_individual_to": true + }, + "template": { + "subject": "[GovOPlaN acceptance ${acceptance_run}] Campaign delivery", + "text": "Hello ${display_name},\n\nThis is an isolated GovOPlaN Campaign SMTP/IMAP acceptance message.\n", + "body_mode": "text" + }, + "attachments": { + "base_path": ".", + "send_without_attachments_behavior": "continue", + "global": [] + }, + "entries": { + "inline": [ + { + "id": "greenmail-recipient", + "to": [ + { + "email": "campaign-test@govoplan.test", + "name": "GreenMail recipient", + "type": "to" + } + ], + "fields": { + "display_name": "GreenMail recipient", + "acceptance_run": "fixture" + } + } + ] + }, + "validation_policy": { + "missing_email": "block", + "template_error": "block" + }, + "delivery": { + "rate_limit": { + "messages_per_minute": 60 + }, + "retry": { + "max_attempts": 3, + "backoff_seconds": [ + 1, + 5, + 30 + ] + }, + "imap_append_sent": { + "enabled": true, + "folder": "Sent" + } + }, + "status_tracking": { + "enabled": true + } +} diff --git a/examples/greenmail-delivery/fixture.json b/examples/greenmail-delivery/fixture.json new file mode 100644 index 0000000..0e6d6ed --- /dev/null +++ b/examples/greenmail-delivery/fixture.json @@ -0,0 +1,22 @@ +{ + "scenario": "greenmail-delivery", + "campaign_file": "campaign.json", + "required_modules": [ + "core", + "access", + "audit", + "campaigns", + "mail" + ], + "required_capabilities": [ + "mail.campaign_delivery" + ], + "transport": "local GreenMail SMTP/IMAP test bed", + "credentials": "local environment only; never copied into Campaign JSON or evidence", + "expected": { + "entries_count": 1, + "built_count": 1, + "smtp_accepted_count": 1, + "imap_appended_count": 1 + } +} diff --git a/tests/test_mail_testbed_acceptance.py b/tests/test_mail_testbed_acceptance.py new file mode 100644 index 0000000..4a20761 --- /dev/null +++ b/tests/test_mail_testbed_acceptance.py @@ -0,0 +1,388 @@ +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path +from typing import Any + +import pytest + +from govoplan_campaign.backend.campaign import load_campaign_config + + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +RUNNER_PATH = REPOSITORY_ROOT / "dev" / "mail-testbed" / "run_campaign_acceptance.py" +FIXTURE_PATH = REPOSITORY_ROOT / "examples" / "greenmail-delivery" / "campaign.json" + + +def _load_runner(): + spec = importlib.util.spec_from_file_location("govoplan_campaign_greenmail_acceptance", RUNNER_PATH) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +runner = _load_runner() + + +class _Response: + def __init__(self, status_code: int, payload: dict[str, Any]) -> None: + self.status_code = status_code + self._payload = payload + + def json(self) -> dict[str, Any]: + return self._payload + + +class _AcceptanceClient: + def __init__(self) -> None: + self.campaign_json: dict[str, Any] | None = None + self.send_calls = 0 + + def post(self, path: str, **kwargs: Any) -> _Response: + if path == "/api/v1/campaigns": + self.campaign_json = kwargs["json"]["config"] + return _Response( + 200, + { + "campaign": {"id": "campaign-internal"}, + "version": {"id": "version-internal"}, + }, + ) + if path.endswith("/validate"): + return _Response(200, {"ok": True, "error_count": 0, "warning_count": 0}) + if path.endswith("/build"): + return _Response( + 200, + { + "built_count": 1, + "build_failed_count": 0, + "queueable_count": 1, + }, + ) + if path.endswith("/send-now"): + self.send_calls += 1 + if self.send_calls == 2: + return _Response(422, {"detail": "Already accepted"}) + return _Response( + 200, + { + "result": { + "attempted_count": 1, + "sent_count": 1, + "failed_count": 0, + "outcome_unknown_count": 0, + "skipped_count": 0, + "delivery_mode": "synchronous", + "results": [{"job_id": "not-retained", "status": "smtp_accepted"}], + } + }, + ) + if path.endswith("/append-sent"): + return _Response( + 200, + { + "result": { + "pending_count": 1, + "processed_count": 1, + "appended_count": 1, + "failed_count": 0, + "skipped_count": 0, + "results": [{"job_id": "not-retained", "status": "appended"}], + } + }, + ) + raise AssertionError(f"unexpected POST {path}") + + def get(self, path: str, **kwargs: Any) -> _Response: + if path.endswith("/report"): + return _Response( + 200, + { + "cards": { + "jobs_total": 1, + "sent": 1, + "smtp_accepted": 1, + "failed": 0, + "outcome_unknown": 0, + "retryable": 0, + "needs_attention": 0, + "imap_appended": 1, + "imap_failed": 0, + }, + "status_counts": { + "send": {"smtp_accepted": 1}, + "imap": {"appended": 1}, + }, + }, + ) + raise AssertionError(f"unexpected GET {path}") + + +def _settings(): + return runner.TestbedSettings( + smtp_host="127.0.0.1", + smtp_port=3025, + imap_host="127.0.0.1", + imap_port=3143, + username="campaign-test@govoplan.test", + password="local-test-password", + sender="campaign-test@govoplan.test", + recipient="campaign-test@govoplan.test", + sent_folder="Sent", + provider_timeout_seconds=5, + ) + + +@pytest.mark.parametrize("host", ["localhost", "mail.test", "192.168.1.20", "8.8.8.8"]) +def test_testbed_rejects_hostnames_and_non_loopback_addresses(host: str) -> None: + settings = _settings() + rejected = runner.TestbedSettings( + smtp_host=host, + smtp_port=settings.smtp_port, + imap_host=settings.imap_host, + imap_port=settings.imap_port, + username=settings.username, + password=settings.password, + sender=settings.sender, + recipient=settings.recipient, + sent_folder=settings.sent_folder, + provider_timeout_seconds=settings.provider_timeout_seconds, + ) + + with pytest.raises(runner.AcceptanceError, match="literal loopback|restricted to the loopback"): + rejected.assert_local_testbed() + + +@pytest.mark.parametrize("host", ["127.0.0.1", "127.8.9.10", "::1"]) +def test_testbed_accepts_literal_loopback_and_preserves_it_in_profile(host: str) -> None: + settings = _settings() + accepted = runner.TestbedSettings( + smtp_host=host, + smtp_port=settings.smtp_port, + imap_host=host, + imap_port=settings.imap_port, + username=settings.username, + password=settings.password, + sender=settings.sender, + recipient=settings.recipient, + sent_folder=settings.sent_folder, + provider_timeout_seconds=settings.provider_timeout_seconds, + ) + + accepted.assert_local_testbed() + profile = runner._profile_payload(accepted, name="Literal loopback") + assert profile["smtp"]["host"] == host + assert profile["imap"]["host"] == host + + +def test_campaign_acceptance_orchestration_retains_only_profile_reference_and_safe_evidence() -> None: + client = _AcceptanceClient() + + def snapshot_probe(_version_id: str): + assert client.campaign_json is not None + return client.campaign_json, { + "mail_profile_id": "profile-1", + "smtp_transport_revision": "opaque-smtp-revision", + "imap_transport_revision": "opaque-imap-revision", + "delivery": {"imap_append_sent": {"enabled": True, "folder": "Sent"}}, + } + + audit = { + "campaign.created": 1, + "campaign.validated": 1, + "campaign.messages_built": 1, + "campaign.sent_now": 1, + "campaign.send_now_rejected": 1, + "campaign.append_sent_enqueued": 1, + } + evidence, subject = runner.execute_campaign_scenario( + client, + {"Authorization": "not-retained"}, + fixture_path=FIXTURE_PATH, + profile_id="profile-1", + settings=_settings(), + scenario="success", + snapshot_probe=snapshot_probe, + audit_probe=lambda _campaign_id, _version_id: audit, + append_sent=True, + repeat_send=True, + ) + evidence["provider_verification"] = { + "inbox_increment": 1, + "sent_increment": 1, + "unique_subject_matches_in_inbox": 1, + "unique_subject_matches_in_sent": 1, + } + + runner._assert_success_evidence(evidence) + runner._assert_evidence_safe( + { + "schema_version": runner.EVIDENCE_SCHEMA, + "coverage": { + "smtp_acceptance": True, + "partial_envelope_refusal": False, + "post_data_connection_loss_outcome_unknown": False, + "source_artifact_provenance": False, + "worker_restart_interruption": False, + }, + "success": evidence, + }, + settings=_settings(), + ) + assert subject.startswith("[GovOPlaN acceptance ") + assert client.send_calls == 2 + assert client.campaign_json is not None + assert client.campaign_json["server"] == {"mail_profile_id": "profile-1"} + assert "credentials" not in json.dumps(client.campaign_json).casefold() + serialized = json.dumps(evidence, sort_keys=True) + assert "not-retained" not in serialized + assert "local-test-password" not in serialized + + +def test_campaign_boundary_rejects_resolved_transport_material() -> None: + with pytest.raises(runner.AcceptanceError, match="forbidden transport material"): + runner.assert_campaign_boundary( + {"server": {"mail_profile_id": "profile-1"}}, + { + "mail_profile_id": "profile-1", + "smtp_transport_revision": "smtp-revision", + "imap_transport_revision": "imap-revision", + "smtp": {"host": "should-not-be-here"}, + }, + profile_id="profile-1", + ) + + +def test_success_projection_fails_closed_on_inconsistent_campaign_report() -> None: + evidence = { + "send": { + "attempted_count": 1, + "sent_count": 1, + "failed_count": 0, + "outcome_unknown_count": 0, + "skipped_count": 0, + "delivery_mode": "synchronous", + "statuses": {"smtp_accepted": 1}, + }, + "append_sent": { + "pending_count": 1, + "processed_count": 1, + "appended_count": 1, + "failed_count": 0, + "skipped_count": 0, + "statuses": {"appended": 1}, + }, + "report": { + "cards": { + "jobs_total": 1, + "sent": 0, + "smtp_accepted": 0, + "failed": 0, + "outcome_unknown": 0, + "needs_attention": 0, + "imap_appended": 0, + "imap_failed": 0, + }, + "send_status_counts": {}, + "imap_status_counts": {}, + }, + "provider_verification": { + "inbox_increment": 1, + "sent_increment": 1, + "unique_subject_matches_in_inbox": 1, + "unique_subject_matches_in_sent": 1, + }, + "campaign_mail_boundary": { + "profile_reference_only": True, + "smtp_revision_frozen": True, + "imap_revision_frozen": True, + "resolved_transport_material_present": False, + }, + } + + with pytest.raises(runner.AcceptanceError, match="report does not agree"): + runner._assert_success_evidence(evidence) + + +def test_evidence_projection_rejects_unknown_status_keys() -> None: + with pytest.raises(runner.AcceptanceError, match="unsupported status"): + runner._send_evidence( + { + "delivery_mode": "synchronous", + "results": [{"status": "provider diagnostic: recipient@example.test"}], + } + ) + + with pytest.raises(runner.AcceptanceError, match="synchronous delivery mode"): + runner._send_evidence( + { + "delivery_mode": "provider diagnostic: recipient@example.test", + "results": [], + } + ) + + with pytest.raises(runner.AcceptanceError, match="unsupported"): + runner._report_evidence( + { + "cards": {}, + "status_counts": { + "send": {"smtp_accepted": 1, "provider-secret": 1}, + "imap": {}, + }, + } + ) + + +def test_fixture_contains_no_transport_credentials() -> None: + raw = json.loads(FIXTURE_PATH.read_text(encoding="utf-8")) + runner._assert_no_forbidden_campaign_keys(raw) + assert raw["server"] == {"mail_profile_id": "00000000-0000-4000-8000-000000000001"} + config = load_campaign_config(FIXTURE_PATH) + assert config.server.mail_profile_id == "00000000-0000-4000-8000-000000000001" + + +def test_fixture_composition_versions_are_complete_and_exact() -> None: + versions = { + "core": "0.1.13", + "access": "0.1.11", + "audit": "0.1.8", + "campaigns": "0.1.10", + "mail": "0.1.10", + "files": "0.1.9", + } + + assert runner.required_composition_versions(FIXTURE_PATH, versions) == { + "core": "0.1.13", + "access": "0.1.11", + "audit": "0.1.8", + "campaigns": "0.1.10", + "mail": "0.1.10", + } + + +def test_fixture_composition_fails_closed_when_a_required_version_is_missing() -> None: + with pytest.raises(runner.AcceptanceError, match="versions are unavailable"): + runner.required_composition_versions( + FIXTURE_PATH, + { + "core": "0.1.13", + "access": "0.1.11", + "campaigns": "0.1.10", + "mail": "0.1.10", + }, + ) + + +def test_testbed_documentation_distinguishes_proven_and_open_failure_drills() -> None: + testbed = (REPOSITORY_ROOT / "dev" / "mail-testbed" / "README.md").read_text(encoding="utf-8") + runbook = (REPOSITORY_ROOT / "docs" / "CAMPAIGN_DELIVERY_RUNBOOK.md").read_text(encoding="utf-8") + + assert "second ordinary send must be rejected before another provider effect" in testbed + assert "post-DATA connection loss" in testbed + assert "real worker restart" in testbed + assert "does not simulate a connection loss after SMTP DATA" in runbook + assert "raw provider diagnostics" in runbook