#!/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 subprocess 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.v2" EXPECTED_AUDIT_ACTIONS = frozenset( { "campaign.created", "campaign.validated", "campaign.messages_built", "campaign.built", "campaign.sent_now", "campaign.send_now_rejected", "campaign.append_sent_enqueued", "campaign.queued", } ) 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", } ) DURABLE_ATTEMPT_STATUSES = frozenset( { "failed_permanent", "failed_temporary", "outcome_unknown", "smtp_accepted", "smtp_accepted_with_refusals", "smtp_in_progress", } ) SMTP_FAULT_MODES = frozenset( { "temporary_data_response", "partial_recipient_refusal", "post_data_disconnect", "post_data_hold", } ) WORKER_TASK_CODE = """ import sys from govoplan_core.celery_app import send_email result = send_email.run(sys.argv[1]) if not isinstance(result, dict) or result.get("status") not in { "outcome_unknown", "smtp_accepted", }: raise SystemExit(2) """ 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, } @dataclass(frozen=True, slots=True) class PreparedCampaignScenario: campaign_id: str version_id: str subject: str validation: dict[str, Any] build: dict[str, Any] boundary: CampaignBoundary def public_evidence(self) -> dict[str, Any]: return { "validation": dict(self.validation), "build": dict(self.build), "campaign_mail_boundary": self.boundary.as_dict(), } 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, additional_envelope_recipient: bool = False, ) -> 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 if additional_envelope_recipient: try: local_part, domain = settings.recipient.rsplit("@", 1) except ValueError as exc: raise AcceptanceError("Campaign acceptance recipient must be an email address") from exc to_addresses.append( { "email": f"{local_part}+partial-{run_token}@{domain}", "name": "Controlled partial-refusal recipient", "type": "to", } ) 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 _durable_state_evidence(payload: Mapping[str, Any]) -> dict[str, Any]: job_count = payload.get("job_count") unfinished_attempt_count = payload.get("unfinished_attempt_count") if ( not isinstance(job_count, int) or isinstance(job_count, bool) or job_count < 0 or not isinstance(unfinished_attempt_count, int) or isinstance(unfinished_attempt_count, bool) or unfinished_attempt_count < 0 ): raise AcceptanceError("Campaign durable delivery state count is invalid") send_statuses = payload.get("send_status_counts") attempt_statuses = payload.get("attempt_status_counts") if not isinstance(send_statuses, dict) or not isinstance(attempt_statuses, dict): raise AcceptanceError("Campaign durable delivery state is invalid") return { "job_count": job_count, "send_status_counts": _allowlisted_count_map( send_statuses, allowed=REPORT_SEND_STATUSES, label="Campaign durable send status", ), "attempt_status_counts": _allowlisted_count_map( attempt_statuses, allowed=DURABLE_ATTEMPT_STATUSES, label="Campaign durable attempt status", ), "unfinished_attempt_count": unfinished_attempt_count, } def prepare_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]]], additional_envelope_recipient: bool = False, ) -> PreparedCampaignScenario: run_token = uuid4().hex[:12] raw, subject = materialize_campaign_fixture( fixture_path, profile_id=profile_id, settings=settings, scenario=scenario, run_token=run_token, additional_envelope_recipient=additional_envelope_recipient, ) 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) return PreparedCampaignScenario( campaign_id=campaign_id, version_id=version_id, subject=subject, 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), }, boundary=boundary, ) 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, delivery_probe: Callable[[str, str], Mapping[str, Any]] | None = None, additional_envelope_recipient: bool = False, ) -> tuple[dict[str, Any], str]: prepared = prepare_campaign_scenario( client, headers, fixture_path=fixture_path, profile_id=profile_id, settings=settings, scenario=scenario, snapshot_probe=snapshot_probe, additional_envelope_recipient=additional_envelope_recipient, ) campaign_id = prepared.campaign_id version_id = prepared.version_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 = { **prepared.public_evidence(), "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 if delivery_probe is not None: evidence["durable_state"] = _durable_state_evidence(delivery_probe(campaign_id, version_id)) return evidence, prepared.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 _SmtpFaultServer(_LoopbackServer): def __init__(self, mode: str) -> None: if mode not in SMTP_FAULT_MODES: raise AcceptanceError("Unsupported SMTP fault mode") self.mode = mode self.data_received = threading.Event() self.release_data_response = threading.Event() self._counter_lock = threading.Lock() self._counters = { "connection_count": 0, "accepted_rcpt_commands": 0, "refused_rcpt_commands": 0, "data_transactions": 0, } super().__init__(("127.0.0.1", 0), _FaultSmtpHandler) def increment(self, key: str) -> int: with self._counter_lock: self._counters[key] += 1 return self._counters[key] def evidence(self) -> dict[str, int]: with self._counter_lock: return dict(self._counters) class _FaultSmtpHandler(socketserver.StreamRequestHandler): def handle(self) -> None: server = self.server if not isinstance(server, _SmtpFaultServer): return server.increment("connection_count") self._write(b"220 local acceptance SMTP\r\n") while line := self.rfile.readline(65_536): command = line.split(None, 1)[0].upper() if command in {b"EHLO", b"HELO"}: self._write(b"250-localhost\r\n250 AUTH PLAIN\r\n") elif command == b"AUTH": self._write(b"235 2.7.0 Authentication successful\r\n") elif command == b"MAIL": self._write(b"250 2.1.0 Sender accepted\r\n") elif command == b"RCPT": rcpt_number = ( server.evidence()["accepted_rcpt_commands"] + server.evidence()["refused_rcpt_commands"] + 1 ) if server.mode == "partial_recipient_refusal" and rcpt_number > 1: server.increment("refused_rcpt_commands") self._write(b"550 5.1.1 Controlled recipient refusal\r\n") else: server.increment("accepted_rcpt_commands") self._write(b"250 2.1.5 Recipient accepted\r\n") elif command == b"DATA": self._write(b"354 End data with .\r\n") if not self._read_data_transaction(): return server.increment("data_transactions") server.data_received.set() if server.mode == "temporary_data_response": self._write(b"451 4.3.0 Controlled temporary rejection\r\n") elif server.mode == "post_data_disconnect": return elif server.mode == "post_data_hold": server.release_data_response.wait() return else: self._write(b"250 2.0.0 Message accepted\r\n") elif command == b"RSET": self._write(b"250 2.0.0 Reset\r\n") elif command == b"NOOP": self._write(b"250 2.0.0 OK\r\n") elif command == b"QUIT": self._write(b"221 2.0.0 Bye\r\n") return else: self._write(b"500 5.5.1 Unsupported acceptance command\r\n") def _read_data_transaction(self) -> bool: while line := self.rfile.readline(65_536): if line in {b".\n", b".\r\n"}: return True return False def _write(self, value: bytes) -> None: self.wfile.write(value) self.wfile.flush() @dataclass(frozen=True, slots=True) class SmtpFaultEndpoint: host: str port: int _server: _SmtpFaultServer def wait_for_data(self, timeout_seconds: float) -> bool: return self._server.data_received.wait(timeout_seconds) def evidence(self) -> dict[str, int]: return self._server.evidence() def release_held_connection(self) -> None: self._server.release_data_response.set() 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 smtp_fault_endpoint(mode: str) -> Iterator[SmtpFaultEndpoint]: server = _SmtpFaultServer(mode) thread = threading.Thread( target=server.serve_forever, name=f"govoplan-smtp-{mode.replace('_', '-')}", daemon=True, ) thread.start() try: host, port = server.server_address yield SmtpFaultEndpoint(str(host), int(port), server) finally: server.release_data_response.set() server.shutdown() server.server_close() thread.join(timeout=2) @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 _start_campaign_worker_task(job_id: str) -> subprocess.Popen[bytes]: """Start the exact Celery send task body in an isolated OS process. The acceptance environment deliberately has no Redis broker. Calling the registered task's ``run`` method exercises the same task/capability path a Celery child process uses, without claiming broker redelivery coverage. """ return subprocess.Popen( [sys.executable, "-c", WORKER_TASK_CODE, job_id], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, close_fds=True, ) def _terminate_worker_process(process: subprocess.Popen[bytes]) -> None: if process.poll() is not None: raise AcceptanceError("Campaign worker task exited before controlled interruption") process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() try: process.wait(timeout=5) except subprocess.TimeoutExpired as exc: raise AcceptanceError("Campaign worker task could not be stopped") from exc def _wait_for_worker_process(process: subprocess.Popen[bytes], *, timeout_seconds: int) -> None: try: return_code = process.wait(timeout=timeout_seconds) except subprocess.TimeoutExpired as exc: process.kill() process.wait(timeout=5) raise AcceptanceError("Restarted Campaign worker task did not complete") from exc if return_code != 0: raise AcceptanceError("Restarted Campaign worker task failed") def execute_worker_interruption_scenario( client: ApiClient, headers: Mapping[str, str], *, fixture_path: Path, profile_id: str, settings: TestbedSettings, endpoint: SmtpFaultEndpoint, snapshot_probe: Callable[[str], tuple[Mapping[str, Any], Mapping[str, Any]]], audit_probe: Callable[[str, str], Mapping[str, int]], delivery_probe: Callable[[str, str], Mapping[str, Any]], worker_job_probe: Callable[[str], str], ) -> dict[str, Any]: prepared = prepare_campaign_scenario( client, headers, fixture_path=fixture_path, profile_id=profile_id, settings=settings, scenario="worker_interruption", snapshot_probe=snapshot_probe, ) queue_payload = _expect( client.post( f"/api/v1/campaigns/{prepared.campaign_id}/queue", headers=dict(headers), json={ "version_id": prepared.version_id, "include_warnings": True, "enqueue_celery": False, "dry_run": False, }, ), 200, "Worker-interruption Campaign queue", ) expected_queue = { "queued_count": 1, "skipped_count": 0, "blocked_count": 0, "enqueued_count": 0, "delivery_mode": "database_queue", "worker_queue_available": False, "dry_run": False, } queue_evidence = {key: queue_payload.get(key) for key in expected_queue} if queue_evidence != expected_queue: raise AcceptanceError("Worker-interruption Campaign was not durably queued once") job_id = worker_job_probe(prepared.version_id) first_worker = _start_campaign_worker_task(job_id) try: if not endpoint.wait_for_data(settings.provider_timeout_seconds): raise AcceptanceError("Campaign worker did not reach the controlled SMTP DATA boundary") _terminate_worker_process(first_worker) finally: if first_worker.poll() is None: first_worker.kill() first_worker.wait(timeout=5) endpoint.release_held_connection() interrupted_state = _durable_state_evidence( delivery_probe(prepared.campaign_id, prepared.version_id) ) if interrupted_state != { "job_count": 1, "send_status_counts": {"sending": 1}, "attempt_status_counts": {"smtp_in_progress": 1}, "unfinished_attempt_count": 1, }: raise AcceptanceError("Interrupted worker state was not durably retained at SMTP in-progress") restarted_worker = _start_campaign_worker_task(job_id) _wait_for_worker_process( restarted_worker, timeout_seconds=settings.provider_timeout_seconds, ) restarted_state = _durable_state_evidence( delivery_probe(prepared.campaign_id, prepared.version_id) ) expected_restarted_state = { "job_count": 1, "send_status_counts": {"outcome_unknown": 1}, "attempt_status_counts": {"outcome_unknown": 1}, "unfinished_attempt_count": 0, } if restarted_state != expected_restarted_state: raise AcceptanceError("Restarted worker did not freeze the unfinished SMTP attempt") protocol_evidence = endpoint.evidence() if protocol_evidence != { "connection_count": 1, "accepted_rcpt_commands": 1, "refused_rcpt_commands": 0, "data_transactions": 1, }: raise AcceptanceError("Restarted worker contacted SMTP or produced an unexpected transaction") report = _expect( client.get( f"/api/v1/campaigns/{prepared.campaign_id}/report", headers=dict(headers), params={"version_id": prepared.version_id}, ), 200, "Worker-interruption Campaign report", ) report_evidence = _report_evidence(report) if report_evidence["send_status_counts"] != {"outcome_unknown": 1}: raise AcceptanceError("Worker-interruption Campaign report did not retain outcome unknown") audit_actions = dict( sorted(audit_probe(prepared.campaign_id, prepared.version_id).items()) ) if not { "campaign.created", "campaign.validated", "campaign.messages_built", "campaign.queued", }.issubset(audit_actions): raise AcceptanceError("Worker-interruption Campaign audit evidence is incomplete") return { **prepared.public_evidence(), "queue": queue_evidence, "interrupted_durable_state": interrupted_state, "restarted_durable_state": restarted_state, "protocol": protocol_evidence, "report": report_evidence, "audit_actions": audit_actions, "process_boundary": { "dedicated_task_process_terminated_after_data": True, "fresh_task_process_completed": True, "duplicate_smtp_transaction_prevented": True, "celery_broker_redelivery_exercised": False, }, } 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]], delivery_probe: Callable[[str, str], Mapping[str, Any]], worker_job_probe: Callable[[str], str], 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, delivery_probe=delivery_probe, ) 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, delivery_probe=delivery_probe, ) 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, delivery_probe=delivery_probe, ) 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, delivery_probe=delivery_probe, ) 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 with smtp_fault_endpoint("temporary_data_response") as endpoint: temporary_response_profile = create_mail_profile( client, headers, settings, name="Explicit temporary SMTP response drill", smtp_host=endpoint.host, smtp_port=endpoint.port, ) temporary_response, _ = execute_campaign_scenario( client, headers, fixture_path=fixture_path, profile_id=temporary_response_profile, settings=settings, scenario="smtp_temporary_response", snapshot_probe=snapshot_probe, audit_probe=audit_probe, append_sent=False, repeat_send=False, delivery_probe=delivery_probe, ) temporary_response["protocol"] = endpoint.evidence() if ( temporary_response["report"]["send_status_counts"] != {"failed_temporary": 1} or temporary_response["durable_state"]["attempt_status_counts"] != {"failed_temporary": 1} or temporary_response["protocol"] != { "connection_count": 1, "accepted_rcpt_commands": 1, "refused_rcpt_commands": 0, "data_transactions": 1, } ): raise AcceptanceError("Explicit SMTP 451 response was not retained as temporary") drills["smtp_temporary_response"] = temporary_response with smtp_fault_endpoint("partial_recipient_refusal") as endpoint: partial_profile = create_mail_profile( client, headers, settings, name="Partial SMTP recipient-refusal drill", smtp_host=endpoint.host, smtp_port=endpoint.port, ) partial, _ = execute_campaign_scenario( client, headers, fixture_path=fixture_path, profile_id=partial_profile, settings=settings, scenario="partial_envelope_refusal", snapshot_probe=snapshot_probe, audit_probe=audit_probe, append_sent=False, repeat_send=False, delivery_probe=delivery_probe, additional_envelope_recipient=True, ) partial["protocol"] = endpoint.evidence() if ( partial["report"]["send_status_counts"] != {"smtp_accepted": 1} or partial["durable_state"]["attempt_status_counts"] != {"smtp_accepted_with_refusals": 1} or partial["protocol"] != { "connection_count": 1, "accepted_rcpt_commands": 1, "refused_rcpt_commands": 1, "data_transactions": 1, } ): raise AcceptanceError("Partial SMTP envelope refusal was not durably distinguished") drills["partial_envelope_refusal"] = partial with smtp_fault_endpoint("post_data_disconnect") as endpoint: ambiguous_profile = create_mail_profile( client, headers, settings, name="Post-DATA SMTP ambiguity drill", smtp_host=endpoint.host, smtp_port=endpoint.port, ) ambiguous, _ = execute_campaign_scenario( client, headers, fixture_path=fixture_path, profile_id=ambiguous_profile, settings=settings, scenario="post_data_connection_loss", snapshot_probe=snapshot_probe, audit_probe=audit_probe, append_sent=False, repeat_send=False, delivery_probe=delivery_probe, ) ambiguous["protocol"] = endpoint.evidence() if ( ambiguous["report"]["send_status_counts"] != {"outcome_unknown": 1} or ambiguous["durable_state"]["attempt_status_counts"] != {"outcome_unknown": 1} or ambiguous["protocol"] != { "connection_count": 1, "accepted_rcpt_commands": 1, "refused_rcpt_commands": 0, "data_transactions": 1, } ): raise AcceptanceError("Post-DATA connection loss was not frozen as outcome unknown") drills["post_data_connection_loss"] = ambiguous with smtp_fault_endpoint("post_data_hold") as endpoint: worker_profile = create_mail_profile( client, headers, settings, name="Campaign worker interruption drill", smtp_host=endpoint.host, smtp_port=endpoint.port, ) worker_interruption = execute_worker_interruption_scenario( client, headers, fixture_path=fixture_path, profile_id=worker_profile, settings=settings, endpoint=endpoint, snapshot_probe=snapshot_probe, audit_probe=audit_probe, delivery_probe=delivery_probe, worker_job_probe=worker_job_probe, ) drills["worker_interruption"] = worker_interruption 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": include_failure_drills, "smtp_permanent_authentication_failure": include_failure_drills, "partial_envelope_refusal": include_failure_drills, "imap_authentication_failure": include_failure_drills, "post_data_connection_loss_outcome_unknown": include_failure_drills, "source_artifact_provenance": False, "worker_restart_interruption": include_failure_drills, "celery_broker_redelivery": 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)) def delivery_probe(campaign_id: str, version_id: str) -> Mapping[str, Any]: from govoplan_campaign.backend.db.models import CampaignJob, SendAttempt with database.SessionLocal() as session: jobs = ( session.query(CampaignJob) .filter( CampaignJob.campaign_id == campaign_id, CampaignJob.campaign_version_id == version_id, ) .all() ) job_ids = [job.id for job in jobs] attempts = ( session.query(SendAttempt) .filter(SendAttempt.job_id.in_(job_ids)) .all() if job_ids else [] ) return { "job_count": len(jobs), "send_status_counts": dict(Counter(job.send_status for job in jobs)), "attempt_status_counts": dict(Counter(attempt.status for attempt in attempts)), "unfinished_attempt_count": sum( 1 for attempt in attempts if attempt.finished_at is None ), } def worker_job_probe(version_id: str) -> str: from govoplan_campaign.backend.db.models import CampaignJob with database.SessionLocal() as session: jobs = ( session.query(CampaignJob) .filter(CampaignJob.campaign_version_id == version_id) .all() ) if len(jobs) != 1: raise AcceptanceError("Worker-interruption Campaign did not contain one job") return jobs[0].id 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, delivery_probe=delivery_probe, worker_job_probe=worker_job_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())