test(campaign): prove Celery broker redelivery
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
852
dev/mail-testbed/run_celery_redelivery_acceptance.py
Normal file
852
dev/mail-testbed/run_celery_redelivery_acceptance.py
Normal file
@@ -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())
|
||||
Reference in New Issue
Block a user