diff --git a/dev/mail-testbed/.env.example b/dev/mail-testbed/.env.example index fa566e4..5ff5b8d 100644 --- a/dev/mail-testbed/.env.example +++ b/dev/mail-testbed/.env.example @@ -9,3 +9,5 @@ GOVOPLAN_MAIL_TEST_IMAP_PORT=3143 GOVOPLAN_MAIL_TEST_SENT_FOLDER=Sent GOVOPLAN_MAIL_TEST_ZIP_PASSWORD=zip-test-password GOVOPLAN_MAIL_TEST_READY_TIMEOUT_SECONDS=45 +GOVOPLAN_CAMPAIGN_TEST_REDIS_PORT=36379 +GOVOPLAN_CAMPAIGN_TEST_REDIS_VISIBILITY_TIMEOUT_SECONDS=3 diff --git a/dev/mail-testbed/docker-compose.yml b/dev/mail-testbed/docker-compose.yml index 9aedd4e..0ad279e 100644 --- a/dev/mail-testbed/docker-compose.yml +++ b/dev/mail-testbed/docker-compose.yml @@ -15,3 +15,19 @@ services: - "${GOVOPLAN_MAIL_TEST_SMTP_PORT:-3025}:3025" - "${GOVOPLAN_MAIL_TEST_IMAP_PORT:-3143}:3143" - "127.0.0.1:38080:8080" + + redis: + image: redis:7-alpine + command: ["redis-server", "--appendonly", "yes"] + ports: + - "127.0.0.1:${GOVOPLAN_CAMPAIGN_TEST_REDIS_PORT:-36379}:6379" + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 1s + timeout: 1s + retries: 30 + volumes: + - campaign-redis-data:/data + +volumes: + campaign-redis-data: diff --git a/dev/mail-testbed/run_celery_redelivery_acceptance.py b/dev/mail-testbed/run_celery_redelivery_acceptance.py new file mode 100644 index 0000000..bc44b03 --- /dev/null +++ b/dev/mail-testbed/run_celery_redelivery_acceptance.py @@ -0,0 +1,852 @@ +#!/usr/bin/env python3 +"""Prove Redis/Celery redelivery does not repeat an ambiguous SMTP effect. + +The default run starts an isolated Redis Compose service, two successive real +Celery worker processes, and a controlled loopback SMTP endpoint. It kills the +first worker after complete DATA but before a final SMTP response. The same +unacknowledged broker task must be delivered to the replacement worker, which +must freeze the unfinished durable attempt as ``outcome_unknown`` without a +second SMTP connection or DATA transaction. +""" + +from __future__ import annotations + +import argparse +from collections import Counter +from contextlib import contextmanager +from dataclasses import dataclass, replace +from datetime import datetime, timezone +import hashlib +import json +import os +from pathlib import Path +import re +import shutil +import socket +import subprocess +import sys +import tempfile +import time +from typing import Any, Callable, Iterator, Mapping +from uuid import uuid4 + +from redis import Redis +from redis.exceptions import RedisError + + +SCRIPT_ROOT = Path(__file__).resolve().parent +REPOSITORY_ROOT = SCRIPT_ROOT.parents[1] +if str(SCRIPT_ROOT) not in sys.path: + sys.path.insert(0, str(SCRIPT_ROOT)) + +from run_campaign_acceptance import ( # noqa: E402 + AcceptanceError, + DEFAULT_FIXTURE, + EXPECTED_AUDIT_ACTIONS, + TestbedSettings, + _assert_evidence_safe, + _core_package_version, + _durable_state_evidence, + _expect, + _report_evidence, + create_mail_profile, + prepare_campaign_scenario, + required_composition_versions, + smtp_fault_endpoint, +) + + +EVIDENCE_SCHEMA = "govoplan.campaign.celery-redelivery-acceptance.v1" +MAX_EVIDENCE_BYTES = 128 * 1024 +DEFAULT_COMPOSE_FILE = SCRIPT_ROOT / "docker-compose.yml" +TASK_RECEIVED_PATTERN = re.compile( + r"Task govoplan[.]campaigns[.]send_email\[([0-9a-f-]{36})\] received", + re.IGNORECASE, +) +TASK_SUCCEEDED_PATTERN = re.compile( + r"Task govoplan[.]campaigns[.]send_email\[([0-9a-f-]{36})\] succeeded", + re.IGNORECASE, +) +WORKER_BOOTSTRAP = r""" +import os +import sys + +from govoplan_core.celery_app import celery + +visibility_timeout = int(os.environ["GOVOPLAN_CAMPAIGN_TEST_REDIS_VISIBILITY_TIMEOUT_SECONDS"]) +celery.conf.broker_transport_options = { + **dict(celery.conf.broker_transport_options or {}), + "polling_interval": 0.25, + "visibility_timeout": visibility_timeout, +} +celery.worker_main( + [ + "worker", + "--loglevel=INFO", + "--pool=solo", + "--concurrency=1", + "--queues=send_email", + f"--hostname={sys.argv[1]}@%h", + "--without-gossip", + "--without-mingle", + "--without-heartbeat", + ] +) +""" + + +@dataclass(slots=True) +class WorkerProcess: + process: subprocess.Popen[bytes] + log_path: Path + log_handle: Any + + def text(self) -> str: + self.log_handle.flush() + try: + return self.log_path.read_text(encoding="utf-8", errors="replace") + except OSError as exc: + raise AcceptanceError("Celery worker evidence log could not be read") from exc + + def received_task_ids(self) -> tuple[str, ...]: + return tuple(TASK_RECEIVED_PATTERN.findall(self.text())) + + def succeeded_task_ids(self) -> tuple[str, ...]: + return tuple(TASK_SUCCEEDED_PATTERN.findall(self.text())) + + +@dataclass(frozen=True, slots=True) +class RedisBrokerState: + queue_depth: int + unacked_hash_count: int + unacked_index_count: int + + def as_dict(self) -> dict[str, int]: + return { + "queue_depth": self.queue_depth, + "unacked_hash_count": self.unacked_hash_count, + "unacked_index_count": self.unacked_index_count, + } + + +def _positive_int(value: str, *, label: str) -> int: + try: + parsed = int(value) + except ValueError as exc: + raise argparse.ArgumentTypeError(f"{label} must be a positive integer") from exc + if parsed <= 0: + raise argparse.ArgumentTypeError(f"{label} must be a positive integer") + return parsed + + +def _unused_loopback_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: + probe.bind(("127.0.0.1", 0)) + return int(probe.getsockname()[1]) + + +def _compose_command( + *, compose_file: Path, project_name: str, operation: str +) -> list[str]: + prefix = [ + "docker", + "compose", + "--file", + str(compose_file), + "--project-name", + project_name, + ] + if operation == "up": + return [*prefix, "up", "--detach", "redis"] + if operation == "down": + return [*prefix, "down", "--volumes", "--remove-orphans"] + raise AcceptanceError("Unsupported Redis Compose operation") + + +def _run_compose( + command: list[str], + *, + environment: Mapping[str, str], + timeout_seconds: int, +) -> None: + try: + completed = subprocess.run( + command, + env=dict(environment), + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=timeout_seconds, + check=False, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise AcceptanceError("Redis Compose lifecycle command failed") from exc + if completed.returncode != 0: + raise AcceptanceError("Redis Compose lifecycle command failed") + + +def _wait_for_redis(redis_url: str, *, timeout_seconds: int) -> None: + deadline = time.monotonic() + timeout_seconds + client = Redis.from_url( + redis_url, + socket_connect_timeout=1, + socket_timeout=1, + decode_responses=False, + ) + try: + while time.monotonic() < deadline: + try: + if client.ping() is True: + return + except RedisError: + pass + time.sleep(0.25) + finally: + client.close() + raise AcceptanceError("Isolated Redis broker did not become ready") + + +@contextmanager +def isolated_redis_broker( + *, + compose_file: Path, + timeout_seconds: int, + requested_port: int | None = None, +) -> Iterator[str]: + if shutil.which("docker") is None: + raise AcceptanceError("Docker CLI is required to start the isolated Redis broker") + if not compose_file.is_file(): + raise AcceptanceError("Redis Compose definition is unavailable") + port = requested_port or _unused_loopback_port() + if port <= 0 or port > 65_535: + raise AcceptanceError("Redis test port is invalid") + project_name = f"govoplan-campaign-redelivery-{uuid4().hex[:12]}" + environment = { + **os.environ, + "GOVOPLAN_CAMPAIGN_TEST_REDIS_PORT": str(port), + } + lifecycle_attempted = False + try: + lifecycle_attempted = True + _run_compose( + _compose_command( + compose_file=compose_file, + project_name=project_name, + operation="up", + ), + environment=environment, + timeout_seconds=timeout_seconds, + ) + redis_url = f"redis://127.0.0.1:{port}/0" + _wait_for_redis(redis_url, timeout_seconds=timeout_seconds) + yield redis_url + finally: + if lifecycle_attempted: + _run_compose( + _compose_command( + compose_file=compose_file, + project_name=project_name, + operation="down", + ), + environment=environment, + timeout_seconds=timeout_seconds, + ) + + +def _start_worker(runtime_root: Path, *, label: str) -> WorkerProcess: + log_path = runtime_root / f"{label}.log" + log_handle = log_path.open("wb") + environment = {**os.environ, "PYTHONUNBUFFERED": "1"} + try: + process = subprocess.Popen( + [sys.executable, "-c", WORKER_BOOTSTRAP, label], + env=environment, + stdin=subprocess.DEVNULL, + stdout=log_handle, + stderr=subprocess.STDOUT, + close_fds=True, + ) + except Exception: + log_handle.close() + raise + return WorkerProcess(process=process, log_path=log_path, log_handle=log_handle) + + +def _wait_for_worker_ready(worker: WorkerProcess, *, timeout_seconds: int) -> None: + deadline = time.monotonic() + timeout_seconds + while time.monotonic() < deadline: + if worker.process.poll() is not None: + raise AcceptanceError("Celery worker exited before becoming ready") + if " ready." in worker.text(): + return + time.sleep(0.2) + raise AcceptanceError("Celery worker did not become ready") + + +def _wait_for_received_task( + worker: WorkerProcess, + *, + timeout_seconds: int, + expected_task_id: str | None = None, +) -> str: + deadline = time.monotonic() + timeout_seconds + while time.monotonic() < deadline: + received = worker.received_task_ids() + if received: + if len(set(received)) != 1: + raise AcceptanceError("Celery worker received more than one task identity") + task_id = received[0] + if expected_task_id is not None and task_id != expected_task_id: + raise AcceptanceError("Replacement worker received a different broker task") + return task_id + if worker.process.poll() is not None: + raise AcceptanceError("Celery worker exited before receiving the task") + time.sleep(0.2) + raise AcceptanceError("Celery worker did not receive the broker task") + + +def _wait_for_task_success( + worker: WorkerProcess, + *, + task_id: str, + timeout_seconds: int, +) -> None: + deadline = time.monotonic() + timeout_seconds + while time.monotonic() < deadline: + succeeded = worker.succeeded_task_ids() + if task_id in succeeded: + return + if worker.process.poll() is not None: + raise AcceptanceError("Replacement Celery worker exited before task success") + time.sleep(0.2) + raise AcceptanceError("Redelivered Celery task did not complete") + + +def _kill_worker(worker: WorkerProcess, *, timeout_seconds: int) -> int: + if worker.process.poll() is not None: + raise AcceptanceError("Celery worker exited before controlled termination") + worker.process.kill() + try: + return_code = worker.process.wait(timeout=timeout_seconds) + except subprocess.TimeoutExpired as exc: + raise AcceptanceError("Celery worker could not be killed") from exc + if return_code == 0: + raise AcceptanceError("Celery worker termination was not forced") + return return_code + + +def _stop_worker(worker: WorkerProcess, *, timeout_seconds: int) -> None: + if worker.process.poll() is None: + worker.process.terminate() + try: + worker.process.wait(timeout=timeout_seconds) + except subprocess.TimeoutExpired: + worker.process.kill() + worker.process.wait(timeout=timeout_seconds) + worker.log_handle.close() + + +def _broker_state(redis_url: str) -> RedisBrokerState: + client = Redis.from_url( + redis_url, + socket_connect_timeout=2, + socket_timeout=2, + decode_responses=False, + ) + try: + return RedisBrokerState( + queue_depth=int(client.llen("send_email")), + unacked_hash_count=int(client.hlen("unacked")), + unacked_index_count=int(client.zcard("unacked_index")), + ) + except RedisError as exc: + raise AcceptanceError("Redis broker state could not be inspected") from exc + finally: + client.close() + + +def _wait_for_broker_drained( + redis_url: str, + *, + timeout_seconds: int, +) -> RedisBrokerState: + deadline = time.monotonic() + timeout_seconds + last = RedisBrokerState(0, 0, 0) + while time.monotonic() < deadline: + last = _broker_state(redis_url) + if last == RedisBrokerState(0, 0, 0): + return last + time.sleep(0.2) + raise AcceptanceError("Redis broker retained delivery state after recovery") + + +def _queue_evidence(payload: Mapping[str, Any]) -> dict[str, Any]: + expected = { + "queued_count": 1, + "skipped_count": 0, + "blocked_count": 0, + "enqueued_count": 1, + "delivery_mode": "worker_queue", + "worker_queue_available": True, + "dry_run": False, + } + evidence = {key: payload.get(key) for key in expected} + if evidence != expected: + raise AcceptanceError("Campaign was not durably queued to one Celery task") + return evidence + + +def execute_redelivery_scenario( + client: Any, + headers: Mapping[str, str], + *, + fixture_path: Path, + settings: TestbedSettings, + endpoint: Any, + redis_url: str, + runtime_root: 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]], +) -> dict[str, Any]: + profile_id = create_mail_profile( + client, + headers, + settings, + name="Campaign Redis Celery redelivery drill", + smtp_host=endpoint.host, + smtp_port=endpoint.port, + ) + prepared = prepare_campaign_scenario( + client, + headers, + fixture_path=fixture_path, + profile_id=profile_id, + settings=settings, + scenario="celery_broker_redelivery", + snapshot_probe=snapshot_probe, + ) + + first_worker = _start_worker(runtime_root, label="first-worker") + replacement_worker: WorkerProcess | None = None + try: + _wait_for_worker_ready( + first_worker, + timeout_seconds=settings.provider_timeout_seconds, + ) + queued = _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": True, + "dry_run": False, + }, + ), + 200, + "Celery-redelivery Campaign queue", + ) + queue_evidence = _queue_evidence(queued) + first_task_id = _wait_for_received_task( + first_worker, + timeout_seconds=settings.provider_timeout_seconds, + ) + if not endpoint.wait_for_data(settings.provider_timeout_seconds): + raise AcceptanceError("Celery worker did not reach complete SMTP DATA") + first_exit_code = _kill_worker( + first_worker, + timeout_seconds=settings.provider_timeout_seconds, + ) + endpoint.release_held_connection() + + interrupted_state = _durable_state_evidence( + delivery_probe(prepared.campaign_id, prepared.version_id) + ) + expected_interrupted = { + "job_count": 1, + "send_status_counts": {"sending": 1}, + "attempt_status_counts": {"smtp_in_progress": 1}, + "unfinished_attempt_count": 1, + } + if interrupted_state != expected_interrupted: + raise AcceptanceError("Killed worker state was not durably SMTP-in-progress") + + replacement_worker = _start_worker(runtime_root, label="replacement-worker") + _wait_for_worker_ready( + replacement_worker, + timeout_seconds=settings.provider_timeout_seconds, + ) + redelivered_task_id = _wait_for_received_task( + replacement_worker, + timeout_seconds=settings.provider_timeout_seconds, + expected_task_id=first_task_id, + ) + _wait_for_task_success( + replacement_worker, + task_id=redelivered_task_id, + timeout_seconds=settings.provider_timeout_seconds, + ) + recovered_state = _durable_state_evidence( + delivery_probe(prepared.campaign_id, prepared.version_id) + ) + expected_recovered = { + "job_count": 1, + "send_status_counts": {"outcome_unknown": 1}, + "attempt_status_counts": {"outcome_unknown": 1}, + "unfinished_attempt_count": 0, + } + if recovered_state != expected_recovered: + raise AcceptanceError("Redelivered task did not freeze the unfinished attempt") + + protocol = endpoint.evidence() + expected_protocol = { + "connection_count": 1, + "accepted_rcpt_commands": 1, + "refused_rcpt_commands": 0, + "data_transactions": 1, + } + if protocol != expected_protocol: + raise AcceptanceError("Broker redelivery caused an unexpected SMTP transaction") + broker_after = _wait_for_broker_drained( + redis_url, + timeout_seconds=settings.provider_timeout_seconds, + ) + first_received = first_worker.received_task_ids() + replacement_received = replacement_worker.received_task_ids() + if first_received != (first_task_id,) or replacement_received != ( + redelivered_task_id, + ): + raise AcceptanceError( + "Celery workers did not each receive the broker task exactly once" + ) + report = _report_evidence( + _expect( + client.get( + f"/api/v1/campaigns/{prepared.campaign_id}/report", + headers=dict(headers), + params={"version_id": prepared.version_id}, + ), + 200, + "Celery-redelivery Campaign report", + ) + ) + if report["send_status_counts"] != {"outcome_unknown": 1}: + raise AcceptanceError("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("Celery-redelivery Campaign audit evidence is incomplete") + + return { + **prepared.public_evidence(), + "queue": queue_evidence, + "interrupted_durable_state": interrupted_state, + "recovered_durable_state": recovered_state, + "protocol": protocol, + "report": report, + "audit_actions": audit_actions, + "broker": { + "transport": "redis", + "same_task_identity_redelivered": first_task_id + == redelivered_task_id, + "first_worker_received_count": len(first_received), + "replacement_worker_received_count": len(replacement_received), + **broker_after.as_dict(), + }, + "supervision": { + "first_worker_killed_after_complete_data": True, + "first_worker_forced_exit": first_exit_code != 0, + "replacement_worker_started": True, + "replacement_worker_completed_redelivery": True, + }, + } + finally: + endpoint.release_held_connection() + _stop_worker(first_worker, timeout_seconds=5) + if replacement_worker is not None: + _stop_worker(replacement_worker, timeout_seconds=5) + + +def _runtime_module_versions(registry: Any) -> dict[str, str]: + versions = {"core": _core_package_version()} + versions.update( + { + manifest.id: manifest.version + for manifest in registry.manifests() + if manifest.id in {"access", "audit", "campaigns", "mail"} + } + ) + return versions + + +def _bootstrap_and_run( + *, + settings: TestbedSettings, + fixture_path: Path, + redis_url: str, + visibility_timeout_seconds: int, +) -> dict[str, Any]: + runtime_root = Path( + tempfile.mkdtemp(prefix="govoplan-campaign-celery-redelivery-", 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": "true", + "REDIS_URL": redis_url, + "GOVOPLAN_CAMPAIGN_TEST_REDIS_VISIBILITY_TIMEOUT_SECONDS": str( + visibility_timeout_seconds + ), + "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="celery-redelivery-unused-api-key", + user_password="celery-redelivery-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 + ), + } + + with TestClient(app) as client: + login = _expect( + client.post( + "/api/v1/auth/login", + json={ + "email": "admin@example.local", + "password": "celery-redelivery-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") + composition_versions = required_composition_versions( + fixture_path, + _runtime_module_versions(registry), + ) + with smtp_fault_endpoint("post_data_hold") as endpoint: + scenario = execute_redelivery_scenario( + client, + {"Authorization": f"Bearer {access_token}"}, + fixture_path=fixture_path, + settings=settings, + endpoint=endpoint, + redis_url=redis_url, + runtime_root=runtime_root, + snapshot_probe=snapshot_probe, + audit_probe=audit_probe, + delivery_probe=delivery_probe, + ) + + 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_redis_celery_controlled_smtp", + "isolated_temporary_database": True, + "redis_started_by_runner": True, + "worker_pool": "solo", + "worker_prefetch_multiplier": 1, + "task_acks_late": True, + "task_reject_on_worker_lost": True, + "visibility_timeout_seconds": visibility_timeout_seconds, + }, + "scenario": scenario, + "coverage": { + "redis_broker_delivery": True, + "celery_worker_processes": True, + "forced_worker_loss_after_complete_data": True, + "same_task_broker_redelivery": True, + "durable_outcome_unknown_recovery": True, + "duplicate_smtp_transaction_prevented": True, + "production_daemon_supervisor": False, + "target_provider": False, + "source_artifact_provenance": False, + }, + } + _assert_evidence_safe(evidence, settings=settings) + rendered = json.dumps( + evidence, + ensure_ascii=False, + indent=2, + sort_keys=True, + ).encode("utf-8") + b"\n" + if len(rendered) > MAX_EVIDENCE_BYTES: + raise AcceptanceError("Celery-redelivery evidence exceeds its size limit") + return evidence + 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("--compose-file", type=Path, default=DEFAULT_COMPOSE_FILE) + parser.add_argument("--redis-port", type=int) + parser.add_argument( + "--visibility-timeout-seconds", + type=lambda value: _positive_int(value, label="visibility timeout"), + default=os.environ.get( + "GOVOPLAN_CAMPAIGN_TEST_REDIS_VISIBILITY_TIMEOUT_SECONDS", + "3", + ), + ) + parser.add_argument( + "--timeout-seconds", + type=lambda value: _positive_int(value, label="timeout"), + default=60, + ) + parser.add_argument("--evidence", type=Path) + args = parser.parse_args(argv) + + try: + settings = TestbedSettings.from_environment() + settings.assert_local_testbed() + settings = replace( + settings, + provider_timeout_seconds=args.timeout_seconds, + ) + with isolated_redis_broker( + compose_file=args.compose_file.resolve(), + timeout_seconds=args.timeout_seconds, + requested_port=args.redis_port, + ) as redis_url: + evidence = _bootstrap_and_run( + settings=settings, + fixture_path=args.fixture.resolve(), + redis_url=redis_url, + visibility_timeout_seconds=args.visibility_timeout_seconds, + ) + 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 Celery-redelivery acceptance failed: {exc}", file=sys.stderr) + return 1 + except Exception as exc: + print( + "Campaign Celery-redelivery acceptance failed unexpectedly " + f"({type(exc).__name__}); inspect local service logs.", + file=sys.stderr, + ) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_celery_redelivery_acceptance.py b/tests/test_celery_redelivery_acceptance.py new file mode 100644 index 0000000..0dc161e --- /dev/null +++ b/tests/test_celery_redelivery_acceptance.py @@ -0,0 +1,286 @@ +from __future__ import annotations + +import importlib.util +import json +from pathlib import Path +from types import SimpleNamespace +import sys +import tempfile +from unittest import mock + +import pytest + + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +RUNNER_PATH = ( + REPOSITORY_ROOT + / "dev" + / "mail-testbed" + / "run_celery_redelivery_acceptance.py" +) +COMPOSE_PATH = REPOSITORY_ROOT / "dev" / "mail-testbed" / "docker-compose.yml" +FIXTURE_PATH = REPOSITORY_ROOT / "examples" / "greenmail-delivery" / "campaign.json" +TASK_ID = "12345678-1234-4234-8234-123456789abc" + + +def _load_runner(): + spec = importlib.util.spec_from_file_location( + "govoplan_campaign_celery_redelivery_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) -> None: + self.status_code = status_code + self._payload = payload + + def json(self) -> dict: + return self._payload + + +class _Client: + def post(self, path: str, **_kwargs) -> _Response: + assert path.endswith("/queue") + return _Response( + 200, + { + "queued_count": 1, + "skipped_count": 0, + "blocked_count": 0, + "enqueued_count": 1, + "delivery_mode": "worker_queue", + "worker_queue_available": True, + "dry_run": False, + }, + ) + + def get(self, path: str, **_kwargs) -> _Response: + assert path.endswith("/report") + return _Response( + 200, + { + "cards": { + "jobs_total": 1, + "outcome_unknown": 1, + "needs_attention": 1, + }, + "status_counts": { + "send": {"outcome_unknown": 1}, + "imap": {"pending": 1}, + }, + }, + ) + + +class _Endpoint: + host = "127.0.0.1" + port = 4025 + + def __init__(self) -> None: + self.release_count = 0 + + def wait_for_data(self, _timeout_seconds: int) -> bool: + return True + + def release_held_connection(self) -> None: + self.release_count += 1 + + def evidence(self) -> dict[str, int]: + return { + "connection_count": 1, + "accepted_rcpt_commands": 1, + "refused_rcpt_commands": 0, + "data_transactions": 1, + } + + +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, + ) + + +def test_compose_redis_is_isolated_durable_and_health_checked() -> None: + compose = COMPOSE_PATH.read_text(encoding="utf-8") + + assert "redis:7-alpine" in compose + assert '"--appendonly", "yes"' in compose + assert "127.0.0.1:${GOVOPLAN_CAMPAIGN_TEST_REDIS_PORT:-36379}:6379" in compose + assert 'test: ["CMD", "redis-cli", "ping"]' in compose + assert "campaign-redis-data:/data" in compose + + +def test_compose_lifecycle_targets_only_isolated_redis_service() -> None: + up = runner._compose_command( + compose_file=COMPOSE_PATH, + project_name="govoplan-campaign-redelivery-test", + operation="up", + ) + down = runner._compose_command( + compose_file=COMPOSE_PATH, + project_name="govoplan-campaign-redelivery-test", + operation="down", + ) + + assert up[-3:] == ["up", "--detach", "redis"] + assert down[-3:] == ["down", "--volumes", "--remove-orphans"] + assert "greenmail" not in up + assert "--project-name" in up + + +def test_worker_bootstrap_uses_real_late_ack_solo_celery_worker() -> None: + source = runner.WORKER_BOOTSTRAP + + assert "celery.worker_main" in source + assert '"--pool=solo"' in source + assert '"--queues=send_email"' in source + assert '"visibility_timeout"' in source + assert '"polling_interval"' in source + assert "send_email.run" not in source + + +def test_worker_log_projection_matches_redelivered_task_without_retaining_id() -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + log_path = Path(temporary_directory) / "worker.log" + log_path.write_text( + "\n".join( + [ + f"Task govoplan.campaigns.send_email[{TASK_ID}] received", + f"Task govoplan.campaigns.send_email[{TASK_ID}] succeeded in 0.1s", + ] + ), + encoding="utf-8", + ) + with log_path.open("ab") as handle: + worker = runner.WorkerProcess( + process=SimpleNamespace(), + log_path=log_path, + log_handle=handle, + ) + + assert worker.received_task_ids() == (TASK_ID,) + assert worker.succeeded_task_ids() == (TASK_ID,) + + +def test_queue_projection_fails_closed_if_no_task_was_published() -> None: + with pytest.raises(runner.AcceptanceError, match="one Celery task"): + runner._queue_evidence( + { + "queued_count": 1, + "skipped_count": 0, + "blocked_count": 0, + "enqueued_count": 0, + "delivery_mode": "database_queue", + "worker_queue_available": False, + "dry_run": False, + } + ) + + +def test_redelivery_orchestration_requires_same_task_and_no_second_smtp_effect( + monkeypatch, +) -> None: + first_worker = mock.Mock() + first_worker.received_task_ids.return_value = (TASK_ID,) + replacement_worker = mock.Mock() + replacement_worker.received_task_ids.return_value = (TASK_ID,) + workers = iter([first_worker, replacement_worker]) + endpoint = _Endpoint() + durable_states = iter( + [ + { + "job_count": 1, + "send_status_counts": {"sending": 1}, + "attempt_status_counts": {"smtp_in_progress": 1}, + "unfinished_attempt_count": 1, + }, + { + "job_count": 1, + "send_status_counts": {"outcome_unknown": 1}, + "attempt_status_counts": {"outcome_unknown": 1}, + "unfinished_attempt_count": 0, + }, + ] + ) + prepared = SimpleNamespace( + campaign_id="campaign-internal", + version_id="version-internal", + public_evidence=lambda: { + "validation": {"ok": True}, + "build": {"built_count": 1}, + "campaign_mail_boundary": { + "profile_reference_only": True, + "smtp_revision_frozen": True, + "imap_revision_frozen": True, + "resolved_transport_material_present": False, + }, + }, + ) + + monkeypatch.setattr(runner, "create_mail_profile", lambda *args, **kwargs: "profile-internal") + monkeypatch.setattr(runner, "prepare_campaign_scenario", lambda *args, **kwargs: prepared) + monkeypatch.setattr(runner, "_start_worker", lambda *args, **kwargs: next(workers)) + monkeypatch.setattr(runner, "_wait_for_worker_ready", lambda *args, **kwargs: None) + received = iter([TASK_ID, TASK_ID]) + monkeypatch.setattr(runner, "_wait_for_received_task", lambda *args, **kwargs: next(received)) + monkeypatch.setattr(runner, "_wait_for_task_success", lambda *args, **kwargs: None) + monkeypatch.setattr(runner, "_kill_worker", lambda *args, **kwargs: -9) + monkeypatch.setattr(runner, "_stop_worker", lambda *args, **kwargs: None) + monkeypatch.setattr( + runner, + "_wait_for_broker_drained", + lambda *args, **kwargs: runner.RedisBrokerState(0, 0, 0), + ) + + evidence = runner.execute_redelivery_scenario( + _Client(), + {"Authorization": "not-retained"}, + fixture_path=FIXTURE_PATH, + settings=_settings(), + endpoint=endpoint, + redis_url="redis://127.0.0.1:36379/0", + runtime_root=Path("/not-used"), + snapshot_probe=lambda _version_id: ({}, {}), + audit_probe=lambda _campaign_id, _version_id: { + "campaign.created": 1, + "campaign.validated": 1, + "campaign.messages_built": 1, + "campaign.queued": 1, + }, + delivery_probe=lambda _campaign_id, _version_id: next(durable_states), + ) + + assert evidence["broker"] == { + "transport": "redis", + "same_task_identity_redelivered": True, + "first_worker_received_count": 1, + "replacement_worker_received_count": 1, + "queue_depth": 0, + "unacked_hash_count": 0, + "unacked_index_count": 0, + } + assert evidence["protocol"]["connection_count"] == 1 + assert evidence["protocol"]["data_transactions"] == 1 + assert evidence["recovered_durable_state"]["send_status_counts"] == { + "outcome_unknown": 1 + } + assert TASK_ID not in json.dumps(evidence, sort_keys=True) + assert endpoint.release_count >= 1